KAFKA-12648: Pt. 1 - Add NamedTopology to protocol and state directory structure

checkstyle/suppressions.xml

@@ -195,7 +195,7 @@
 
     <!-- Streams tests -->
     <suppress checks="ClassFanOutComplexity"
-              files="(StreamsPartitionAssignorTest|StreamThreadTest|StreamTaskTest|TopologyTestDriverTest).java"/>
+              files="(StreamsPartitionAssignorTest|StreamThreadTest|StreamTaskTest|TaskManagerTest|TopologyTestDriverTest).java"/>
 
     <suppress checks="MethodLength"
               files="(EosIntegrationTest|EosV2UpgradeIntegrationTest|KStreamKStreamJoinTest|RocksDBWindowStoreTest).java"/>

streams/src/main/java/org/apache/kafka/streams/KafkaStreams.java

@@ -827,7 +827,7 @@ private KafkaStreams(final InternalTopologyBuilder internalTopologyBuilder,
             (hasGlobalTopology && globalTaskTopology.hasPersistentGlobalStore());
 
         try {
-            stateDirectory = new StateDirectory(config, time, hasPersistentStores);
+            stateDirectory = new StateDirectory(config, time, hasPersistentStores, internalTopologyBuilder.hasNamedTopologies());
             processId = stateDirectory.initializeProcessId();
         } catch (final ProcessorStateException fatal) {
             throw new StreamsException(fatal);

streams/src/main/java/org/apache/kafka/streams/processor/internals/InternalTopologyBuilder.java

@@ -2039,6 +2039,11 @@ public synchronized List<String> fullSourceTopicNames() {
         return maybeDecorateInternalSourceTopics(sourceTopicNames);
     }
 
+    public boolean hasNamedTopologies() {
+        // TODO KAFKA-12648: covered by Pt. 2
+        return false;
+    }
+
     // following functions are for test only
     public synchronized Set<String> sourceTopicNames() {
         return sourceTopicNames;

streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamsPartitionAssignor.java

@@ -1255,6 +1255,7 @@ public void onAssignment(final Assignment assignment, final ConsumerGroupMetadat
             case 7:
             case 8:
             case 9:
+            case 10:
                 validateActiveTaskEncoding(partitions, info, logPrefix);
 
                 activeTasks = getActiveTasks(partitions, info);

streams/src/main/java/org/apache/kafka/streams/processor/internals/TaskManager.java

@@ -36,9 +36,11 @@
 import org.apache.kafka.streams.errors.TaskIdFormatException;
 import org.apache.kafka.streams.errors.TaskMigratedException;
 import org.apache.kafka.streams.processor.TaskId;
+import org.apache.kafka.streams.processor.internals.StateDirectory.TaskDirectory;
 import org.apache.kafka.streams.processor.internals.Task.State;
 import org.apache.kafka.streams.processor.internals.metrics.StreamsMetricsImpl;
 import org.apache.kafka.streams.state.internals.OffsetCheckpoint;
+
 import org.slf4j.Logger;
 
 import java.io.File;
@@ -682,9 +684,11 @@ private void tryToLockAllNonEmptyTaskDirectories() {
         // current set of actually-locked tasks.
         lockedTaskDirectories.clear();
 
-        for (final File dir : stateDirectory.listNonEmptyTaskDirectories()) {
+        for (final TaskDirectory taskDir : stateDirectory.listNonEmptyTaskDirectories()) {
+            final File dir = taskDir.file();
+            final String namedTopology = taskDir.namedTopology();
             try {
-                final TaskId id = parseTaskDirectoryName(dir.getName(), null);
+                final TaskId id = parseTaskDirectoryName(dir.getName(), namedTopology);
                 if (stateDirectory.lock(id)) {
                     lockedTaskDirectories.add(id);
                     if (!tasks.owned(id)) {

streams/src/main/java/org/apache/kafka/streams/processor/internals/assignment/AssignmentInfo.java

@@ -184,6 +184,7 @@ public ByteBuffer encode() {
                 case 7:
                 case 8:
                 case 9:
+                case 10:
                     out.writeInt(usedVersion);
                     out.writeInt(commonlySupportedVersion);
                     encodeActiveAndStandbyTaskAssignment(out);
@@ -359,6 +360,7 @@ public static AssignmentInfo decode(final ByteBuffer data) {
                 case 7:
                 case 8:
                 case 9:
+                case 10:
                     commonlySupportedVersion = in.readInt();
                     assignmentInfo = new AssignmentInfo(usedVersion, commonlySupportedVersion);
                     decodeActiveTasks(assignmentInfo, in);

streams/src/main/java/org/apache/kafka/streams/processor/internals/assignment/ConsumerProtocolUtils.java

@@ -16,6 +16,7 @@
  */
 package org.apache.kafka.streams.processor.internals.assignment;
 
+import org.apache.kafka.streams.errors.TaskAssignmentException;
 import org.apache.kafka.streams.processor.TaskId;
 
 import java.io.DataInputStream;
@@ -43,6 +44,8 @@ public static void writeTaskIdTo(final TaskId taskId, final DataOutputStream out
             } else {
                 out.writeInt(0);
             }
+        } else if (taskId.namedTopology() != null) {
+            throw new TaskAssignmentException("Named topologies are not compatible with protocol version " + version);
         }
     }
 
@@ -78,6 +81,8 @@ public static void writeTaskIdTo(final TaskId taskId, final ByteBuffer buf, fina
             } else {
                 buf.putInt(0);
             }
+        } else if (taskId.namedTopology() != null) {
+            throw new TaskAssignmentException("Named topologies are not compatible with protocol version " + version);
         }
     }
 

streams/src/main/java/org/apache/kafka/streams/processor/internals/assignment/StreamsAssignmentProtocolVersions.java

@@ -20,10 +20,18 @@ public final class StreamsAssignmentProtocolVersions {
     public static final int UNKNOWN = -1;
     public static final int EARLIEST_PROBEABLE_VERSION = 3;
     public static final int MIN_NAMED_TOPOLOGY_VERSION = 10;
-    public static final int LATEST_SUPPORTED_VERSION = 9;
-    // When changing the versions:
-    // 1) Update the version_probing_message and end_of_upgrade_message in streams_upgrade_test.py::StreamsUpgradeTest.test_version_probing_upgrade
-    // 2) Add a unit test in SubscriptionInfoTest and/or AssignmentInfoTest
+    public static final int LATEST_SUPPORTED_VERSION = 10;
+    /*
+     * Any time you modify the subscription or assignment info, you need to bump the latest supported version, unless
+     * the version has already been bumped within the current release cycle.
+     *
+     * Last version bump: May 2021, before 3.0
+     *
+     * When changing the version:
+     * 1) Update the version_probing_message and end_of_upgrade_message in streams_upgrade_test.py::StreamsUpgradeTest.test_version_probing_upgrade
+     * 2) Add a unit test in SubscriptionInfoTest and/or AssignmentInfoTest
+     * 3) Note the date & corresponding Kafka version of this bump
+     */
 
     private StreamsAssignmentProtocolVersions() {}
 }

streams/src/main/java/org/apache/kafka/streams/processor/internals/assignment/SubscriptionInfo.java

@@ -31,6 +31,7 @@
 import org.apache.kafka.streams.internals.generated.SubscriptionInfoData.TaskOffsetSum;
 import org.apache.kafka.streams.processor.TaskId;
 import org.apache.kafka.streams.processor.internals.Task;
+
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -42,6 +43,7 @@
 import java.util.stream.Collectors;
 
 import static org.apache.kafka.streams.processor.internals.assignment.StreamsAssignmentProtocolVersions.LATEST_SUPPORTED_VERSION;
+import static org.apache.kafka.streams.processor.internals.assignment.StreamsAssignmentProtocolVersions.MIN_NAMED_TOPOLOGY_VERSION;
 
 public class SubscriptionInfo {
     private static final Logger LOG = LoggerFactory.getLogger(SubscriptionInfo.class);
@@ -94,8 +96,8 @@ public SubscriptionInfo(final int version,
 
         if (version >= 2) {
             data.setUserEndPoint(userEndPoint == null
-                                     ? new byte[0]
-                                     : userEndPoint.getBytes(StandardCharsets.UTF_8));
+                    ? new byte[0]
+                    : userEndPoint.getBytes(StandardCharsets.UTF_8));
         }
         if (version >= 3) {
             data.setLatestSupportedVersion(latestSupportedVersion);
@@ -109,7 +111,9 @@ public SubscriptionInfo(final int version,
 
         this.data = data;
 
-        if (version >= MIN_VERSION_OFFSET_SUM_SUBSCRIPTION) {
+        if (version >= MIN_NAMED_TOPOLOGY_VERSION) {
+            setTaskOffsetSumDataWithNamedTopologiesFromTaskOffsetSumMap(taskOffsetSums);
+        } else if (version >= MIN_VERSION_OFFSET_SUM_SUBSCRIPTION) {
             setTaskOffsetSumDataFromTaskOffsetSumMap(taskOffsetSums);
         } else {
             setPrevAndStandbySetsFromParsedTaskOffsetSumMap(taskOffsetSums);
@@ -125,10 +129,27 @@ public int errorCode() {
         return data.errorCode();
     }
 
+    // For version > MIN_NAMED_TOPOLOGY_VERSION
+    private void setTaskOffsetSumDataWithNamedTopologiesFromTaskOffsetSumMap(final Map<TaskId, Long> taskOffsetSums) {
+        data.setTaskOffsetSums(taskOffsetSums.entrySet().stream().map(t -> {
+            final SubscriptionInfoData.TaskOffsetSum taskOffsetSum = new SubscriptionInfoData.TaskOffsetSum();
+            final TaskId task = t.getKey();
+            taskOffsetSum.setTopicGroupId(task.subtopology());
+            taskOffsetSum.setPartition(task.partition());
+            taskOffsetSum.setNamedTopology(task.namedTopology());
+            taskOffsetSum.setOffsetSum(t.getValue());
+            return taskOffsetSum;
+        }).collect(Collectors.toList()));
+    }
+
+    // For MIN_NAMED_TOPOLOGY_VERSION > version > MIN_VERSION_OFFSET_SUM_SUBSCRIPTION
     private void setTaskOffsetSumDataFromTaskOffsetSumMap(final Map<TaskId, Long> taskOffsetSums) {
         final Map<Integer, List<SubscriptionInfoData.PartitionToOffsetSum>> topicGroupIdToPartitionOffsetSum = new HashMap<>();
         for (final Map.Entry<TaskId, Long> taskEntry : taskOffsetSums.entrySet()) {
             final TaskId task = taskEntry.getKey();
+            if (task.namedTopology() != null) {
+                throw new TaskAssignmentException("Named topologies are not compatible with older protocol versions");
+            }
             topicGroupIdToPartitionOffsetSum.computeIfAbsent(task.subtopology(), t -> new ArrayList<>()).add(
                 new SubscriptionInfoData.PartitionToOffsetSum()
                     .setPartition(task.partition())
@@ -143,11 +164,15 @@ private void setTaskOffsetSumDataFromTaskOffsetSumMap(final Map<TaskId, Long> ta
         }).collect(Collectors.toList()));
     }
 
+    // For MIN_VERSION_OFFSET_SUM_SUBSCRIPTION > version
     private void setPrevAndStandbySetsFromParsedTaskOffsetSumMap(final Map<TaskId, Long> taskOffsetSums) {
         final Set<TaskId> prevTasks = new HashSet<>();
         final Set<TaskId> standbyTasks = new HashSet<>();
 
         for (final Map.Entry<TaskId, Long> taskOffsetSum : taskOffsetSums.entrySet()) {
+            if (taskOffsetSum.getKey().namedTopology() != null) {
+                throw new TaskAssignmentException("Named topologies are not compatible with older protocol versions");
+            }
             if (taskOffsetSum.getValue() == Task.LATEST_OFFSET) {
                 prevTasks.add(taskOffsetSum.getKey());
             } else {
@@ -218,12 +243,20 @@ public Map<TaskId, Long> taskOffsetSums() {
             taskOffsetSumsCache = new HashMap<>();
             if (data.version() >= MIN_VERSION_OFFSET_SUM_SUBSCRIPTION) {
                 for (final TaskOffsetSum taskOffsetSum : data.taskOffsetSums()) {
-                    for (final PartitionToOffsetSum partitionOffsetSum : taskOffsetSum.partitionToOffsetSum()) {
+                    if (data.version() >= MIN_NAMED_TOPOLOGY_VERSION) {
                         taskOffsetSumsCache.put(
                             new TaskId(taskOffsetSum.topicGroupId(),
-                                       partitionOffsetSum.partition()),
-                            partitionOffsetSum.offsetSum()
-                        );
+                                       taskOffsetSum.partition(),
+                                       taskOffsetSum.namedTopology()),
+                            taskOffsetSum.offsetSum());
+                    } else {
+                        for (final PartitionToOffsetSum partitionOffsetSum : taskOffsetSum.partitionToOffsetSum()) {
+                            taskOffsetSumsCache.put(
+                                new TaskId(taskOffsetSum.topicGroupId(),
+                                           partitionOffsetSum.partition()),
+                                partitionOffsetSum.offsetSum()
+                            );
+                        }
                     }
                 }
             } else {

streams/src/main/java/org/apache/kafka/streams/processor/internals/namedtopology/NamedTopology.java

@@ -26,9 +26,7 @@ public class NamedTopology extends Topology {
 
     private final Logger log = LoggerFactory.getLogger(NamedTopology.class);
     private String name;
-
-
-
+
     void setTopologyName(final String newTopologyName) {
         if (name != null) {
             log.error("Unable to set topologyName = {} because the name is already set to {}", newTopologyName, name);
@@ -41,7 +39,6 @@ public String name() {
         return name;
     }
 
-
     public List<String> sourceTopics() {
         return super.internalTopologyBuilder.fullSourceTopicNames();
     }

streams/src/main/resources/common/message/SubscriptionInfoData.json

@@ -15,7 +15,7 @@
 
 {
   "name": "SubscriptionInfoData",
-  "validVersions": "1-9",
+  "validVersions": "1-10",
   "fields": [
     {
       "name": "version",
@@ -33,6 +33,7 @@
       "versions": "1+",
       "type": "uuid"
     },
+    /***** Protocol version 1-6 only (after 6 this is encoded in task offset sum map) *****/
     {
       "name": "prevTasks",
       "versions": "1-6",
@@ -43,6 +44,7 @@
       "versions": "1-6",
       "type": "[]TaskId"
     },
+    /***************/
     {
       "name": "userEndPoint",
       "versions": "2+",
@@ -65,18 +67,19 @@
     }
   ],
   "commonStructs": [
+    // TaskId was only used from 1-6, after 6 we encode each field of the TaskId separately along with the other information for that map entry
     {
       "name": "TaskId",
-      "versions": "1+",
+      "versions": "1-6",
       "fields": [
         {
           "name": "topicGroupId",
-          "versions": "1+",
+          "versions": "1-6",
           "type": "int32"
         },
         {
           "name": "partition",
-          "versions": "1+",
+          "versions": "1-6",
           "type": "int32"
         }
       ]
@@ -90,25 +93,45 @@
           "versions": "7+",
           "type": "int32"
         },
+        // Prior to version 10, in 7-9, the below fields (partition and offsetSum) were encoded via the nested
+        // partitionToOffsetSum struct. In 10+ all fields are encoded directly in the TaskOffsetSum struct
+        {
+          "name": "partition",
+          "versions": "10+",
+          "type": "int32"
+        },
+        {
+          "name": "offsetSum",
+          "versions": "10+",
+          "type": "int64"
+        },
+        {
+          "name": "namedTopology",
+          "versions": "10+",
+          "nullableVersions": "10+",
+          "ignorable":  "false",  // namedTopology is not ignorable because if you do, a TaskId may not be unique
+          "type": "string"
+        },
         {
           "name": "partitionToOffsetSum",
-          "versions": "7+",
+          "versions": "7-9",
           "type": "[]PartitionToOffsetSum"
         }
       ]
     },
+
     {
       "name": "PartitionToOffsetSum",
-      "versions": "7+",
+      "versions": "7-9",
       "fields": [
         {
           "name": "partition",
-          "versions": "7+",
+          "versions": "7-9",
           "type": "int32"
         },
         {
           "name": "offsetSum",
-          "versions": "7+",
+          "versions": "7-9",
           "type": "int64"
         }
       ]

streams/src/test/java/org/apache/kafka/streams/KafkaStreamsTest.java

@@ -97,6 +97,8 @@
 import static org.apache.kafka.streams.state.QueryableStoreTypes.keyValueStore;
 import static org.apache.kafka.streams.integration.utils.IntegrationTestUtils.safeUniqueTestName;
 import static org.apache.kafka.streams.integration.utils.IntegrationTestUtils.waitForApplicationState;
+
+import static org.easymock.EasyMock.anyBoolean;
 import static org.apache.kafka.test.TestUtils.waitForCondition;
 import static org.easymock.EasyMock.anyInt;
 import static org.easymock.EasyMock.anyLong;
@@ -948,7 +950,8 @@ public void shouldCleanupOldStateDirs() throws Exception {
         PowerMock.expectNew(StateDirectory.class,
             anyObject(StreamsConfig.class),
             anyObject(Time.class),
-            EasyMock.eq(true)
+            EasyMock.eq(true),
+            anyBoolean()
         ).andReturn(stateDirectory);
         EasyMock.expect(stateDirectory.initializeProcessId()).andReturn(UUID.randomUUID());
         stateDirectory.close();
@@ -1119,7 +1122,8 @@ private void startStreamsAndCheckDirExists(final Topology topology,
         PowerMock.expectNew(StateDirectory.class,
             anyObject(StreamsConfig.class),
             anyObject(Time.class),
-            EasyMock.eq(shouldFilesExist)
+            EasyMock.eq(shouldFilesExist),
+            anyBoolean()
         ).andReturn(stateDirectory);
         EasyMock.expect(stateDirectory.initializeProcessId()).andReturn(UUID.randomUUID());