You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@iceberg.apache.org by "danielcweeks (via GitHub)" <gi...@apache.org> on 2025/11/07 04:23:33 UTC
[PR] Merge control topic and last persisted offests [iceberg]
danielcweeks opened a new pull request, #14525:
URL: https://github.com/apache/iceberg/pull/14525
The control group topic offsets that are persisted in the table state my include a subset of partitions of some partitions do not have values. We currently overwrite the previous value, but that may remove partition offset information of not all partitions are included in the most recent commit.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Merge control topic and last persisted offests [iceberg]
Posted by "ajreid21 (via GitHub)" <gi...@apache.org>.
ajreid21 commented on code in PR #14525:
URL: https://github.com/apache/iceberg/pull/14525#discussion_r2504438680
##########
kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/channel/Coordinator.java:
##########
@@ -153,16 +154,15 @@ private void commit(boolean partialCommit) {
private void doCommit(boolean partialCommit) {
Map<TableReference, List<Envelope>> commitMap = commitState.tableCommitMap();
-
- String offsetsJson = offsetsJson();
OffsetDateTime validThroughTs = commitState.validThroughTs(partialCommit);
Tasks.foreach(commitMap.entrySet())
.executeWith(exec)
.stopOnFailure()
.run(
entry -> {
- commitToTable(entry.getKey(), entry.getValue(), offsetsJson, validThroughTs);
+ commitToTable(
+ entry.getKey(), entry.getValue(), controlTopicOffsets(), validThroughTs);
});
// we should only get here if all tables committed successfully...
Review Comment:
@danielcweeks @bryanck Does the method below (committing the control topic offsets) need to have the merged offsets as well?
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Merge control topic and last persisted offests [iceberg]
Posted by "bryanck (via GitHub)" <gi...@apache.org>.
bryanck commented on PR #14525:
URL: https://github.com/apache/iceberg/pull/14525#issuecomment-3503018235
Very nice catch, thanks for fixing this!
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Merge control topic and last persisted offests [iceberg]
Posted by "singhpk234 (via GitHub)" <gi...@apache.org>.
singhpk234 commented on code in PR #14525:
URL: https://github.com/apache/iceberg/pull/14525#discussion_r2501947197
##########
kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/channel/Coordinator.java:
##########
@@ -206,7 +206,15 @@ private void commitToTable(
String branch = config.tableConfig(tableIdentifier.toString()).commitBranch();
+ // Control topic partition offsets may include a subset of partition ids if there were no
+ // records for other partitions. Merge the updated topic partitions with the last committed
+ // offsets.
Map<Integer, Long> committedOffsets = lastCommittedOffsetsForTable(table, branch);
+ Map<Integer, Long> mergedOffsets =
+ Stream.of(committedOffsets, controlTopicOffsets)
+ .flatMap(map -> map.entrySet().stream())
+ .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue, Long::max));
Review Comment:
IIUC this will merge both maps and in case of matching keys take the max Val ? wondering this case:
Lets say my committed offset :
[0, 100] [2, 250]
my control topic offset :
[1, 150] [2, 200]
my output [0, 100] [1,150] [2, 250], but i think we did went back as we [2, 200] in control offset ? is this check flagged already somewhere ?
##########
kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/channel/Coordinator.java:
##########
@@ -206,7 +206,15 @@ private void commitToTable(
String branch = config.tableConfig(tableIdentifier.toString()).commitBranch();
+ // Control topic partition offsets may include a subset of partition ids if there were no
+ // records for other partitions. Merge the updated topic partitions with the last committed
+ // offsets.
Map<Integer, Long> committedOffsets = lastCommittedOffsetsForTable(table, branch);
+ Map<Integer, Long> mergedOffsets =
+ Stream.of(committedOffsets, controlTopicOffsets)
+ .flatMap(map -> map.entrySet().stream())
+ .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue, Long::max));
Review Comment:
[doubt] IIUC this will merge both maps and in case of matching keys take the max Val ? wondering this case:
Lets say my committed offset :
[0, 100] [2, 250]
my control topic offset :
[1, 150] [2, 200]
my output [0, 100] [1,150] [2, 250], but i think we did went back as we [2, 200] in control offset ? is this check flagged already somewhere ?
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Merge control topic and last persisted offests [iceberg]
Posted by "bryanck (via GitHub)" <gi...@apache.org>.
bryanck commented on PR #14525:
URL: https://github.com/apache/iceberg/pull/14525#issuecomment-3503047985
One thing to note is that tables that currently have only a subset of offsets might need some commit cycles until they are fixed.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Merge control topic and last persisted offests [iceberg]
Posted by "bryanck (via GitHub)" <gi...@apache.org>.
bryanck commented on PR #14525:
URL: https://github.com/apache/iceberg/pull/14525#issuecomment-3503019861
cc @kumarpritam863 you might be interested
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Merge control topic and last persisted offests [iceberg]
Posted by "danielcweeks (via GitHub)" <gi...@apache.org>.
danielcweeks merged PR #14525:
URL: https://github.com/apache/iceberg/pull/14525
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Merge control topic and last persisted offests [iceberg]
Posted by "bryanck (via GitHub)" <gi...@apache.org>.
bryanck commented on code in PR #14525:
URL: https://github.com/apache/iceberg/pull/14525#discussion_r2503975452
##########
kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/channel/Coordinator.java:
##########
@@ -206,7 +206,15 @@ private void commitToTable(
String branch = config.tableConfig(tableIdentifier.toString()).commitBranch();
+ // Control topic partition offsets may include a subset of partition ids if there were no
+ // records for other partitions. Merge the updated topic partitions with the last committed
+ // offsets.
Map<Integer, Long> committedOffsets = lastCommittedOffsetsForTable(table, branch);
+ Map<Integer, Long> mergedOffsets =
+ Stream.of(committedOffsets, controlTopicOffsets)
+ .flatMap(map -> map.entrySet().stream())
+ .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue, Long::max));
Review Comment:
This is the correct behavior. The offsets in the table correspond to the data committed. If the user did want to roll back the offsets, they also need to roll back the offsets in the table, either by rolling back the snapshot or by updating the offsets manually. (The hope was we'd have tools to help with that at some point.)
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org