Repository navigation
Merge control topic and last persisted offests - #14525
Conversation
| Stream.of(committedOffsets, controlTopicOffsets) | ||
| .flatMap(map -> map.entrySet().stream()) | ||
| .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue, Long::max)); |
There was a problem hiding this 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 ?
There was a problem hiding this 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.)
|
Very nice catch, thanks for fixing this! |
|
cc @kumarpritam863 you might be interested |
|
One thing to note is that tables that currently have only a subset of offsets might need some commit cycles until they are fixed. |
There was a problem hiding this comment.
LGTM too, Great find, thanks @danielcweeks !
thanks @bryanck for the pointer (here)
should we mark it 1.10.1 ?
| entry.getKey(), entry.getValue(), controlTopicOffsets(), validThroughTs); | ||
| }); | ||
|
|
||
| // we should only get here if all tables committed successfully... |
There was a problem hiding this comment.
@danielcweeks @bryanck Does the method below (committing the control topic offsets) need to have the merged offsets as well?
* Merge control topic and last persisted offests * Add tests (cherry picked from commit 8da07dc)
* Merge control topic and last persisted offests * Add tests
* Merge control topic and last persisted offests * Add tests
4 participants
The control group topic offsets that are persisted in the table state my include a subset of partitions if 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.