cadonna commented on a change in pull request #10163:
URL: https://github.com/apache/kafka/pull/10163#discussion_r585546908
##########
File path:
streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamsPartitionAssignor.java
##########
@@ -741,41 +666,35 @@ private boolean populateClientStatesMap(final Map<UUID,
ClientState> clientState
}
/**
- * @param changelogsByStatefulTask map from stateful task to its set of
changelog topic partitions
* @param endOffsets the listOffsets result from the adminClient
* @param sourceChangelogEndOffsets the end (committed) offsets of
optimized source changelogs
- * @param newlyCreatedChangelogPartitions any changelogs that were just
created duringthis assignment
+ * @param changelogTopics object that manages changelog topics
*
* @return Map from stateful task to its total end offset summed across
all changelog partitions
*/
- private Map<TaskId, Long> computeEndOffsetSumsByTask(final Map<TaskId,
Set<TopicPartition>> changelogsByStatefulTask,
- final
Map<TopicPartition, ListOffsetsResultInfo> endOffsets,
+ private Map<TaskId, Long> computeEndOffsetSumsByTask(final
Map<TopicPartition, ListOffsetsResultInfo> endOffsets,
final
Map<TopicPartition, Long> sourceChangelogEndOffsets,
- final
Collection<TopicPartition> newlyCreatedChangelogPartitions) {
+ final ChangelogTopics
changelogTopics) {
+
final Map<TaskId, Long> taskEndOffsetSums = new HashMap<>();
- for (final Map.Entry<TaskId, Set<TopicPartition>> taskEntry :
changelogsByStatefulTask.entrySet()) {
- final TaskId task = taskEntry.getKey();
- final Set<TopicPartition> changelogs = taskEntry.getValue();
-
- taskEndOffsetSums.put(task, 0L);
- for (final TopicPartition changelog : changelogs) {
- final long changelogEndOffset;
- if (newlyCreatedChangelogPartitions.contains(changelog)) {
- changelogEndOffset = 0L;
- } else if (sourceChangelogEndOffsets.containsKey(changelog)) {
- changelogEndOffset =
sourceChangelogEndOffsets.get(changelog);
- } else if (endOffsets.containsKey(changelog)) {
- changelogEndOffset = endOffsets.get(changelog).offset();
+ for (final TaskId taskId : changelogTopics.taskIds()) {
Review comment:
Will do!
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
[email protected]
