From 951c245be7fbbe3a43c96b7f4f21956593ef0613 Mon Sep 17 00:00:00 2001
From: ableegoldman <sophie@confluent.io>
Date: Wed, 4 Mar 2020 14:22:07 -0500
Subject: [PATCH] checkstyle

---
 .../internals/StreamsPartitionAssignor.java         |  2 +-
 .../internals/StreamsPartitionAssignorTest.java     | 13 +++++++------
 .../assignment/LegacySubscriptionInfoSerde.java     |  3 ---
 3 files changed, 8 insertions(+), 10 deletions(-)

diff --git a/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamsPartitionAssignor.java b/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamsPartitionAssignor.java
index 60739891a9a43..f84bb3d4a67bd 100644
--- a/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamsPartitionAssignor.java
+++ b/streams/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/streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamsPartitionAssignorTest.java b/streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamsPartitionAssignorTest.java
index 96059031b7709..9ece80b09d389 100644
--- a/streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamsPartitionAssignorTest.java
+++ b/streams/src/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/streams/src/test/java/org/apache/kafka/streams/processor/internals/assignment/LegacySubscriptionInfoSerde.java b/streams/src/test/java/org/apache/kafka/streams/processor/internals/assignment/LegacySubscriptionInfoSerde.java
index 8ea2b9d41edc6..e8976ee53ea48 100644
--- a/streams/src/test/java/org/apache/kafka/streams/processor/internals/assignment/LegacySubscriptionInfoSerde.java
+++ b/streams/src/test/java/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;