kumarpritam863 commented on code in PR #17450:
URL: https://github.com/apache/iceberg/pull/17450#discussion_r3712299288
##########
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());
+ TopicPartition leader = leaderPartition(subscribedTopics);
+ if (leader != null && context.assignment().contains(leader)) {
+ startCoordinator(subscribedTopics);
+ } else {
+ stopCoordinator();
+ }
+ }
+
private void processControlEvents() {
if (coordinatorThread != null && coordinatorThread.isTerminated()) {
- throw new NotRunningException(
- String.format("Coordinator unexpectedly terminated on committer %s",
taskId));
+ if (isProducerFenced(coordinatorThread.exception())) {
+ LOG.warn("Committer {} coordinator was fenced by a newer coordinator;
clearing it", taskId);
+ stopCoordinator();
Review Comment:
Yeah we can do that but this is fenced means there was another coordinator
elected somewhere, means it got the leaderPartition so in case we
stopCoordinator on this task and that task then looses that leaderPartition, it
will be ultimately assigned to some task which will set reconcile needed in the
open/close and evaluate on next save.
--
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]
