laskoviymishka commented on code in PR #17450:
URL: https://github.com/apache/iceberg/pull/17450#discussion_r3779124686
##########
kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/channel/Channel.java:
##########
@@ -53,20 +54,29 @@ abstract class Channel {
private final Admin admin;
private final Map<Integer, Long> controlTopicOffsets = Maps.newHashMap();
private final String producerId;
+ private final String channelId;
Channel(
String name,
String consumerGroupId,
IcebergSinkConfig config,
KafkaClientFactory clientFactory,
SinkTaskContext context) {
+ this.channelId = config.connectorName() + "-" + config.taskId() + "-" +
name;
this.controlTopic = config.controlTopic();
this.connectGroupId = config.connectGroupId();
this.context = context;
- String transactionalId = config.transactionalPrefix() + name +
config.transactionalSuffix();
+ String transactionalId =
+ "worker".equalsIgnoreCase(name)
+ ? config.transactionalPrefix() + name +
config.transactionalSuffix()
+ : connectGroupId + "-" + config.connectorName() + "-coord";
Review Comment:
Agreed, thanks for fixing.
##########
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:
Fair point, agreed.
##########
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:
Agreed, fair.
##########
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:
You're right, agreed.
##########
kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/channel/Coordinator.java:
##########
@@ -195,6 +193,14 @@ private void commit(boolean partialCommit) {
}
}
+ @VisibleForTesting
+ static boolean isRetryable(RuntimeException exception) {
+ return exception instanceof CommitFailedException
+ || exception instanceof
org.apache.kafka.clients.consumer.CommitFailedException
+ || exception instanceof
org.apache.kafka.common.errors.RebalanceInProgressException
+ || exception instanceof
org.apache.kafka.common.errors.RetriableException;
Review Comment:
Agreed, thanks.
--
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]
