kumarpritam863 commented on code in PR #17450:
URL: https://github.com/apache/iceberg/pull/17450#discussion_r3712367794
##########
kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/channel/CommitterImpl.java:
##########
@@ -192,13 +130,32 @@ public void save(Collection<SinkRecord> sinkRecords) {
startWorker();
worker.save(sinkRecords);
}
+ if (reconcileNeeded) {
+ reconcileLeadership();
+ reconcileNeeded = false;
+ }
processControlEvents();
}
+ private void reconcileLeadership() {
+ Set<String> subscribedTopics =
Sets.newTreeSet(sourceConsumer().subscription());
Review Comment:
`With a static list every task sees the same set, so leaderPartition agrees.
With a regex, task A might see [event-aaa, event-bbb] and task B [event-ccc,
...], ` -> I need to check this as as far as I am aware the sunscription holds
the entire list of topics even in the case of regex.
`Subscribe to all topics matching specified pattern to get dynamically
assigned partitions. The pattern matching will be done periodically against all
topics existing at the time of check. This can be controlled through the
metadata.max.age.ms configuration.`
while there can be brief window at the time of create/delete but that is
the case in the current scenario as well. WDYT?
--
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]
