kafka 1.1.0開始增加了目錄間副本遷移功能，本文介紹下操作步驟。

1、展示分區(qū)目錄分布情況

./kafka-log-dirs.sh --bootstrap-server $bootstrap --describe --topic-list "test_1"
{
    "version": 1,
    "brokers": [
        {
            "broker": 0,
            "logDirs": [
                {
                    "logDir": "/tmp/kafka-logs-tmp",
                    "error": null,
                    "partitions": [
                        {
                            "partition": "test_1-1",
                            "size": 0,
                            "offsetLag": 0,
                            "isFuture": false
                        }
                    ]
                },
                {
                    "logDir": "/tmp/kafka-logs-tmp2",
                    "error": null,
                    "partitions": []
                },
                {
                    "logDir": "/tmp/kafka-logs-tmp1",
                    "error": null,
                    "partitions": [
                        {
                            "partition": "test_1-0",
                            "size": 0,
                            "offsetLag": 0,
                            "isFuture": false
                        }
                    ]
                }
            ]
        },
        {
            "broker": 1,
            "logDirs": [
                {
                    "logDir": "/tmp/kafka-logs-tmp",
                    "error": null,
                    "partitions": [
                        {
                            "partition": "test_1-1",
                            "size": 0,
                            "offsetLag": 0,
                            "isFuture": false
                        }
                    ]
                },
                {
                    "logDir": "/tmp/kafka-logs-tmp2",
                    "error": null,
                    "partitions": []
                },
                {
                    "logDir": "/tmp/kafka-logs-tmp1",
                    "error": null,
                    "partitions": [
                        {
                            "partition": "test_1-0",
                            "size": 0,
                            "offsetLag": 0,
                            "isFuture": false
                        }
                    ]
                }
            ]
        }
    ]
}

2、生成遷移計(jì)劃

./kafka-reassign-partitions.sh --zookeeper $zk3 --topics-to-move-json-file topics-to-move.json --broker-list "0,1" --generate
topics-to-move.json:
{
    "topics": [
        {"topic": "test_1"}
    ],
    "version": 1
}
輸出內(nèi)容
{
    "version": 1,
    "partitions": [
        {
            "topic": "test_1",
            "partition": 1,
            "replicas": [
                1,
                0
            ],
            "log_dirs": [
                "any",
                "any"
            ]
        },
        {
            "topic": "test_1",
            "partition": 0,
            "replicas": [
                0,
                1
            ],
            "log_dirs": [
                "any",
                "any"
            ]
        }
    ]
}
注意，此處log_dirs是any，需要手工改期望遷移到的目錄
{
    "version": 1,
    "partitions": [
        {
            "topic": "test_1",
            "partition": 1,
            "replicas": [
                1,
                0
            ],
            "log_dirs": [
                "/tmp/kafka-logs-tmp",
                "/tmp/kafka-logs-tmp"
            ]
        },
        {
            "topic": "test_1",
            "partition": 0,
            "replicas": [
                0,
                1
            ],
            "log_dirs": [
                "/tmp/kafka-logs-tmp",
                "/tmp/kafka-logs-tmp"
            ]
        }
    ]
}

3、執(zhí)行遷移計(jì)劃
 ./kafka-reassign-partitions.sh --zookeeper $zk --execute --reassignment-json-file   topic.json --throttle 50000000 --bootstrap-server $bootstrap

遷移后的分區(qū)分配情況

{
    "version": 1,
    "brokers": [
        {
            "broker": 0,
            "logDirs": [
                {
                    "logDir": "/tmp/kafka-logs-tmp",
                    "error": null,
                    "partitions": [
                        {
                            "partition": "test_1-0",
                            "size": 0,
                            "offsetLag": 0,
                            "isFuture": false
                        },
                        {
                            "partition": "test_1-1",
                            "size": 0,
                            "offsetLag": 0,
                            "isFuture": false
                        }
                    ]
                },
                {
                    "logDir": "/tmp/kafka-logs-tmp2",
                    "error": null,
                    "partitions": []
                },
                {
                    "logDir": "/tmp/kafka-logs-tmp1",
                    "error": null,
                    "partitions": []
                }
            ]
        },
        {
            "broker": 1,
            "logDirs": [
                {
                    "logDir": "/tmp/kafka-logs-tmp",
                    "error": null,
                    "partitions": [
                        {
                            "partition": "test_1-0",
                            "size": 0,
                            "offsetLag": 0,
                            "isFuture": false
                        },
                        {
                            "partition": "test_1-1",
                            "size": 0,
                            "offsetLag": 0,
                            "isFuture": false
                        }
                    ]
                },
                {
                    "logDir": "/tmp/kafka-logs-tmp2",
                    "error": null,
                    "partitions": []
                },
                {
                    "logDir": "/tmp/kafka-logs-tmp1",
                    "error": null,
                    "partitions": []
                }
            ]
        }
    ]
}