kumarpritam863 commented on code in PR #17450:
URL: https://github.com/apache/iceberg/pull/17450#discussion_r3712275354
##########
kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/channel/CommitterImpl.java:
##########
@@ -214,11 +171,21 @@ private void startWorker() {
}
}
- private void startCoordinator() {
+ private void startCoordinator(Set<String> subscribedTopics) {
if (null == this.coordinatorThread) {
- LOG.info("Task {} elected leader, starting commit coordinator", taskId);
+ int topicPartitionCount = 0;
+ for (String topic : subscribedTopics) {
+ List<PartitionInfo> partitions = sourceConsumer().partitionsFor(topic);
+ if (partitions != null) {
Review Comment:
`broker still loading, topic` -> isn't this thew case in the current
scenario as well. Also connect gurantees that a save will only be called after
the rebalance has completed. So I am expecting that it should not return null.``
`The old code summed partitions from the group members, which was
authoritative. I'd either restore an authoritative count or treat count == 0`
-> even in this case that topic will not be assigned to anyone so I think it
will be the same.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
