### Kafka MirrorMaker 源集群扩展后新分区无任务传输解决方案 当源集群中的主题（topic）进行了分区扩展之后，如果发现新的分区内数据未能通过MirrorMaker同步到目标集群，则可能是因为MirrorMaker并未感知到这些变化。为了确保新增加的分区能够被正确复制，可以采取以下措施： #### 更新消费者组偏移量重置策略 对于已经存在的消费者实例来说，默认情况下它们只会关注之前订阅过的那些分区；因此，当有新的分区加入时，除非特别指定，否则不会自动开始消费来自新分区的数据流。可以通过设置`--reset-offsets`参数来强制让现有的镜像进程重新读取最新的元数据并调整其内部状态以适应更改后的布局[^1]。 ```bash kafka-consumer-groups.sh --bootstrap-server <broker-list> \ --group <mirror-maker-group-id> \ --reset-offsets --to-latest --execute \ --all-topics ``` 此命令将会把所有正在使用的消费者组内的成员都指向各自跟踪的主题里最末端的位置，从而使得任何后续启动的服务都能够立即处理最新写入的信息片段——当然也包括最近才创建出来的那部分区域。 #### 修改配置文件使能动态发现特性 为了让Kafka Connect框架下的MirrorMaker组件能够在不停止服务的前提下识别出结构变动，可以在对应的worker.properties或者connect-distributed.properties中开启特定选项允许它周期性地查询上游Broker获取更新的通知。具体做法是在上述提到的属性列表里面添加如下几项声明[^2]: ```properties offset.storage.topic=<your_offset_topic> config.storage.topic=<your_config_topic> status.storage.topic=<your_status_topic> internal.key.converter=org.apache.kafka.connect.json.JsonConverter internal.value.converter=org.apache.kafka.connect.json.JsonConverter key.converter.schemas.enable=false value.converter.schemas.enable=false plugin.path=/path/to/your/plugins # Enable dynamic topic creation and partition reassignment support consumer.override.auto.offset.reset=earliest producer.override.acks=all rest.host.name=localhost rest.port=8083 offset.flush.interval.ms=10000 offset.flush.timeout.ms=5000 retry.backoff.ms=100 request.timeout.ms=30000 connections.max.idle.ms=540000 metadata.max.age.ms=5000 refresh.topologies.enabled=true topology.refresh.time.ms=60000 ``` 其中最关键的部分在于最后四行定义了如何频繁刷新拓扑视图(`refresh.topologies.enabled`)以及具体的间隔时间(`topology.refresh.time.ms`)，这样即使在生产环境中发生了一些结构性的变化也能及时得到响应而无需重启整个应用程序。 #### 手动触发再平衡事件 有时尽管做了以上努力仍然无法解决问题的话，那么还可以尝试主动发起一次完整的再均衡过程，即促使所有的参与者暂时中断当前的工作流程以便于重新协商各自的职责范围。这一步骤通常涉及到向负责协调工作的Zookeeper节点发送特殊指令或者是直接利用API接口请求重建连接关系网状结构[^3]。 ```java // Java API example to trigger a rebalance by revoking all assignments. Map<TopicPartition, OffsetAndMetadata> currentOffsets = new HashMap<>(); for (TopicPartition tp : consumer.assignment()) { long offset = consumer.position(tp); currentOffsets.put(tp, new OffsetAndMetadata(offset)); } consumer.commitSync(currentOffsets); // Commit offsets before leaving group consumer.unsubscribe(); // Unsubscribe from topics/topics patterns consumer.close(Duration.ofMillis(0)); // Close the consumer cleanly Thread.sleep(5000); // Wait for some time to ensure complete closure newConsumer.subscribe(topicsList); // Resubscribe with same or different set of topics ``` 这段代码展示了怎样优雅地关闭旧版本客户端的同时又迅速切换至另一套全新的实例上继续工作，期间会经历一轮标准意义上的资源分配调整机制，进而达到激活未映射过的新分区的目的。