laskoviymishka commented on code in PR #18039:
URL: https://github.com/apache/iceberg/pull/18039#discussion_r3981219594
##########
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:
The format change here is the right call, but it quietly changes two things
worth calling out. The suffix is gone from the coordinator id — anyone who set
the internal transactional suffix prop loses it on upgrade with no warning, so
I'd either drop it deliberately and note it, or keep honoring it.
The other one is the upgrade window: the old coordinator holds the epoch on
the old `...coordinator` id while the new one initializes on
`...<connectGroupId>-coordinator`, so for the duration of a rolling restart
they're in different transactional-id namespaces and can't fence each other.
Snapshot validation still protects the data, but the fencing guarantee has a
gap exactly during the transition. An upgrade note covering both would do it —
wdyt?
##########
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:
`beginTransaction()` is outside the try, so when the producer is fenced the
exception comes from `sendOffsetsToTransaction()`/`commitTransaction()` and we
land in this catch. But `abortTransaction()` on an already-fenced producer is
itself fatal — it throws again, we swallow it, and emit `LOG.warn("Error
aborting producer transaction", ...)`.
So every clean, expected fencing event — the exact happy path this PR is
built around — produces a scary WARN that looks like a real failure. The
original `ProducerFencedException` still propagates via `throw e`, so detection
works; it's just noisy and alert-prone. I'd skip the abort when the cause is
already fatal:
```java
} catch (Exception e) {
if (!(e instanceof ProducerFencedException)
&& !(e instanceof InvalidProducerEpochException)) {
try {
producer.abortTransaction();
} catch (Exception ex) {
LOG.warn("Error aborting producer transaction", ex);
}
}
throw e;
}
```
The same shape exists in `send()` — worth fixing both while we're here.
##########
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:
Quick check on this one: if `transactionalSuffix()` can return null when the
suffix prop is unset, this concatenates to `...workernull` for every task on
the connector, and they'd all share one transactional id and fence each other
on `initTransactions()`. With eager `initTransactions()` that surfaces as
`ProducerFencedException` on every worker commit.
Is the suffix guaranteed to default to `""` here rather than null? If so
this is fine; if not, I'd add a default so it can't null-concatenate.
##########
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:
This exercises the CoordinatorThread → CommitterImpl dispatch, but the
Coordinator is mocked so it doesn't cover the layer that actually matters: a
real Channel producer getting fenced, `Coordinator.process()` propagating it,
and CoordinatorThread seeing it as fenced. That's the same chain where the
abort-catch in `commitConsumerOffsets()` could swallow or reshape the exception.
A test that builds a Coordinator over a `MockProducer` with
`fenceProducer()` called, runs a real commit cycle, and asserts `isFenced()`
then that the committer clears without throwing would lock the propagation end
to end. wdyt?
##########
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:
This reads like fencing covers the commit end to end, but it's the offset
commit that's fenced — the table snapshot commit in `Coordinator.doCommit()`
runs outside any Kafka transaction and still relies on
`SnapshotAncestryValidator` for the stale-coordinator case.
A stale coordinator can still land a stale snapshot in the window between
the new coordinator's `initTransactions()` and its own fenced offset commit;
that's caught at the Iceberg level, not by epoch fencing. I'd make that
boundary explicit here (and maybe a line in `doCommit()`) so nobody over-trusts
the Kafka guarantee. wdyt?
##########
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();
+
+ CoordinatorThread coordinatorThread = new CoordinatorThread(coordinator);
+ coordinatorThread.start();
+
+ verify(coordinator, timeout(1000)).stop();
+ assertThat(coordinatorThread.isTerminated()).isTrue();
+ assertThat(coordinatorThread.isFenced()).isTrue();
+
+ CommitterImpl committer = new CommitterImpl();
+ Field field = CommitterImpl.class.getDeclaredField("coordinatorThread");
+ field.setAccessible(true);
+ field.set(committer, coordinatorThread);
+
+ committer.save(Collections.emptyList());
+ assertThat(field.get(committer)).isNull();
+ }
+
+ @Test
+ public void testRequestedTerminationWithNoRecordedErrorIsTreatedAsFatal()
Review Comment:
This asserts the `terminated && !isFenced() && error == null` branch, but I
don't think production can reach that state from explicit termination —
`stopCoordinator()` does `terminate()` then nulls `coordinatorThread` on the
same thread, so `processControlEvents()` never observes a
terminated-with-no-error thread. Either drop this or add a comment that it's
guarding an unreachable case and point at `stopCoordinator()`.
##########
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:
All three of us landed on this independently: the whole test hinges on
`staleCoordinatorId` and `newCoordinatorId` being the same id, but that's never
asserted. It won't pass silently if they diverge — the `assertThatThrownBy`
would just see no exception and fail — but the precondition is invisible to
anyone reading it.
I'd assert it directly so the stability property is the thing under test,
not an assumption:
```java
assertThat(newCoordinatorId)
.as("coordinator transactional id must be stable for the same connector
config")
.isEqualTo(staleCoordinatorId);
```
A single `transactionalId` variable plus a comment that both coordinators
share it and `initTransactions()` on the second bumps the epoch would read even
more clearly.
##########
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:
`values().iterator().next()` pulls an arbitrary group's offsets — the outer
map is keyed by consumer group id, so once more than one group has committed
this returns whatever the map's iteration order hands back. Keying by the known
group id and asserting `containsKey` would make it deterministic. Same shape in
`TestCoordinator.lastCommittedOffset()` — worth fixing together.
##########
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:
When nothing was ever committed this returns a boxed `null`, and the failure
then reads `null is not equal to 100`, which hides that
`sendOffsetsToTransaction` was never called at all. An
`assertThat(result).isNotNull().as(...)` before returning would make that
failure mode obvious (and avoids an unboxing NPE if a caller ever uses the
value as a primitive `long`).
--
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]
