laskoviymishka commented on code in PR #17552:
URL: https://github.com/apache/iceberg/pull/17552#discussion_r3749659467
##########
kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/channel/Channel.java:
##########
@@ -143,12 +144,31 @@ protected Map<Integer, Long> controlTopicOffsets() {
}
protected void commitConsumerOffsets() {
+ Set<TopicPartition> partitions =
+ controlTopicOffsets().keySet().stream()
+ .map(k -> new TopicPartition(controlTopic, k))
+ .collect(Collectors.toSet());
+ Map<TopicPartition, OffsetAndMetadata> committed =
consumer.committed(partitions);
Review Comment:
I think this new `consumer.committed(partitions)` is the riskiest part of
the change. It's a synchronous round-trip to the group coordinator on every
commit cycle, and the no-timeout overload falls back to
`default.api.timeout.ms` (60s). It also isn't a `CommitFailedException`, so if
the broker is briefly unreachable the `TimeoutException` propagates straight
through `doCommit()` into `commit()`, where anything that isn't
`CommitFailedException` is treated as non-retryable and terminates the task
(`Coordinator.java:171-175`). The old path never made this RPC, so this is a
new way to kill the connector on a transient blip.
I'd avoid the round-trip entirely: cache the last-committed offset per
partition in a local field, update it after each successful `commitSync`, and
run the monotonicity check against the cache. That keeps the guard purely local
and drops the extra `controlTopicOffsets()` calls below too. If we do want to
keep the RPC, at minimum pass an explicit `Duration` and catch
`TimeoutException` so we fall back to committing rather than dying. wdyt?
##########
kafka-connect/kafka-connect/src/test/java/org/apache/iceberg/connect/channel/TestCoordinator.java:
##########
@@ -316,6 +319,35 @@ private void triggerCommitCycle(Coordinator coordinator) {
coordinator.process();
}
+ @Test
+ public void commitConsumerOffsetsShouldNotCommitLowerOffset() {
+ when(config.commitIntervalMs()).thenReturn(0);
+ when(config.commitTimeoutMs()).thenReturn(Integer.MAX_VALUE);
+
+ SinkTaskContext context = mock(SinkTaskContext.class);
+ Coordinator coordinator =
+ new Coordinator(catalog, config, ImmutableList.of(), clientFactory,
context);
+ coordinator.start();
+ initConsumer();
+
+ TopicPartition ctl = new TopicPartition(CTL_TOPIC_NAME, 0);
+
+ long healthyWatermark = 100L;
+ consumer.commitSync(ImmutableMap.of(ctl, new
OffsetAndMetadata(healthyWatermark)));
+
+ coordinator.controlTopicOffsets().put(0, 5L);
Review Comment:
This only exercises the single-partition, fully-behind case. The branch I'd
most want covered is `lastCommitted == null` — a fresh coordinator that's never
committed this partition, which is exactly the recovery path the guard has to
get right — plus a mixed case where one partition is ahead and another behind,
to confirm only the ahead one commits. Right now the `null` branch in the
production code is never hit by any test. Could we add those two?
##########
kafka-connect/kafka-connect/src/test/java/org/apache/iceberg/connect/channel/TestCoordinator.java:
##########
@@ -316,6 +319,35 @@ private void triggerCommitCycle(Coordinator coordinator) {
coordinator.process();
}
+ @Test
+ public void commitConsumerOffsetsShouldNotCommitLowerOffset() {
+ when(config.commitIntervalMs()).thenReturn(0);
+ when(config.commitTimeoutMs()).thenReturn(Integer.MAX_VALUE);
+
+ SinkTaskContext context = mock(SinkTaskContext.class);
+ Coordinator coordinator =
+ new Coordinator(catalog, config, ImmutableList.of(), clientFactory,
context);
+ coordinator.start();
+ initConsumer();
+
+ TopicPartition ctl = new TopicPartition(CTL_TOPIC_NAME, 0);
+
+ long healthyWatermark = 100L;
+ consumer.commitSync(ImmutableMap.of(ctl, new
OffsetAndMetadata(healthyWatermark)));
+
+ coordinator.controlTopicOffsets().put(0, 5L);
+ coordinator.commitConsumerOffsets();
+
+ long committed =
+ consumer.committed(Set.of(ctl)).get(ctl) == null
+ ? 0L
+ : consumer.committed(Set.of(ctl)).get(ctl).offset();
+
+ assertThat(committed)
+ .as("commitConsumerOffsets should not rewind the shared -coord
consumer group offsets")
+ .isGreaterThanOrEqualTo(healthyWatermark);
Review Comment:
This doesn't quite prove the guard fired.
`isGreaterThanOrEqualTo(healthyWatermark)` passes for any value ≥ 100, so it
wouldn't fail if something committed 200, and it can't tell "guard blocked the
rewind" apart from "guard never ran and 100 was already there." I'd assert
`isEqualTo(healthyWatermark)` — the fix's whole contract is that the committed
offset doesn't move off 100 when the local offset (5) is behind, so pin it to
exactly 100. wdyt?
##########
kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/channel/Channel.java:
##########
@@ -143,12 +144,31 @@ protected Map<Integer, Long> controlTopicOffsets() {
}
protected void commitConsumerOffsets() {
+ Set<TopicPartition> partitions =
+ controlTopicOffsets().keySet().stream()
+ .map(k -> new TopicPartition(controlTopic, k))
+ .collect(Collectors.toSet());
+ Map<TopicPartition, OffsetAndMetadata> committed =
consumer.committed(partitions);
+
Map<TopicPartition, OffsetAndMetadata> offsetsToCommit = Maps.newHashMap();
controlTopicOffsets()
.forEach(
- (k, v) ->
- offsetsToCommit.put(new TopicPartition(controlTopic, k), new
OffsetAndMetadata(v)));
- consumer.commitSync(offsetsToCommit);
+ (partition, offsetToCommit) -> {
+ TopicPartition tp = new TopicPartition(controlTopic, partition);
+ OffsetAndMetadata lastCommitted = committed.get(tp);
+ if (lastCommitted == null || offsetToCommit >
lastCommitted.offset()) {
Review Comment:
You already call this out in the description, so I mostly want to settle the
framing rather than block on it: the read-compare-commit here is still a TOCTOU
window. If A reads committed=100, then B advances committed to 150, then A
passes the guard with in-memory 120 (120 > 100) and commits, we've rewound 150
back to 120.
The "duplicates not data loss" framing holds for the table state, but the
rewound `-coord` offset means the next coordinator re-reads stale control
records, and a re-processed commit can throw `ValidationException` — which with
`commitMaxConsecutiveFailures=1` terminates the connector. So this closes the
gross fully-stale case but not the concurrent-coordinator one. I don't think we
need to solve the general case here, but could we note in the code that the
window is intentionally left open, so a follow-up doesn't assume it's fully
closed? wdyt?
##########
kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/channel/Channel.java:
##########
@@ -143,12 +144,31 @@ protected Map<Integer, Long> controlTopicOffsets() {
}
protected void commitConsumerOffsets() {
+ Set<TopicPartition> partitions =
+ controlTopicOffsets().keySet().stream()
+ .map(k -> new TopicPartition(controlTopic, k))
+ .collect(Collectors.toSet());
+ Map<TopicPartition, OffsetAndMetadata> committed =
consumer.committed(partitions);
+
Map<TopicPartition, OffsetAndMetadata> offsetsToCommit = Maps.newHashMap();
controlTopicOffsets()
Review Comment:
Small thing: `controlTopicOffsets()` gets called three times in this method
(the `partitions` build, the `.forEach`, and the skip-path log). It's
`protected` and overridable, so I'd read it into a local once at the top and
use that throughout — makes the intent explicit and removes any chance of a
subclass returning something different across the three reads.
##########
kafka-connect/kafka-connect/src/test/java/org/apache/iceberg/connect/channel/TestCoordinator.java:
##########
@@ -316,6 +319,35 @@ private void triggerCommitCycle(Coordinator coordinator) {
coordinator.process();
}
+ @Test
+ public void commitConsumerOffsetsShouldNotCommitLowerOffset() {
+ when(config.commitIntervalMs()).thenReturn(0);
+ when(config.commitTimeoutMs()).thenReturn(Integer.MAX_VALUE);
+
+ SinkTaskContext context = mock(SinkTaskContext.class);
+ Coordinator coordinator =
+ new Coordinator(catalog, config, ImmutableList.of(), clientFactory,
context);
+ coordinator.start();
+ initConsumer();
+
+ TopicPartition ctl = new TopicPartition(CTL_TOPIC_NAME, 0);
+
+ long healthyWatermark = 100L;
+ consumer.commitSync(ImmutableMap.of(ctl, new
OffsetAndMetadata(healthyWatermark)));
+
+ coordinator.controlTopicOffsets().put(0, 5L);
+ coordinator.commitConsumerOffsets();
+
+ long committed =
+ consumer.committed(Set.of(ctl)).get(ctl) == null
Review Comment:
`consumer.committed(Set.of(ctl))` runs twice here — once for the null check,
once for `.offset()`. Harmless against `MockConsumer`, but it's the same
two-RPC pattern we'd want to avoid in real code, so I'd read it once into a
local and branch off that. While we're here, the rest of the file uses the
relocated `ImmutableSet.of(...)` rather than JDK `Set.of` — worth matching for
consistency.
##########
kafka-connect/kafka-connect/src/test/java/org/apache/iceberg/connect/channel/TestCoordinator.java:
##########
@@ -316,6 +319,35 @@ private void triggerCommitCycle(Coordinator coordinator) {
coordinator.process();
}
+ @Test
+ public void commitConsumerOffsetsShouldNotCommitLowerOffset() {
Review Comment:
Every other test in this file uses the `testXxx` prefix — mind renaming to
something like `testCommitConsumerOffsetsDoesNotRewind` to match?
##########
kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/channel/Channel.java:
##########
@@ -143,12 +144,31 @@ protected Map<Integer, Long> controlTopicOffsets() {
}
protected void commitConsumerOffsets() {
+ Set<TopicPartition> partitions =
+ controlTopicOffsets().keySet().stream()
+ .map(k -> new TopicPartition(controlTopic, k))
+ .collect(Collectors.toSet());
+ Map<TopicPartition, OffsetAndMetadata> committed =
consumer.committed(partitions);
+
Map<TopicPartition, OffsetAndMetadata> offsetsToCommit = Maps.newHashMap();
controlTopicOffsets()
.forEach(
- (k, v) ->
- offsetsToCommit.put(new TopicPartition(controlTopic, k), new
OffsetAndMetadata(v)));
- consumer.commitSync(offsetsToCommit);
+ (partition, offsetToCommit) -> {
+ TopicPartition tp = new TopicPartition(controlTopic, partition);
+ OffsetAndMetadata lastCommitted = committed.get(tp);
+ if (lastCommitted == null || offsetToCommit >
lastCommitted.offset()) {
+ offsetsToCommit.put(tp, new OffsetAndMetadata(offsetToCommit));
+ }
+ });
+ if (!offsetsToCommit.isEmpty()) {
+ LOG.info("Coordinator committing offsets: {}", offsetsToCommit);
+ consumer.commitSync(offsetsToCommit);
+ } else {
+ LOG.info(
Review Comment:
This skip branch fires on every commit cycle where nothing advanced — for an
idle connector with a short commit interval that's a steady stream of INFO
lines, so I'd drop it to `LOG.debug`. Also worth noting the message reads "not
ahead of committed offsets" for both the equal and behind cases; if that
distinction ever matters for debugging it'd be nice to separate them, but
debug-level makes it far less pressing.
--
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]
