wombatu-kun commented on code in PR #16360:
URL: https://github.com/apache/iceberg/pull/16360#discussion_r3360125288
##########
kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/channel/Coordinator.java:
##########
@@ -239,6 +268,9 @@ private void commitToTable(
envelopeList.stream()
.filter(
envelope -> {
+ if (controlTopicReset) {
Review Comment:
The merged-offset computation already routes through `baseOffsets` (line
260), which is `Map.of()` when `controlTopicReset` is true. If the dedup filter
below looked up `baseOffsets.get(envelope.partition())` instead of
`committedOffsets`, the existing `minOffset == null` branch would already pass
every envelope on reset, making this `if (controlTopicReset) return true;`
removable. That folds the dedup baseline and the merge baseline into one source
of truth, leaving `controlTopicReset` to gate only the warn log. The
`offsetValidator` calls must keep using `committedOffsets`.
--
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]
