introduce NamedTaskId to avoid public interface

streams/src/main/java/org/apache/kafka/streams/processor/TaskId.java

@@ -23,7 +23,6 @@
 import java.io.IOException;
 import java.nio.ByteBuffer;
 import java.util.Objects;
-import java.util.Optional;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -41,7 +40,7 @@ public class TaskId implements Comparable<TaskId> {
     /** The ID of the partition. */
     public final int partition;
     /** The namedTopology that this task belongs to, or null if it does not belong to one */
-    private final String namedTopology;
+    protected final String namedTopology;
 
     public TaskId(final int topicGroupId, final int partition) {
         this(topicGroupId, partition, null);
@@ -59,10 +58,6 @@ public TaskId(final int topicGroupId, final int partition, final String namedTop
         }
     }
 
-    public String namedTopology() {
-        return namedTopology;
-    }
-
     @Override
     public String toString() {
         return namedTopology != null ? namedTopology + "_" + topicGroupId + "_" + partition : topicGroupId + "_" + partition;

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

@@ -50,6 +50,7 @@
 import org.apache.kafka.streams.processor.internals.assignment.StickyTaskAssignor;
 import org.apache.kafka.streams.processor.internals.assignment.SubscriptionInfo;
 import org.apache.kafka.streams.processor.internals.assignment.TaskAssignor;
+import org.apache.kafka.streams.processor.internals.namedtopology.NamedTaskId;
 import org.apache.kafka.streams.state.HostInfo;
 import org.slf4j.Logger;
 
@@ -515,7 +516,7 @@ private void populateTasksForMaps(final Map<TopicPartition, TaskId> taskForParti
             }
             allAssignedPartitions.addAll(partitions);
 
-            tasksForTopicGroup.computeIfAbsent(new Subtopology(id.topicGroupId, id.namedTopology()), k -> new HashSet<>()).add(id);
+            tasksForTopicGroup.computeIfAbsent(new Subtopology(id.topicGroupId, NamedTaskId.namedTopology(id)), k -> new HashSet<>()).add(id);
         }
 
         checkAllPartitions(allSourceTopics, partitionsForTask, allAssignedPartitions, fullMetadata);

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

@@ -0,0 +1,40 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.streams.processor.internals.namedtopology;
+
+import org.apache.kafka.streams.processor.TaskId;
+
+public class NamedTaskId extends TaskId {
+    public NamedTaskId(final int topicGroupId, final int partition, final String namedTopology) {
+        super(topicGroupId, partition, namedTopology);
+        if (namedTopology == null) {
+            throw new IllegalStateException("NamedTopology is required for a NamedTaskId");
+        }
+    }
+
+    public String namedTopology() {
+        return namedTopology;
+    }
+
+    public static String namedTopology(final TaskId taskId) {
+        if (taskId instanceof NamedTaskId) {
+            return ((NamedTaskId) taskId).namedTopology();
+        } else {
+            return null;
+        }
+    }
+}