之前写过一篇文章，介绍我使用kafka一段时间后的总结，Kafka应用总结。
里面有提到过回收服务器时遇到了问题，导致最后终止了回收操作。今天就来讲一讲这个问题。
这就是kafka中的reassign问题。这个问题不止出现在回收服务器的时候，其实也出现在增加新服务器的时候。
在kafka中，一个Topic会包含1个及以上的Partition，每个Partition包含该Topic的部分数据；所有的Partition的数据合并后得到完整的Topic数据。
而每个Partition会对应着一组Replication，其中包括1个主Replication和若干个（>=0）副Replication。每个Replication保存着完整的Replication数据；副Replication会和主Replication保持数据同步。与主Replication保持同步的Replication，被称为ISR(In-Sync Replicas)。
当我们使用命令：
bin/kafka-topics.sh --bootstrap-server $kafka_host:$port --describe --topic __consumer_offsets
查看某个Topic的详细信息时，我们可以得到如下所示的结果（只列出前十个Partition）：
Topic: __consumer_offsets PartitionCount: 50 ReplicationFactor: 3 Configs: compression.type=producer,cleanup.policy=compact,segment.bytes=104857600,retention.ms=43200000,max.message.bytes=10000120,unclean.leader.election.enable=false,retention.bytes=-1
Topic: __consumer_offsets Partition: 0 Leader: 1 Replicas: 1,2,3 Isr: 1,2,3
Topic: __consumer_offsets Partition: 1 Leader: 1 Replicas: 1,2,3 Isr: 1,2,3
Topic: __consumer_offsets Partition: 2 Leader: 2 Replicas: 1,2,3 Isr: 2,3,1
Topic: __consumer_offsets Partition: 3 Leader: 1 Replicas: 1,2,3 Isr: 1,3,2
Topic: __consumer_offsets Partition: 4 Leader: 2 Replicas: 1,2,3 Isr: 1,2,3
Topic: __consumer_offsets Partition: 5 Leader: 3 Replicas: 1,2,3 Isr: 1,2,3
Topic: __consumer_offsets Partition: 6 Leader: 1 Replicas: 1,2,3 Isr: 1,2,3
Topic: __consumer_offsets Partition: 7 Leader: 1 Replicas: 1,2,3 Isr: 1,2,3
Topic: __consumer_offsets Partition: 8 Leader: 2 Replicas: 1,2,3 Isr: 2,3,1
Topic: __consumer_offsets Partition: 9 Leader: 3 Replicas: 1,2,3 Isr: 1,2,3
我们以第一行数据为例，
Topic: __consumer_offsets，表示Topic的名称，此处为__consumer_offsets。
Partition: 0 ，表示Partition的编号，此处表示编号为0的Partition。
Leader: 1，表示此Partition的Leader配置对应的broker id。此处表示该Partition的Leader配置位于broker id=1的Broker上。
Replicas: 1,2,3，表示该Partition所有的配置Replication对应的broker id集合。此处表示该Partition的所有副本分别位于broker id=1, broker id=2和broker id=3的三个Broker上。
Isr: 1,2,3，表示目前与主副本保持数据同步的broker id集合。此处表示三个副本都与主副本保持数据同步。
了解了这些知识后，我们来看一下在扩缩容的时候，kafka是怎么做的？
扩容时
当我们往集群里面加入新的服务器时，我们希望kafka能够把以前服务器上的一些负载转移到新的服务器上，这样才达到了动态扩容的效果。那么kafka是怎么做的呢？
答案是，kafka什么都没做。
所以，当新加入服务器到集群中时，对现有的Topic而言，是没有任何效果的。但是对于新创建的Topic而言，kafka会评估各个broker的性能，可能会把新的Topic对应的Replication创建到新的服务器上。
缩容时
当服务器有空闲时，我们希望能够直接把多余的服务器从集群中删除，并且希望kafka集群可以自动转移副本数据到一台可用的服务器上去。实际情况如何呢？
答案是，kafka不会调整Replicas。从而导致如果把Isr里面的服务器都回收掉了，该Partition将不可用。
当扩容和缩容时，kafka都做不到自动转移负载，这也太令人失望了。但是现实就是如此。那么当我们需要进行扩缩容时，该怎么做呢？
这里就需要用到kafka在bin目录下提供的一个脚本kafka-reassign-partitions.sh。该脚本接受一个参数--reassignment-json-file。通过给出一个手动分配Topic中所有的Partition的Replication所在的broker id集合的文件，来达到重新分配的效果。该文件如下所示：
{
"version": 1,
"partitions": [
{"topic": "gameLog", "partition": 0, "replicas":[1,2,3]},
{"topic": "gameLog", "partition": 1, "replicas":[1,2,3]},
{"topic": "gameLog", "partition": 2, "replicas":[1,2,3]},
{"topic": "gameLog", "partition": 3, "replicas":[1,2,3]},
{"topic": "gameLog", "partition": 4, "replicas":[1,2,3]},
{"topic": "gameLog", "partition": 5, "replicas":[1,2,3]},
{"topic": "gameLog", "partition": 6, "replicas":[1,2,3]},
{"topic": "gameLog", "partition": 7, "replicas":[1,2,3]}
]
}
将该文件保存为gameLog.json。然后执行命令：
bin/kafka-reassign-partitions.sh --reassignment-json-file offset.json --zookeeper $zookeeper:$zkPort --execute
kafka集群将会按照gameLog.json中给出的配置将所有的Partition和对应的Replication转移到指定的Broker上。从而达到了重新分配服务器资源的目的。
总结
在对kafka集群进行扩缩容时，kafka集群不像Elastic Search或TiDB等分布式系统一样，提供资源自动重新分配的功能。而是通过提供kafka-reassign-partitions.sh这个脚本，把资源的分配任务交给使用者，来达到目的。
当然，如果集群不够智能，在自动处理时，总是表现不如人工，那么kafka的设计是合理的。因为扩缩容并不经常发生；但是如果系统能够智能地进行资源的调度，那么由系统进行自动处理是合理的。
网友评论