KAFKA-18988: Connect Multiversion Support (Updates to status and metrics)

connect/runtime/src/main/java/org/apache/kafka/connect/runtime/AbstractHerder.java

@@ -204,83 +204,91 @@ public boolean isReady() {
     @Override
     public void onStartup(String connector) {
         statusBackingStore.put(new ConnectorStatus(connector, ConnectorStatus.State.RUNNING,
-                workerId, generation()));
+                workerId, generation(), worker.connectorVersion(connector)));
     }
 
     @Override
     public void onStop(String connector) {
         statusBackingStore.put(new ConnectorStatus(connector, AbstractStatus.State.STOPPED,
-                workerId, generation()));
+                workerId, generation(), worker.connectorVersion(connector)));
     }
 
     @Override
     public void onPause(String connector) {
         statusBackingStore.put(new ConnectorStatus(connector, ConnectorStatus.State.PAUSED,
-                workerId, generation()));
+                workerId, generation(), worker.connectorVersion(connector)));
     }
 
     @Override
     public void onResume(String connector) {
         statusBackingStore.put(new ConnectorStatus(connector, TaskStatus.State.RUNNING,
-                workerId, generation()));
+                workerId, generation(), worker.connectorVersion(connector)));
     }
 
     @Override
     public void onShutdown(String connector) {
         statusBackingStore.putSafe(new ConnectorStatus(connector, ConnectorStatus.State.UNASSIGNED,
-                workerId, generation()));
+                workerId, generation(), worker.connectorVersion(connector)));
     }
 
     @Override
     public void onFailure(String connector, Throwable cause) {
         statusBackingStore.putSafe(new ConnectorStatus(connector, ConnectorStatus.State.FAILED,
-                trace(cause), workerId, generation()));
+                trace(cause), workerId, generation(), worker.connectorVersion(connector)));
     }
 
     @Override
     public void onStartup(ConnectorTaskId id) {
-        statusBackingStore.put(new TaskStatus(id, TaskStatus.State.RUNNING, workerId, generation()));
+        statusBackingStore.put(new TaskStatus(id, TaskStatus.State.RUNNING, workerId, generation(), null,
+                worker.taskVersion(id)));
     }
 
     @Override
     public void onFailure(ConnectorTaskId id, Throwable cause) {
-        statusBackingStore.putSafe(new TaskStatus(id, TaskStatus.State.FAILED, workerId, generation(), trace(cause)));
+        statusBackingStore.putSafe(new TaskStatus(id, TaskStatus.State.FAILED, workerId, generation(), trace(cause),
+                worker.taskVersion(id)));
     }
 
     @Override
     public void onShutdown(ConnectorTaskId id) {
-        statusBackingStore.putSafe(new TaskStatus(id, TaskStatus.State.UNASSIGNED, workerId, generation()));
+        statusBackingStore.putSafe(new TaskStatus(id, TaskStatus.State.UNASSIGNED, workerId, generation(), null,
+                worker.taskVersion(id)));
     }
 
     @Override
     public void onResume(ConnectorTaskId id) {
-        statusBackingStore.put(new TaskStatus(id, TaskStatus.State.RUNNING, workerId, generation()));
+        statusBackingStore.put(new TaskStatus(id, TaskStatus.State.RUNNING, workerId, generation(), null,
+                worker.taskVersion(id)));
     }
 
     @Override
     public void onPause(ConnectorTaskId id) {
-        statusBackingStore.put(new TaskStatus(id, TaskStatus.State.PAUSED, workerId, generation()));
+        statusBackingStore.put(new TaskStatus(id, TaskStatus.State.PAUSED, workerId, generation(), null,
+                worker.taskVersion(id)));
     }
 
     @Override
     public void onDeletion(String connector) {
         for (TaskStatus status : statusBackingStore.getAll(connector))
             onDeletion(status.id());
-        statusBackingStore.put(new ConnectorStatus(connector, ConnectorStatus.State.DESTROYED, workerId, generation()));
+        statusBackingStore.put(new ConnectorStatus(connector, ConnectorStatus.State.DESTROYED, workerId, generation(),
+                worker.connectorVersion(connector)));
     }
 
     @Override
     public void onDeletion(ConnectorTaskId id) {
-        statusBackingStore.put(new TaskStatus(id, TaskStatus.State.DESTROYED, workerId, generation()));
+        statusBackingStore.put(new TaskStatus(id, TaskStatus.State.DESTROYED, workerId, generation(), null,
+                worker.taskVersion(id)));
     }
 
     public void onRestart(String connector) {
         statusBackingStore.put(new ConnectorStatus(connector, ConnectorStatus.State.RESTARTING,
-                workerId, generation()));
+                workerId, generation(), worker.connectorVersion(connector)));
     }
 
     public void onRestart(ConnectorTaskId id) {
-        statusBackingStore.put(new TaskStatus(id, TaskStatus.State.RESTARTING, workerId, generation()));
+        statusBackingStore.put(new TaskStatus(id, TaskStatus.State.RESTARTING, workerId, generation(), null,
+                worker.taskVersion(id)));
     }
 
     @Override
@@ -348,12 +356,12 @@ public ConnectorStateInfo connectorStatus(String connName) {
         Collection<TaskStatus> tasks = statusBackingStore.getAll(connName);
 
         ConnectorStateInfo.ConnectorState connectorState = new ConnectorStateInfo.ConnectorState(
-                connector.state().toString(), connector.workerId(), connector.trace());
+                connector.state().toString(), connector.workerId(), connector.trace(), connector.version());
         List<ConnectorStateInfo.TaskState> taskStates = new ArrayList<>();
 
         for (TaskStatus status : tasks) {
             taskStates.add(new ConnectorStateInfo.TaskState(status.id().task(),
-                    status.state().toString(), status.workerId(), status.trace()));
+                    status.state().toString(), status.workerId(), status.trace(), status.version()));
         }
 
         Collections.sort(taskStates);
@@ -389,7 +397,7 @@ public ConnectorStateInfo.TaskState taskStatus(ConnectorTaskId id) {
             throw new NotFoundException("No status found for task " + id);
 
         return new ConnectorStateInfo.TaskState(id.task(), status.state().toString(),
-                status.workerId(), status.trace());
+                status.workerId(), status.trace(), status.version());
     }
 
     @Override
@@ -627,7 +635,8 @@ public Optional<RestartPlan> buildRestartPlan(RestartRequest request) {
         ConnectorStateInfo.ConnectorState connectorInfoState = new ConnectorStateInfo.ConnectorState(
                 connectorState.toString(),
                 connectorStatus.workerId(),
-                connectorStatus.trace()
+                connectorStatus.trace(),
+                connectorStatus.version()
         );
 
         // Collect the task states, If requested, mark the task as restarting
@@ -639,7 +648,8 @@ public Optional<RestartPlan> buildRestartPlan(RestartRequest request) {
                             taskStatus.id().task(),
                             taskState.toString(),
                             taskStatus.workerId(),
-                            taskStatus.trace()
+                            taskStatus.trace(),
+                            taskStatus.version()
                     );
                 })
                 .collect(Collectors.toList());

connect/runtime/src/main/java/org/apache/kafka/connect/runtime/AbstractStatus.java

@@ -34,18 +34,29 @@ public enum State {
     private final State state;
     private final String trace;
     private final String workerId;
+    private final String version;
     private final int generation;
 
     public AbstractStatus(T id,
                           State state,
                           String workerId,
                           int generation,
-                          String trace) {
+                          String trace,
+                          String version) {
         this.id = id;
         this.state = state;
         this.workerId = workerId;
         this.generation = generation;
         this.trace = trace;
+        this.version = version;
+    }
+
+    public AbstractStatus(T id,
+                          State state,
+                          String workerId,
+                          int generation,
+                          String trace) {
+        this(id, state, workerId, generation, trace, null);
     }
 
     public T id() {
@@ -68,12 +79,17 @@ public int generation() {
         return generation;
     }
 
+    public String version() {
+        return version;
+    }
+
     @Override
     public String toString() {
         return "Status{" +
                 "id=" + id +
                 ", state=" + state +
                 ", workerId='" + workerId + '\'' +
+                ", version='" + version + '\'' +
                 ", generation=" + generation +
                 '}';
     }

connect/runtime/src/main/java/org/apache/kafka/connect/runtime/AbstractWorkerSourceTask.java

@@ -391,6 +391,11 @@ public void execute() {
         finalOffsetCommit(false);
     }
 
+    @Override
+    public String taskVersion() {
+        return task.version();
+    }
+
     /**
      * Try to send a batch of records. If a send fails and is retriable, this saves the remainder of the batch so it can
      * be retried after backing off. If a send fails and is not retriable, this will throw a ConnectException.

connect/runtime/src/main/java/org/apache/kafka/connect/runtime/ConnectMetricsRegistry.java

@@ -37,6 +37,10 @@ public class ConnectMetricsRegistry {
     public static final String WORKER_GROUP_NAME = "connect-worker-metrics";
     public static final String WORKER_REBALANCE_GROUP_NAME = "connect-worker-rebalance-metrics";
     public static final String TASK_ERROR_HANDLING_GROUP_NAME = "task-error-metrics";
+    public static final String TRANSFORMS_GROUP = "connector-transform-metrics";
+    public static final String PREDICATES_GROUP = "connector-predicate-metrics";
+    public static final String TRANSFORM_TAG_NAME = "transform";
+    public static final String PREDICATE_TAG_NAME = "predicate";
 
     private final List<MetricNameTemplate> allTemplates = new ArrayList<>();
     public final MetricNameTemplate connectorStatus;
@@ -59,6 +63,17 @@ public class ConnectMetricsRegistry {
     public final MetricNameTemplate taskBatchSizeAvg;
     public final MetricNameTemplate taskCommitFailurePercentage;
     public final MetricNameTemplate taskCommitSuccessPercentage;
+    public final MetricNameTemplate taskConnectorClass;
+    public final MetricNameTemplate taskConnectorClassVersion;
+    public final MetricNameTemplate taskConnectorType;
+    public final MetricNameTemplate taskClass;
+    public final MetricNameTemplate taskVersion;
+    public final MetricNameTemplate taskKeyConverterClass;
+    public final MetricNameTemplate taskValueConverterClass;
+    public final MetricNameTemplate taskKeyConverterVersion;
+    public final MetricNameTemplate taskValueConverterVersion;
+    public final MetricNameTemplate taskHeaderConverterClass;
+    public final MetricNameTemplate taskHeaderConverterVersion;
     public final MetricNameTemplate sourceRecordPollRate;
     public final MetricNameTemplate sourceRecordPollTotal;
     public final MetricNameTemplate sourceRecordWriteRate;
@@ -115,6 +130,10 @@ public class ConnectMetricsRegistry {
     public final MetricNameTemplate transactionSizeMin;
     public final MetricNameTemplate transactionSizeMax;
     public final MetricNameTemplate transactionSizeAvg;
+    public final MetricNameTemplate transformClass;
+    public final MetricNameTemplate transformVersion;
+    public final MetricNameTemplate predicateClass;
+    public final MetricNameTemplate predicateVersion;
 
     public Map<MetricNameTemplate, TaskStatus.State> connectorStatusMetrics;
 
@@ -164,6 +183,43 @@ public ConnectMetricsRegistry(Set<String> tags) {
         taskCommitSuccessPercentage = createTemplate("offset-commit-success-percentage", TASK_GROUP_NAME,
                                                      "The average percentage of this task's offset commit attempts that succeeded.",
                                                      workerTaskTags);
+        taskConnectorClass = createTemplate("connector-class", TASK_GROUP_NAME, "The name of the connector class.", workerTaskTags);
+        taskConnectorClassVersion = createTemplate("connector-version", TASK_GROUP_NAME,
+                                                   "The version of the connector class, as reported by the connector.", workerTaskTags);
+        taskConnectorType = createTemplate("connector-type", TASK_GROUP_NAME, "The type of the connector. One of 'source' or 'sink'.",
+                                           workerTaskTags);
+        taskClass = createTemplate("task-class", TASK_GROUP_NAME, "The class name of the task.", workerTaskTags);
+        taskVersion = createTemplate("task-version", TASK_GROUP_NAME, "The version of the task.", workerTaskTags);
+        taskKeyConverterClass = createTemplate("key-converter-class", TASK_GROUP_NAME,
+                                            "The fully qualified class name from key.converter", workerTaskTags);
+        taskValueConverterClass = createTemplate("value-converter-class", TASK_GROUP_NAME,
+                                            "The fully qualified class name from value.converter", workerTaskTags);
+        taskKeyConverterVersion = createTemplate("key-converter-version", TASK_GROUP_NAME,
+                                            "The version instantiated for key.converter. May be undefined", workerTaskTags);
+        taskValueConverterVersion = createTemplate("value-converter-version", TASK_GROUP_NAME,
+                                                "The version instantiated for value.converter. May be undefined", workerTaskTags);
+        taskHeaderConverterClass = createTemplate("header-converter-class", TASK_GROUP_NAME,
+                                                "The fully qualified class name from header.converter", workerTaskTags);
+        taskHeaderConverterVersion = createTemplate("header-converter-version", TASK_GROUP_NAME,
+                                                    "The version instantiated for header.converter. May be undefined", workerTaskTags);
+
+        /* Transformation Metrics */
+        Set<String> transformTags = new LinkedHashSet<>(tags);
+        transformTags.addAll(workerTaskTags);
+        transformTags.add(TRANSFORM_TAG_NAME);
+        transformClass = createTemplate("transform-class", TRANSFORMS_GROUP,
+                "The class name of the transformation class", transformTags);
+        transformVersion = createTemplate("transform-version", TRANSFORMS_GROUP,
+                "The version of the transformation class", transformTags);
+
+        /* Predicate Metrics */
+        Set<String> predicateTags = new LinkedHashSet<>(tags);
+        predicateTags.addAll(workerTaskTags);
+        predicateTags.add(PREDICATE_TAG_NAME);
+        predicateClass = createTemplate("predicate-class", PREDICATES_GROUP,
+                "The class name of the predicate class", predicateTags);
+        predicateVersion = createTemplate("predicate-version", PREDICATES_GROUP,
+                "The version of the predicate class", predicateTags);
 
         /* Source worker task level */
         Set<String> sourceTaskTags = new LinkedHashSet<>(tags);
@@ -379,6 +435,7 @@ public ConnectMetricsRegistry(Set<String> tags) {
                 "The number of failed writes to the dead letter queue.", taskErrorHandlingTags);
         lastErrorTimestamp = createTemplate("last-error-timestamp", TASK_ERROR_HANDLING_GROUP_NAME,
                 "The epoch timestamp when this task last encountered an error.", taskErrorHandlingTags);
+
     }
 
     private MetricNameTemplate createTemplate(String name, String group, String doc, Set<String> tags) {
@@ -426,4 +483,20 @@ public String workerRebalanceGroupName() {
     public String taskErrorHandlingGroupName() {
         return TASK_ERROR_HANDLING_GROUP_NAME;
     }
+
+    public String transformsGroupName() {
+        return TRANSFORMS_GROUP;
+    }
+
+    public String transformsTagName() {
+        return TRANSFORM_TAG_NAME;
+    }
+
+    public String predicatesGroupName() {
+        return PREDICATES_GROUP;
+    }
+
+    public String predicateTagName() {
+        return PREDICATE_TAG_NAME;
+    }
 }

connect/runtime/src/main/java/org/apache/kafka/connect/runtime/ConnectorConfig.java

@@ -374,7 +374,7 @@ public <R extends ConnectRecord<R>> List<TransformationStage<R>> transformationS
                 final String versionConfig = prefix + WorkerConfig.PLUGIN_VERSION_SUFFIX;
                 @SuppressWarnings("unchecked") final Transformation<R> transformation = getTransformationOrPredicate(plugins, typeConfig, versionConfig);
                 Map<String, Object> configs = originalsWithPrefix(prefix);
-                Object predicateAlias = configs.remove(TransformationStage.PREDICATE_CONFIG);
+                String predicateAlias = (String) configs.remove(TransformationStage.PREDICATE_CONFIG);
                 Object negate = configs.remove(TransformationStage.NEGATE_CONFIG);
                 transformation.configure(configs);
                 Plugin<Transformation<R>> transformationPlugin = metrics.wrap(transformation, connectorTaskId, alias);
@@ -385,10 +385,10 @@ public <R extends ConnectRecord<R>> List<TransformationStage<R>> transformationS
                     @SuppressWarnings("unchecked")
                     Predicate<R> predicate = getTransformationOrPredicate(plugins, predicateTypeConfig, predicateVersionConfig);
                     predicate.configure(originalsWithPrefix(predicatePrefix));
-                    Plugin<Predicate<R>> predicatePlugin = metrics.wrap(predicate, connectorTaskId, (String) predicateAlias);
-                    transformations.add(new TransformationStage<>(predicatePlugin, negate != null && Boolean.parseBoolean(negate.toString()), transformationPlugin, plugins.safeLoaderSwapper()));
+                    Plugin<Predicate<R>> predicatePlugin = metrics.wrap(predicate, connectorTaskId, predicateAlias);
+                    transformations.add(new TransformationStage<>(predicatePlugin, predicateAlias, negate != null && Boolean.parseBoolean(negate.toString()), transformationPlugin, alias, plugins.safeLoaderSwapper()));
                 } else {
-                    transformations.add(new TransformationStage<>(transformationPlugin, plugins.safeLoaderSwapper()));
+                    transformations.add(new TransformationStage<>(transformationPlugin, alias, plugins.safeLoaderSwapper()));
                 }
             } catch (Exception e) {
                 throw new ConnectException(e);

connect/runtime/src/main/java/org/apache/kafka/connect/runtime/ConnectorStatus.java

@@ -19,10 +19,18 @@
 
 public class ConnectorStatus extends AbstractStatus<String> {
 
+    public ConnectorStatus(String connector, State state, String msg, String workerUrl, int generation, String version) {
+        super(connector, state, workerUrl, generation, msg, version);
+    }
+
     public ConnectorStatus(String connector, State state, String msg, String workerUrl, int generation) {
         super(connector, state, workerUrl, generation, msg);
     }
 
+    public ConnectorStatus(String connector, State state, String workerUrl, int generation, String version) {
+        super(connector, state, workerUrl, generation, null, version);
+    }
+
     public ConnectorStatus(String connector, State state, String workerUrl, int generation) {
         super(connector, state, workerUrl, generation, null);
     }