KAFKA-12779: KIP-740, Introduce public TaskId interface and TaskMetadata.id() API that returns it

docs/connect.html

@@ -311,7 +311,7 @@ <h4><a id="connect_rest" href="#connect_rest">REST API</a></h4>
         <li><code>PUT /connectors/{name}/pause</code> - pause the connector and its tasks, which stops message processing until the connector is resumed</li>
         <li><code>PUT /connectors/{name}/resume</code> - resume a paused connector (or do nothing if the connector is not paused)</li>
         <li><code>POST /connectors/{name}/restart</code> - restart a connector (typically because it has failed)</li>
-        <li><code>POST /connectors/{name}/tasks/{taskId}/restart</code> - restart an individual task (typically because it has failed)</li>
+        <li><code>POST /connectors/{name}/tasks/{taskInfo}/restart</code> - restart an individual task (typically because it has failed)</li>
         <li><code>DELETE /connectors/{name}</code> - delete a connector, halting all tasks and deleting its configuration</li>
         <li><code>GET /connectors/{name}/topics</code> - get the set of topics that a specific connector is using since the connector was created or since a request to reset its set of active topics was issued</li>
         <li><code>PUT /connectors/{name}/topics/reset</code> - send a request to empty the set of active topics of a connector</li>

docs/streams/developer-guide/config-streams.html

@@ -406,8 +406,8 @@ <h4><a class="toc-backref" href="#id27">acceptable.recovery.lag</a><a class="hea
                                                                final Exception exception) {
 
                       log.warn("Exception caught during Deserialization, sending to the dead queue topic; " +
-                          "taskId: {}, topic: {}, partition: {}, offset: {}",
-                          context.taskId(), record.topic(), record.partition(), record.offset(),
+                          "taskInfo: {}, topic: {}, partition: {}, offset: {}",
+                          context.taskInfo(), record.topic(), record.partition(), record.offset(),
                           exception);
 
                       dlqProducer.send(new ProducerRecord&lt;&gt;(dlqTopic, record.timestamp(), record.key(), record.value(), record.headers())).get();

docs/streams/developer-guide/processor-api.html

@@ -384,7 +384,7 @@ <h2>
             <h2><a class="toc-backref" href="#id10">Accessing Processor Context</a><a class="headerlink" href="#accessing-processor-context" title="Permalink to this headline"></a></h2>
             <p>As we have mentioned in the <a href=#defining-a-stream-processor>Defining a Stream Processor</a> section, a <code>ProcessorContext</code> control the processing workflow, such as scheduling a punctuation function, and committing the current processed state.</p>
             <p>This object can also be used to access the metadata related with the application like
-            <code class="docutils literal"><span class="pre">applicationId</span></code>, <code class="docutils literal"><span class="pre">taskId</span></code>,
+            <code class="docutils literal"><span class="pre">applicationId</span></code>, <code class="docutils literal"><span class="pre">taskInfo</span></code>,
                 and <code class="docutils literal"><span class="pre">stateDir</span></code>, and also record related metadata as <code class="docutils literal"><span class="pre">topic</span></code>,
                 <code class="docutils literal"><span class="pre">partition</span></code>, <code class="docutils literal"><span class="pre">offset</span></code>, <code class="docutils literal"><span class="pre">timestamp</span></code> and
                 <code class="docutils literal"><span class="pre">headers</span></code>.</p>

docs/streams/upgrade-guide.html

@@ -984,7 +984,7 @@ <h3 class="anchor-heading"><a id="streams_api_changes_0110" class="anchor-link">
     <p> Producer's <code>client.id</code> naming schema: </p>
     <ul>
         <li> at-least-once (default): <code>[client.Id]-StreamThread-[sequence-number]</code> </li>
-        <li> exactly-once: <code>[client.Id]-StreamThread-[sequence-number]-[taskId]</code> </li>
+        <li> exactly-once: <code>[client.Id]-StreamThread-[sequence-number]-[taskInfo]</code> </li>
         <li> exactly-once-beta: <code>[client.Id]-StreamThread-[sequence-number]</code> </li>
     </ul>
     <p> <code>[client.Id]</code> is either set via Streams configuration parameter <code>client.id</code> or defaults to <code>[application.id]-[processId]</code> (<code>[processId]</code> is a random UUID). </p>

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

@@ -14,27 +14,19 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package org.apache.kafka.streams.processor.internals.namedtopology;
+package org.apache.kafka.streams;
 
-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;
-    }
+/**
+ * A wrapper class for basic task info
+ */
+public interface TaskInfo {
+    /**
+     * @return The ID of the subtopology that this task executes, ie the topicGroupId
+     */
+    int subtopologyId();
 
-    public static String namedTopology(final TaskId taskId) {
-        if (taskId instanceof NamedTaskId) {
-            return ((NamedTaskId) taskId).namedTopology();
-        } else {
-            return null;
-        }
-    }
+    /**
+     * @return The ID number of the partition.
+     */
+    int partition();
 }

streams/src/main/java/org/apache/kafka/streams/errors/LogAndContinueExceptionHandler.java

@@ -38,7 +38,7 @@ public DeserializationHandlerResponse handle(final ProcessorContext context,
 
         log.warn("Exception caught during Deserialization, " +
                  "taskId: {}, topic: {}, partition: {}, offset: {}",
-                 context.taskId(), record.topic(), record.partition(), record.offset(),
+                 context.taskInfo(), record.topic(), record.partition(), record.offset(),
                  exception);
 
         return DeserializationHandlerResponse.CONTINUE;

streams/src/main/java/org/apache/kafka/streams/errors/LogAndFailExceptionHandler.java

@@ -38,7 +38,7 @@ public DeserializationHandlerResponse handle(final ProcessorContext context,
 
         log.error("Exception caught during Deserialization, " +
                   "taskId: {}, topic: {}, partition: {}, offset: {}",
-                  context.taskId(), record.topic(), record.partition(), record.offset(),
+                  context.taskInfo(), record.topic(), record.partition(), record.offset(),
                   exception);
 
         return DeserializationHandlerResponse.FAIL;

streams/src/main/java/org/apache/kafka/streams/errors/TaskCorruptedException.java

@@ -17,7 +17,7 @@
 package org.apache.kafka.streams.errors;
 
 import org.apache.kafka.clients.consumer.InvalidOffsetException;
-import org.apache.kafka.streams.processor.TaskId;
+import org.apache.kafka.streams.processor.internals.TaskId;
 
 import java.util.Set;
 

streams/src/main/java/org/apache/kafka/streams/errors/TaskIdFormatException.java

@@ -17,8 +17,10 @@
 package org.apache.kafka.streams.errors;
 
 
+import org.apache.kafka.streams.processor.internals.TaskId;
+
 /**
- * Indicates a run time error incurred while trying parse the {@link org.apache.kafka.streams.processor.TaskId task id}
+ * Indicates a run time error incurred while trying parse the {@link TaskId task id}
  * from the read string.
  *
  * @see org.apache.kafka.streams.processor.internals.StreamTask

streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamAggregate.java

@@ -69,7 +69,7 @@ public void init(final ProcessorContext context) {
             super.init(context);
             droppedRecordsSensor = droppedRecordsSensorOrSkippedRecordsSensor(
                 Thread.currentThread().getName(),
-                context.taskId().toString(),
+                context.taskInfo().toString(),
                 (StreamsMetricsImpl) context.metrics());
             store = (TimestampedKeyValueStore<K, T>) context.getStateStore(storeName);
             tupleForwarder = new TimestampedTupleForwarder<>(

streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamKStreamJoin.java

@@ -91,7 +91,7 @@ private class KStreamKStreamJoinProcessor extends AbstractProcessor<K, V1> {
         public void init(final ProcessorContext context) {
             super.init(context);
             metrics = (StreamsMetricsImpl) context.metrics();
-            droppedRecordsSensor = droppedRecordsSensorOrSkippedRecordsSensor(Thread.currentThread().getName(), context.taskId().toString(), metrics);
+            droppedRecordsSensor = droppedRecordsSensorOrSkippedRecordsSensor(Thread.currentThread().getName(), context.taskInfo().toString(), metrics);
             otherWindowStore = context.getStateStore(otherWindowName);
 
             if (StreamsConfig.InternalConfig.getBoolean(context().appConfigs(), ENABLE_KSTREAMS_OUTER_JOIN_SPURIOUS_RESULTS_FIX, true)) {

streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamKTableJoinProcessor.java

@@ -52,7 +52,7 @@ class KStreamKTableJoinProcessor<K1, K2, V1, V2, R> extends AbstractProcessor<K1
     public void init(final ProcessorContext context) {
         super.init(context);
         metrics = (StreamsMetricsImpl) context.metrics();
-        droppedRecordsSensor = droppedRecordsSensorOrSkippedRecordsSensor(Thread.currentThread().getName(), context.taskId().toString(), metrics);
+        droppedRecordsSensor = droppedRecordsSensorOrSkippedRecordsSensor(Thread.currentThread().getName(), context.taskInfo().toString(), metrics);
         valueGetter.init(context);
     }
 

streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamReduce.java

@@ -65,7 +65,7 @@ private class KStreamReduceProcessor extends AbstractProcessor<K, V> {
         public void init(final ProcessorContext context) {
             super.init(context);
             metrics = (StreamsMetricsImpl) context.metrics();
-            droppedRecordsSensor = droppedRecordsSensorOrSkippedRecordsSensor(Thread.currentThread().getName(), context.taskId().toString(), metrics);
+            droppedRecordsSensor = droppedRecordsSensorOrSkippedRecordsSensor(Thread.currentThread().getName(), context.taskInfo().toString(), metrics);
             store = (TimestampedKeyValueStore<K, V>) context.getStateStore(storeName);
             tupleForwarder = new TimestampedTupleForwarder<>(
                 store,

streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamSessionWindowAggregate.java

@@ -97,11 +97,11 @@ public void init(final ProcessorContext context) {
             final String threadId = Thread.currentThread().getName();
             lateRecordDropSensor = droppedRecordsSensorOrLateRecordDropSensor(
                 threadId,
-                context.taskId().toString(),
+                context.taskInfo().toString(),
                 internalProcessorContext.currentNode().name(),
                 metrics
             );
-            droppedRecordsSensor = droppedRecordsSensorOrSkippedRecordsSensor(threadId, context.taskId().toString(), metrics);
+            droppedRecordsSensor = droppedRecordsSensorOrSkippedRecordsSensor(threadId, context.taskInfo().toString(), metrics);
             store = (SessionStore<K, Agg>) context.getStateStore(storeName);
             tupleForwarder = new SessionTupleForwarder<>(store, context, new SessionCacheFlushListener<>(context), sendOldValues);
         }

streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamSlidingWindowAggregate.java

@@ -94,11 +94,11 @@ public void init(final ProcessorContext context) {
             final String threadId = Thread.currentThread().getName();
             lateRecordDropSensor = droppedRecordsSensorOrLateRecordDropSensor(
                 threadId,
-                context.taskId().toString(),
+                context.taskInfo().toString(),
                 internalProcessorContext.currentNode().name(),
                 metrics
             );
-            droppedRecordsSensor = droppedRecordsSensorOrSkippedRecordsSensor(threadId, context.taskId().toString(), metrics);
+            droppedRecordsSensor = droppedRecordsSensorOrSkippedRecordsSensor(threadId, context.taskInfo().toString(), metrics);
             windowStore = (TimestampedWindowStore<K, Agg>) context.getStateStore(storeName);
             tupleForwarder = new TimestampedTupleForwarder<>(
                 windowStore,

streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamWindowAggregate.java

@@ -92,11 +92,11 @@ public void init(final ProcessorContext context) {
             final String threadId = Thread.currentThread().getName();
             lateRecordDropSensor = droppedRecordsSensorOrLateRecordDropSensor(
                 threadId,
-                context.taskId().toString(),
+                context.taskInfo().toString(),
                 internalProcessorContext.currentNode().name(),
                 metrics
             );
-            droppedRecordsSensor = droppedRecordsSensorOrSkippedRecordsSensor(threadId, context.taskId().toString(), metrics);
+            droppedRecordsSensor = droppedRecordsSensorOrSkippedRecordsSensor(threadId, context.taskInfo().toString(), metrics);
             windowStore = (TimestampedWindowStore<K, Agg>) context.getStateStore(storeName);
             tupleForwarder = new TimestampedTupleForwarder<>(
                 windowStore,

streams/src/main/java/org/apache/kafka/streams/kstream/internals/KTableKTableInnerJoin.java

@@ -78,7 +78,7 @@ private class KTableKTableJoinProcessor extends AbstractProcessor<K, Change<V1>>
         public void init(final ProcessorContext context) {
             super.init(context);
             metrics = (StreamsMetricsImpl) context.metrics();
-            droppedRecordsSensor = droppedRecordsSensorOrSkippedRecordsSensor(Thread.currentThread().getName(), context.taskId().toString(), metrics);
+            droppedRecordsSensor = droppedRecordsSensorOrSkippedRecordsSensor(Thread.currentThread().getName(), context.taskInfo().toString(), metrics);
             valueGetter.init(context);
         }
 

streams/src/main/java/org/apache/kafka/streams/kstream/internals/KTableKTableLeftJoin.java

@@ -77,7 +77,7 @@ private class KTableKTableLeftJoinProcessor extends AbstractProcessor<K, Change<
         public void init(final ProcessorContext context) {
             super.init(context);
             metrics = (StreamsMetricsImpl) context.metrics();
-            droppedRecordsSensor = droppedRecordsSensorOrSkippedRecordsSensor(Thread.currentThread().getName(), context.taskId().toString(), metrics);
+            droppedRecordsSensor = droppedRecordsSensorOrSkippedRecordsSensor(Thread.currentThread().getName(), context.taskInfo().toString(), metrics);
             valueGetter.init(context);
         }
 

streams/src/main/java/org/apache/kafka/streams/kstream/internals/KTableKTableOuterJoin.java

@@ -76,7 +76,7 @@ private class KTableKTableOuterJoinProcessor extends AbstractProcessor<K, Change
         public void init(final ProcessorContext context) {
             super.init(context);
             metrics = (StreamsMetricsImpl) context.metrics();
-            droppedRecordsSensor = droppedRecordsSensorOrSkippedRecordsSensor(Thread.currentThread().getName(), context.taskId().toString(), metrics);
+            droppedRecordsSensor = droppedRecordsSensorOrSkippedRecordsSensor(Thread.currentThread().getName(), context.taskInfo().toString(), metrics);
             valueGetter.init(context);
         }
 

streams/src/main/java/org/apache/kafka/streams/kstream/internals/KTableKTableRightJoin.java

@@ -75,7 +75,7 @@ private class KTableKTableRightJoinProcessor extends AbstractProcessor<K, Change
         public void init(final ProcessorContext context) {
             super.init(context);
             metrics = (StreamsMetricsImpl) context.metrics();
-            droppedRecordsSensor = droppedRecordsSensorOrSkippedRecordsSensor(Thread.currentThread().getName(), context.taskId().toString(), metrics);
+            droppedRecordsSensor = droppedRecordsSensorOrSkippedRecordsSensor(Thread.currentThread().getName(), context.taskInfo().toString(), metrics);
             valueGetter.init(context);
         }
 

streams/src/main/java/org/apache/kafka/streams/kstream/internals/KTableSource.java

@@ -84,7 +84,7 @@ private class KTableSourceProcessor extends AbstractProcessor<K, V> {
         public void init(final ProcessorContext context) {
             super.init(context);
             metrics = (StreamsMetricsImpl) context.metrics();
-            droppedRecordsSensor = droppedRecordsSensorOrSkippedRecordsSensor(Thread.currentThread().getName(), context.taskId().toString(), metrics);
+            droppedRecordsSensor = droppedRecordsSensorOrSkippedRecordsSensor(Thread.currentThread().getName(), context.taskInfo().toString(), metrics);
             if (queryableName != null) {
                 store = (TimestampedKeyValueStore<K, V>) context.getStateStore(queryableName);
                 tupleForwarder = new TimestampedTupleForwarder<>(

streams/src/main/java/org/apache/kafka/streams/kstream/internals/foreignkeyjoin/ForeignJoinSubscriptionProcessorSupplier.java

@@ -66,7 +66,7 @@ public void init(final ProcessorContext context) {
             final InternalProcessorContext internalProcessorContext = (InternalProcessorContext) context;
             droppedRecordsSensor = TaskMetrics.droppedRecordsSensorOrSkippedRecordsSensor(
                 Thread.currentThread().getName(),
-                internalProcessorContext.taskId().toString(),
+                internalProcessorContext.taskInfo().toString(),
                 internalProcessorContext.metrics()
             );
             store = internalProcessorContext.getStateStore(storeBuilder);

streams/src/main/java/org/apache/kafka/streams/kstream/internals/foreignkeyjoin/ForeignJoinSubscriptionSendProcessorSupplier.java

@@ -90,7 +90,7 @@ public void init(final ProcessorContext context) {
             }
             droppedRecordsSensor = TaskMetrics.droppedRecordsSensorOrSkippedRecordsSensor(
                 Thread.currentThread().getName(),
-                context.taskId().toString(),
+                context.taskInfo().toString(),
                 (StreamsMetricsImpl) context.metrics()
             );
         }

streams/src/main/java/org/apache/kafka/streams/kstream/internals/foreignkeyjoin/SubscriptionStoreReceiveProcessorSupplier.java

@@ -64,7 +64,7 @@ public void init(final ProcessorContext context) {
 
                 droppedRecordsSensor = TaskMetrics.droppedRecordsSensorOrSkippedRecordsSensor(
                     Thread.currentThread().getName(),
-                    internalProcessorContext.taskId().toString(),
+                    internalProcessorContext.taskInfo().toString(),
                     internalProcessorContext.metrics()
                 );
                 store = internalProcessorContext.getStateStore(storeBuilder);

streams/src/main/java/org/apache/kafka/streams/kstream/internals/suppress/KTableSuppressProcessorSupplier.java

@@ -144,7 +144,7 @@ public void init(final ProcessorContext context) {
             internalProcessorContext = (InternalProcessorContext) context;
             suppressionEmitSensor = ProcessorNodeMetrics.suppressionEmitSensor(
                 Thread.currentThread().getName(),
-                context.taskId().toString(),
+                context.taskInfo().toString(),
                 internalProcessorContext.currentNode().name(),
                 internalProcessorContext.metrics()
             );

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

@@ -20,6 +20,7 @@
 import org.apache.kafka.common.serialization.Serde;
 import org.apache.kafka.streams.StreamsBuilder;
 import org.apache.kafka.streams.StreamsMetrics;
+import org.apache.kafka.streams.TaskInfo;
 import org.apache.kafka.streams.Topology;
 import org.apache.kafka.streams.errors.StreamsException;
 
@@ -39,12 +40,21 @@ public interface ProcessorContext {
      */
     String applicationId();
 
+    /**
+     * Returns the task info.
+     *
+     * @return the task info
+     */
+    TaskInfo taskInfo();
+
     /**
      * Returns the task id.
      *
      * @return the task id
+     * @deprecated use {@link #taskInfo()} instead.
      */
-    TaskId taskId();
+    @Deprecated
+    org.apache.kafka.streams.processor.TaskId taskId();
 
     /**
      * Returns the default key serde.

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

@@ -18,6 +18,7 @@
 
 import org.apache.kafka.common.serialization.Serde;
 import org.apache.kafka.streams.StreamsMetrics;
+import org.apache.kafka.streams.TaskInfo;
 import org.apache.kafka.streams.errors.StreamsException;
 
 import java.io.File;
@@ -35,12 +36,21 @@ public interface StateStoreContext {
      */
     String applicationId();
 
+    /**
+     * Returns the task info.
+     *
+     * @return the task info
+     */
+    TaskInfo taskInfo();
+
     /**
      * Returns the task id.
      *
      * @return the task id
+     * @deprecated use {@link #taskInfo()} instead.
      */
-    TaskId taskId();
+    @Deprecated
+    org.apache.kafka.streams.processor.TaskId taskId();
 
     /**
      * Returns the default key serde.