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

Reviewers: Greg Harris <[email protected]>

Author: snehashisp

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

@@ -203,83 +203,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
@@ -347,12 +355,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);
@@ -388,7 +396,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
@@ -626,7 +634,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
@@ -638,7 +647,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 +
                 '}';
     }
@@ -89,7 +105,8 @@ public boolean equals(Object o) {
                 && Objects.equals(id, that.id)
                 && state == that.state
                 && Objects.equals(trace, that.trace)
-                && Objects.equals(workerId, that.workerId);
+                && Objects.equals(workerId, that.workerId)
+                && Objects.equals(version, that.version);
     }
 
     @Override
@@ -98,6 +115,7 @@ public int hashCode() {
         result = 31 * result + (state != null ? state.hashCode() : 0);
         result = 31 * result + (trace != null ? trace.hashCode() : 0);
         result = 31 * result + (workerId != null ? workerId.hashCode() : 0);
+        result = 31 * result + (version != null ? version.hashCode() : 0);
         result = 31 * result + generation;
         return result;
     }

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

@@ -203,6 +203,7 @@ protected abstract void producerSendFailed(
     private final boolean topicTrackingEnabled;
     private final TopicCreation topicCreation;
     private final Executor closeExecutor;
+    private final String version;
 
     // Visible for testing
     List<SourceRecord> toSend;
@@ -236,11 +237,12 @@ protected AbstractWorkerSourceTask(ConnectorTaskId id,
                                        StatusBackingStore statusBackingStore,
                                        Executor closeExecutor,
                                        Supplier<List<ErrorReporter<SourceRecord>>> errorReportersSupplier,
+                                       TaskPluginsMetadata pluginsMetadata,
                                        Function<ClassLoader, LoaderSwap> pluginLoaderSwapper) {
 
         super(id, statusListener, initialState, loader, connectMetrics, errorMetrics,
                 retryWithToleranceOperator, transformationChain, errorReportersSupplier,
-                time, statusBackingStore, pluginLoaderSwapper);
+                time, statusBackingStore, pluginsMetadata, pluginLoaderSwapper);
 
         this.workerConfig = workerConfig;
         this.task = task;
@@ -258,6 +260,7 @@ protected AbstractWorkerSourceTask(ConnectorTaskId id,
         this.sourceTaskMetricsGroup = new SourceTaskMetricsGroup(id, connectMetrics);
         this.topicTrackingEnabled = workerConfig.getBoolean(TOPIC_TRACKING_ENABLE_CONFIG);
         this.topicCreation = TopicCreation.newTopicCreation(workerConfig, topicGroups);
+        this.version = task.version();
     }
 
     @Override
@@ -391,6 +394,11 @@ public void execute() {
         finalOffsetCommit(false);
     }
 
+    @Override
+    public String taskVersion() {
+        return 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);
@@ -426,4 +482,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;
+    }
 }