[
https://issues.apache.org/jira/browse/KAFKA-21297?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Chia-Ping Tsai resolved KAFKA-21297.
------------------------------------
Fix Version/s: 4.5
Resolution: Fixed
> Refactor the maybeUpdateLimitOffsetsForStandbyChangelogs
> --------------------------------------------------------
>
> Key: KAFKA-21297
> URL: https://issues.apache.org/jira/browse/KAFKA-21297
> Project: Kafka
> Issue Type: Improvement
> Components: streams
> Reporter: Chia-Ping Tsai
> Assignee: Ziyun Fu
> Priority: Trivial
> Fix For: 4.5
>
>
> {code:java}
> for (final TopicPartition partition : changelogsWithLimitOffsets)
> {
> if (!changelogs.get(partition).bufferedRecords().isEmpty()) {
>
> updateLimitOffsetsForStandbyChangelogs(committedOffsetForChangelogs(tasks,
> changelogsWithLimitOffsets));
> lastUpdateOffsetTime = time.milliseconds();
> break;
> }
> } {code}
> The “break” definitely misled me into thinking we skip the other committed
> offsets. Then I realized what it actually does is update “all” if “any”
> matches.
> It could be simplified by following code
> {code:java}
> if (changelogsWithLimitOffsets.stream().anyMatch(p ->
> !changelogs.get(p).bufferedRecords().isEmpty())) {
>
> updateLimitOffsetsForStandbyChangelogs(committedOffsetForChangelogs(tasks,
> changelogsWithLimitOffsets));
> lastUpdateOffsetTime = time.milliseconds();
> } {code}
--
This message was sent by Atlassian Jira
(v10.3.26#10030026)
