thswlsqls commented on code in PR #17080:
URL: https://github.com/apache/iceberg/pull/17080#discussion_r4129463563
##########
kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/channel/CommitState.java:
##########
@@ -144,16 +144,21 @@ Map<TableReference, List<Envelope>> tableCommitMap() {
}
OffsetDateTime validThroughTs(boolean partialCommit) {
+ List<DataComplete> currentCommitReady =
+ readyBuffer.stream()
+ .filter(payload -> Objects.equals(currentCommitId,
payload.commitId()))
Review Comment:
@bryanck Thanks for reopening this and for the review. Agreed, I'll move the
commit ID check into `addReady()` so only the current commit's `DataComplete`
events go into `readyBuffer`, and drop the extra filter from `validThroughTs()`.
This keeps the same behavior: `startNewCommit()` only runs when no commit is
in progress, and `endCurrentCommit()` always clears the buffer, so filtering on
the way in yields the same set as filtering on read. It also stops zombie
events from piling up in the buffer.
--
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]
