twthorn commented on code in PR #18039:
URL: https://github.com/apache/iceberg/pull/18039#discussion_r3994080722
##########
kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/channel/Channel.java:
##########
@@ -147,10 +146,11 @@ protected Map<Integer, Long> controlTopicOffsets() {
}
/**
- * Commit consumer offsets. Only commits offsets if it has not committed
offsets before or the
- * value is greater than the cached offset.
- *
- * <p>Note: there is a risk that two parallel coordinators may overwrite
each other's offsets.
+ * Commits consumer offsets through the coordinator's transactional
producer, committing a
Review Comment:
Agree with that, updated
##########
kafka-connect/kafka-connect-runtime/src/integration/java/org/apache/iceberg/connect/TestCoordinatorFencing.java:
##########
@@ -0,0 +1,159 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg.connect;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+import java.time.Duration;
+import java.util.Map;
+import java.util.UUID;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
+import org.apache.kafka.clients.admin.Admin;
+import org.apache.kafka.clients.admin.NewTopic;
+import org.apache.kafka.clients.consumer.ConsumerGroupMetadata;
+import org.apache.kafka.clients.consumer.OffsetAndMetadata;
+import org.apache.kafka.clients.producer.KafkaProducer;
+import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.errors.ProducerFencedException;
+import org.awaitility.Awaitility;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+/**
+ * Verifies that a stale coordinator cannot overwrite offsets committed by a
newer coordinator, even
+ * when the stale coordinator commits an offset ahead of its own previous one
but still behind the
+ * newer coordinator's.
+ */
+public class TestCoordinatorFencing {
+
+ private static final long STALE_INITIAL_OFFSET = 100L;
+ private static final long NEW_COORDINATOR_OFFSET = 200L;
+ private static final long NEXT_STALE_OFFSET = 150L;
+
+ private final TestContext context = TestContext.instance();
+
+ private String topicName;
+ private String groupId;
+ private TopicPartition topicPartition;
+ private Admin admin;
+
+ @BeforeEach
+ public void before() {
+ topicName = "coord-fencing-topic-" + UUID.randomUUID();
+ groupId = "coord-fencing-group-" + UUID.randomUUID();
+ topicPartition = new TopicPartition(topicName, 0);
+ admin = context.initLocalAdmin();
+ createTopic(topicName);
+ }
+
+ @AfterEach
+ public void after() {
+ deleteTopic(topicName);
+ admin.close();
+ }
+
+ @Test
+ public void newCoordinatorFencesStaleCoordinatorOffsetCommits() throws
Exception {
+ Map<String, String> connectorProps = connectorProps();
+ String staleCoordinatorId = new
IcebergSinkConfig(connectorProps).coordinatorTransactionalId();
+ String newCoordinatorId = new
IcebergSinkConfig(connectorProps).coordinatorTransactionalId();
Review Comment:
Good point, added the assertion that the ID's are equal
For a single variable, I think this weakens the test. We would rather run
the two calls separately as they would be done, and assert that the value is
the same. Before, this function was non-deterministic (so two calls with same
configs would generate different IDs). By making two calls we verify the same
ID is returned.
##########
kafka-connect/kafka-connect/src/test/java/org/apache/iceberg/connect/channel/TestChannel.java:
##########
@@ -70,8 +70,14 @@ public void
committedControlTopicOffsetsDoNotRegressOnReplay() {
// the offset committed for the group is what a restarted channel resumes
from, and what the
// coordinator stamps on the snapshot, so a regression here is durable
- assertThat(consumer.committed(ImmutableSet.of(CTL_TOPIC_PARTITION)))
- .containsEntry(CTL_TOPIC_PARTITION, new OffsetAndMetadata(5L));
+ assertThat(lastGroupOffsets()).containsEntry(CTL_TOPIC_PARTITION, new
OffsetAndMetadata(5L));
+ }
+
+ private Map<TopicPartition, OffsetAndMetadata> lastGroupOffsets() {
+ List<Map<String, Map<TopicPartition, OffsetAndMetadata>>> history =
+ producer.consumerGroupOffsetsHistory();
+ assertThat(history).isNotEmpty();
+ return history.get(history.size() - 1).values().iterator().next();
Review Comment:
Agreed, fixed and refactored to helper to DRY up code
##########
kafka-connect/kafka-connect/src/test/java/org/apache/iceberg/connect/channel/TestCommitterImpl.java:
##########
@@ -165,4 +173,54 @@ public void
testStartFailurePropagatesAsNotRunningException()
.isInstanceOf(NotRunningException.class)
.hasMessageContaining("Coordinator unexpectedly terminated");
}
+
+ @ParameterizedTest
+ @ValueSource(strings = {"ProducerFenced", "InvalidProducerEpoch"})
+ public void testFencedCoordinatorIsClearedWithoutFailingTask(String
exceptionType)
+ throws NoSuchFieldException, IllegalAccessException {
+ RuntimeException fenceException =
+ "ProducerFenced".equals(exceptionType)
+ ? new ProducerFencedException("fenced by a newer coordinator")
+ : new InvalidProducerEpochException("producer epoch bumped by a
newer coordinator");
+
+ Coordinator coordinator = mock(Coordinator.class);
+ doThrow(fenceException).when(coordinator).process();
Review Comment:
Agreed, removed and added the other test.
##########
kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/channel/Channel.java:
##########
@@ -177,7 +177,20 @@ protected void commitConsumerOffsets() {
if (!offsetsToCommit.isEmpty()) {
LOG.debug("Committing consumer offsets: {}", offsetsToCommit);
- consumer.commitSync(offsetsToCommit);
+ synchronized (producer) {
+ producer.beginTransaction();
+ try {
+ producer.sendOffsetsToTransaction(offsetsToCommit,
consumer.groupMetadata());
+ producer.commitTransaction();
Review Comment:
Good point, updated
##########
kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/IcebergSinkConfig.java:
##########
@@ -443,6 +443,16 @@ public String transactionalPrefix() {
return "";
}
+ /**
+ * The transactional ID for the coordinator's producer: scoped to the
coordinator role, stable
+ * across coordinator instances and task restarts within a connector, and
unique across
+ * connectors. Its stability lets an incoming coordinator's {@code
initTransactions()} bump the
+ * producer epoch and fence a stale coordinator.
+ */
+ public String coordinatorTransactionalId() {
+ return transactionalPrefix() + connectGroupId() + "-coordinator";
Review Comment:
Updated javadoc to mention we intentionally do not use the transactional
suffix.
For upgrading, agree on that. As long as the upgrade is completed in a
timely manner it should be a no-op whether via down-then-up or rolling deploy.
For the upgrade notes I added a section to this PR description. I see there are
various doc .md files but the only upgrade notes is in releases.md which is
added by the release manager. If there's some way we can ensure this gets added
there then that seems the best place.
##########
kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/channel/Worker.java:
##########
@@ -49,8 +49,8 @@ class Worker extends Channel {
SinkTaskContext context) {
// pass transient consumer group ID to which we never commit offsets
super(
- "worker",
config.controlGroupIdPrefix() + UUID.randomUUID(),
+ config.transactionalPrefix() + "worker" + config.transactionalSuffix(),
Review Comment:
Updated so we use getOrDefault. Note suffix will always be
[set](https://github.com/apache/iceberg/blob/main/kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/IcebergSinkConnector.java#L57)
so that case should never occur, but we can still handle that null case.
##########
kafka-connect/kafka-connect/src/test/java/org/apache/iceberg/connect/channel/TestCoordinator.java:
##########
@@ -433,21 +408,30 @@ public void
testCommitConsumerMixedPartitionsRewindOrAdvance() {
coordinator.controlTopicOffsets().put(1, watermarkToSkip);
coordinator.commitConsumerOffsets();
- Map<TopicPartition, OffsetAndMetadata> committedOffsetAndMetadata =
- consumer.committed(ImmutableSet.of(ctl0, ctl1));
-
- OffsetAndMetadata committed0 = committedOffsetAndMetadata.get(ctl0);
- OffsetAndMetadata committed1 = committedOffsetAndMetadata.get(ctl1);
-
- assertThat(committed0 == null ? 0L : committed0.offset())
+ assertThat(lastCommittedOffset(0))
.as("commitConsumerOffsets should advance the consumer group offsets")
.isEqualTo(watermarkToCommit);
- assertThat(committed1 == null ? 0L : committed1.offset())
+ assertThat(lastCommittedOffset(1))
.as("commitConsumerOffsets should not rewind consumer group offsets")
.isEqualTo(healthWatermark1);
}
+ private Long lastCommittedOffset(int partition) {
+ TopicPartition topicPartition = new TopicPartition(CTL_TOPIC_NAME,
partition);
+ Long result = null;
+ for (Map<String, Map<TopicPartition, OffsetAndMetadata>> committed :
+ producer.consumerGroupOffsetsHistory()) {
+ for (Map<TopicPartition, OffsetAndMetadata> offsets :
committed.values()) {
+ OffsetAndMetadata metadata = offsets.get(topicPartition);
+ if (metadata != null) {
+ result = metadata.offset();
+ }
+ }
+ }
+ return result;
Review Comment:
Nice catch, updated to assert before returning.
--
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]
