hadoop►kafka►kafka migrate and scale

kafka migrate and scale

迁移和扩容

1
2
3
迁移和扩容需要满足两点:
1.将指定topic迁移到集群内新增的node上
2.将topic指定的partition迁移到新增的node上

  1. 测试数据

    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15
    16
    17
    18
    19
    20
    penn@ubuntu:~$ /mnt/app/kafka.1/bin/kafka-topics.sh --zookeeper 10.0.2.15:2181/kafka --create --replication-factor 2 --partitions 2 --topic test-1
    Created topic "test-1".
    penn@ubuntu:~$ /mnt/app/kafka.1/bin/kafka-topics.sh --zookeeper 10.0.2.15:2181/kafka --create --replication-factor 2 --partitions 2 --topic test-2
    Created topic "test-2".

    penn@ubuntu:~$ /mnt/app/kafka.1/bin/kafka-topics.sh --zookeeper 10.0.2.15:2181/kafka --list
    test-1
    test-2

    penn@ubuntu:~$ /mnt/app/kafka.1/bin/kafka-topics.sh --zookeeper 10.0.2.15:2181/kafka --describe --topic test-1
    Topic:test-1    PartitionCount:2        ReplicationFactor:2     Configs:
            Topic: test-1   Partition: 0    Leader: 2       Replicas: 2,1   Isr: 2,1
            Topic: test-1   Partition: 1    Leader: 1       Replicas: 1,2   Isr: 1,2

    penn@ubuntu:~$ /mnt/app/kafka.1/bin/kafka-topics.sh --zookeeper 10.0.2.15:2181/kafka --describe --topic test-2
    Topic:test-2    PartitionCount:2        ReplicationFactor:2     Configs:
            Topic: test-2   Partition: 0    Leader: 2       Replicas: 2,1   Isr: 2,1
            Topic: test-2   Partition: 1    Leader: 1       Replicas: 1,2   Isr: 1,2

    通过上面我们看到,我们创建了2个topic,它们存放在1和2两个broker上,此时我们添加第三个broker,并将test-1 topic迁移到3和4broker上

  2. 准备迁移json

    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    //创建migration-test-topic.json文件,并写入要迁移都topic
    penn@ubuntu:~$ cat migration-test-topic.json
    {
      "topics": [
        {
          "topic": "test-1"
        }
      ],
      "version": 1
    }

  3. 根据json文件生成迁移JSON语句

    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    //根据上面的migration-test-topic.json文件生成迁移JSON语句
    penn@ubuntu:~$ /mnt/app/kafka.1/bin/kafka-reassign-partitions.sh --zookeeper 10.0.2.15:2181/kafka --topics-to-move-json-file ./migration-test-topic.json --broker-list "3,4" --generate
    Current partition replica assignment
    {"version":1,"partitions":[{"topic":"test-1","partition":1,"replicas":[1,2]},{"topic":"test-1","partition":0,"replicas":[2,1]}]}
    Proposed partition reassignment configuration
    {"version":1,"partitions":[{"topic":"test-1","partition":1,"replicas":[3,4]},{"topic":"test-1","partition":0,"replicas":[4,3]}]}


    //将JSON保存到文件
    penn@ubuntu:~$ cat move-to-new-broker.json //将下面都json写入到新文件中
    {"version":1,"partitions":[{"topic":"test-1","partition":1,"replicas":[3,4]},{"topic":"test-1","partition":0,"replicas":[4,3]}]}
    1
    2
    3
    4
    5
    //补充:一个错误
    penn@ubuntu:~$ /mnt/app/kafka.1/bin/kafka-reassign-partitions.sh --zookeeper 10.0.2.15:2181/kafka --topics-to-move-json-file ./migration-test-topic.json --broker-list "3" --generate
    Partitions reassignment failed due to replication factor: 2 larger than available brokers: 1
    org.apache.kafka.common.errors.InvalidReplicationFactorException: replication factor: 2 larger than available brokers: 1
    报错原因: 是因为要迁移都topic有两个replica,所以在添加新的broker的个数最好是replica指定数的倍数

  4. 对topic进行迁移

    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15
    16
    17
    18
    19
    20
    21
    22
    23
    24
    25
    //迁移前检查
    penn@ubuntu:~$ /mnt/app/kafka.1/bin/kafka-reassign-partitions.sh --zookeeper 10.0.2.15:2181/kafka --reassignment-json-file ./move-to-new-broker.json --verify
    Status of partition reassignment:
    ERROR: Assigned replicas (1,2) don't match the list of replicas for reassignment (3,4) for partition [test-1,1]
    ERROR: Assigned replicas (2,1) don't match the list of replicas for reassignment (4,3) for partition [test-1,0]
    Reassignment of partition [test-1,1] failed
    Reassignment of partition [test-1,0] failed

    //迁移
    penn@ubuntu:~$ /mnt/app/kafka.1/bin/kafka-reassign-partitions.sh --zookeeper 10.0.2.15:2181/kafka --reassignment-json-file ./move-to-new-broker.json --execute
    Current partition replica assignment
    {"version":1,"partitions":[{"topic":"test-1","partition":1,"replicas":[1,2]},{"topic":"test-1","partition":0,"replicas":[2,1]}]}
    Save this to use as the --reassignment-json-file option during rollback
    Successfully started reassignment of partitions.

    //迁移后检查
    penn@ubuntu:~$ /mnt/app/kafka.1/bin/kafka-reassign-partitions.sh --zookeeper 10.0.2.15:2181/kafka --reassignment-json-file ./move-to-new-broker.json --verify
    Status of partition reassignment:
    Reassignment of partition [test-1,1] completed successfully
    Reassignment of partition [test-1,0] completed successfully

    penn@ubuntu:~$ /mnt/app/kafka.1/bin/kafka-topics.sh --zookeeper 10.0.2.15:2181/kafka --describe --topic test-1
    Topic:test-1    PartitionCount:2        ReplicationFactor:2     Configs:
            Topic: test-1   Partition: 0    Leader: 4       Replicas: 4,3   Isr: 3,4
            Topic: test-1   Partition: 1    Leader: 3       Replicas: 3,4   Isr: 3,4