checkstyle

apache · Mar 4, 2020 · 951c245 · 951c245
1 parent 361e449
commit 951c245
Show file tree

Hide file tree

Showing 3 changed files with 8 additions and 10 deletions.
diff --git a/.../src/main/java/org/apache/kafka/streams/processor/internals/StreamsPartitionAssignor.java b/.../src/main/java/org/apache/kafka/streams/processor/internals/StreamsPartitionAssignor.java
@@ -565,7 +565,7 @@ public GroupAssignment assign(final Cluster metadata, final GroupSubscription gr
             // there are two cases where we need to construct the prevTasks from the ownedPartitions:
             // 1) COOPERATIVE clients on version 2.4-2.5 do not encode active tasks and rely on ownedPartitions instead
             // 2) future clientduring version probing: we can't decode the future subscription info's prev tasks
-            if (!state.ownedPartitions().isEmpty() && (uuid == futureId ||state.prevActiveTasks().isEmpty())) {
+            if (!state.ownedPartitions().isEmpty() && (uuid == futureId || state.prevActiveTasks().isEmpty())) {
                 final Set<TaskId> previousActiveTasks = new HashSet<>();
                 for (final Map.Entry<TopicPartition, String> partitionEntry : state.ownedPartitions().entrySet()) {
                     final TopicPartition tp = partitionEntry.getKey();

diff --git a/.../test/java/org/apache/kafka/streams/processor/internals/StreamsPartitionAssignorTest.java b/.../test/java/org/apache/kafka/streams/processor/internals/StreamsPartitionAssignorTest.java
@@ -72,6 +72,7 @@
 import static org.apache.kafka.common.utils.Utils.mkMap;
 import static org.apache.kafka.common.utils.Utils.mkSet;
 import static org.apache.kafka.streams.processor.internals.assignment.StreamsAssignmentProtocolVersions.LATEST_SUPPORTED_VERSION;
+import static org.apache.kafka.streams.processor.internals.assignment.SubscriptionInfo.ACTIVE_TASK_SENTINEL_OFFSET;
 import static org.hamcrest.CoreMatchers.equalTo;
 import static org.hamcrest.CoreMatchers.not;
 import static org.hamcrest.MatcherAssert.assertThat;
@@ -232,10 +233,10 @@ private static SubscriptionInfo getInfo(final int version,
     }
 
     private static SubscriptionInfo getInfo(final UUID processId,
-                                            final Map<TaskId, Integer> taskLags,
+                                            final Map<TaskId, Long> taskOffsetSums,
                                             final String userEndPoint) {
         return new SubscriptionInfo(
-            LATEST_SUPPORTED_VERSION, LATEST_SUPPORTED_VERSION, processId, userEndPoint, taskLags);
+            LATEST_SUPPORTED_VERSION, LATEST_SUPPORTED_VERSION, processId, userEndPoint, taskOffsetSums);
     }
 
     private static SubscriptionInfo getInfo(final UUID processId, final Set<TaskId> prevTasks, final Set<TaskId> standbyTasks) {
@@ -1982,10 +1983,10 @@ private static void assertEquivalentAssignment(final Map<String, List<TaskId>> t
         }
     }
 
-    static Map<TaskId, Integer> getTaskOffsetSums(final Set<TaskId> activeTasks, final Set<TaskId> standbyTasks) {
-        final Map<TaskId, Integer> taskLags = activeTasks.stream().collect(Collectors.toMap(t -> t, t -> -1));
-        taskLags.putAll(standbyTasks.stream().collect(Collectors.toMap(t -> t, t -> 0)));
-        return taskLags;
+    static Map<TaskId, Long> getTaskOffsetSums(final Set<TaskId> activeTasks, final Set<TaskId> standbyTasks) {
+        final Map<TaskId, Long> taskOffsetSums = activeTasks.stream().collect(Collectors.toMap(t -> t, t -> ACTIVE_TASK_SENTINEL_OFFSET));
+        taskOffsetSums.putAll(standbyTasks.stream().collect(Collectors.toMap(t -> t, t -> 0L)));
+        return taskOffsetSums;
     }
 
 }
diff --git a/.../org/apache/kafka/streams/processor/internals/assignment/LegacySubscriptionInfoSerde.java b/.../org/apache/kafka/streams/processor/internals/assignment/LegacySubscriptionInfoSerde.java
@@ -18,9 +18,6 @@
 
 import static org.apache.kafka.streams.processor.internals.assignment.StreamsAssignmentProtocolVersions.LATEST_SUPPORTED_VERSION;
 
-import java.util.HashMap;
-import java.util.Map;
-import java.util.stream.Collectors;
 import org.apache.kafka.streams.errors.TaskAssignmentException;
 import org.apache.kafka.streams.processor.TaskId;
 import org.slf4j.Logger;