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

snehashisp · 2024-11-29T19:57:07Z

The is one of a set of PRs for KIP-891. The list of total PRs given below all build one the previous one in the list. They can be reviewed individually, or if the complete set of changes is preferrable, please refer to the last PR.

This is PR#5 and contains updates to show the version of the running connector and task in the status endpoint. JMX task metrics have added connector, task and converter version. New metric groups are introduced for transformation.

KAFKA-18182: KIP-891 Connect Multiversion Support (Base PR with Plugin Loading Isolation Changes) #16984 - Plugins Loading Isolation Changes
KAFKA-18215: KIP-891 Connect Multiversioning Support (Configs and Validation changes for Connectors and Converters) #17741 - Versioned Configs and Validation for Connector and Converters
KAFKA-18419: KIP-891 Connect Multiversion Support (Transformation and Predicate Changes) #17742 - Transformation and Predicate Support
KAFKA-18863: Connect Multiversion Support (Versioned Connector Creation and related changes) #17743 - Versioned Connector Creation
KAFKA-18988: Connect Multiversion Support (Updates to status and metrics) #17988 - Updates to status and metrics

Committer Checklist (excluded from commit message)

Verify design and implementation
Verify test coverage and CI build status
Verify documentation (including upgrade notes)

…ates

snehashisp · 2025-03-09T13:08:39Z

Hi @gharris1727. Apologies for the delay in getting this PR updated. Please review when you get some time.

gharris1727

Thanks @snehashisp! Similar to the last PR, can you open a JIRA for this PR?

And are there any further PRs after this one? More code, system tests, documentation, etc?

gharris1727 · 2025-03-10T19:51:25Z

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;


Should this also be part of equals/hashCode?

Yes, it should be, thanks for catching this.

gharris1727 · 2025-03-10T19:54:15Z

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

+    public String connectorVersion(String connName) {
+        if (connectors.get(connName) == null) {
+            return null;
+        }
+        return connectors.get(connName).connectorVersion();
+    }
+
+    public String taskVersion(ConnectorTaskId taskId) {
+        if (tasks.get(taskId) == null) {
+            return null;
+        }
+        return tasks.get(taskId).taskVersion();
+    }
+


These are thread-unsafe and could cause NullPointerExceptions.

gharris1727 · 2025-03-10T20:00:16Z

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();


This is a call into the real task object, which requires a context classloader swap, can throw arbitary exceptions, and may block indefinitely. We should either use Plugins#pluginVersion to compute this, or compute it once during instantiation and then return it whenever necessary.

The same comment applies to WorkerConnector, WorkerSinkTask, and TaskPluginsMetadata.

Opted to compute it during initialization. Plugins don't have description entries for tasks, only connectors so fetching the task versions would be complicated and require us to somehow get the associated connector version. This may also not be accurate if the version overrides for connector and its task don't match.

gharris1727 · 2025-03-10T20:03:24Z

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

+    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) {


nit: This constructor is unused.

gharris1727 · 2025-03-10T20:04:32Z

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

+    public TaskStatus(ConnectorTaskId id, State state, String workerUrl, int generation, String trace, String version) {
+        super(id, state, workerUrl, generation, trace, version);
+    }
+
    public TaskStatus(ConnectorTaskId id, State state, String workerUrl, int generation, String trace) {


nit: This method is unused.

gharris1727 · 2025-03-10T20:33:18Z

...runtime/src/main/java/org/apache/kafka/connect/runtime/rest/entities/ConnectorStateInfo.java

@@ -87,14 +90,26 @@ public String workerId() {
        public String trace() {
            return trace;
        }
+
+        @JsonProperty
+        @JsonInclude(JsonInclude.Include.NON_EMPTY)


The other place in the REST API that surfaces the version is the PluginInfo object, which filters out "undefined". We should probably be consistent here.

That doesn't filter out null as far as i can tell, maybe it should be modified to filter out both null and undefined.

We can keep this consistent and add the custom filter here too. Wdyt of just showing the version (even if its null or undefined) with multi-versioning support. People who are unaware of the internals may be confused as to why its not showing up at all if the version is not set.

Wdyt of just showing the version (even if its null or undefined) with multi-versioning support.

I'm a bit ambivalent on explicit nulls, I think the more important thing to me is that "null" and "undefined" never appear in the results. That just looks like poor data quality to my eyes.

People who are unaware of the internals may be confused as to why its not showing up at all if the version is not set.

Can you elaborate? What is the situation where this happens and how might it mislead someone?

Not so much as misleading users, rather it might cause them to wonder why some plugins don't show any information on what version is installed. Say if they have a bunch of legacy plugins that don't properly set a version. Of course, if they look at the code they can make an educated guess and figure out that they don't implement Versioned and likewise, but it might be good to show null/undefined to make it more readily apparent. Either way is fine by me and we can mention the behavior we choose in the docs.

gharris1727 · 2025-03-10T20:48:44Z

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

+        this.connectorVersion = plugins.pluginVersion(connectorClass.getName(), connectorClass.getClassLoader(), PluginType.SINK, PluginType.SOURCE);
+        this.connectorType = getConnectorType(connectorClass, pluginLoaderSwapper);
+        this.taskClass = task.getClass().getName();
+        this.taskVersion = task.version();


This is a direct call to the task which would require a loader swap.

So this is being called during task build, after switching the context classloader to the connector loader in the calling methods.

gharris1727 · 2025-03-10T20:54:24Z

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

@@ -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);
+


nit: newline

gharris1727 · 2025-03-10T20:58:30Z

connect/runtime/src/main/java/org/apache/kafka/connect/runtime/isolation/PluginUtils.java

@@ -510,4 +512,16 @@ public static VersionRange connectorVersionRequirement(String version) throws In
        version = "[" + version + "]";
        return VersionRange.createFromVersionSpec(version);
    }
+
+    public static <T> String getVersionOrUndefined(T obj, Function<ClassLoader, LoaderSwap> pluginLoaderSwapper) {


IMHO this is an unhealthy method to add. Arbitrary code should not be deriving the version of a plugin from the object, the version should be known at the time the object is being instantiated and should be tracked in parallel.

Essentially, I believe the Versioned#version method should only be called during plugin discovery, and everything downstream from that should use whatever result was returned then.
And interestingly enough, this method doesn't return the right answer for general Task instances :)

Right, because task does not implement Versioned rather has explicitly defines it as a interface method. Anyway, I removed this as per suggestion and opted to used Plugins#PluginVersion instead.

gharris1727 · 2025-03-10T20:59:17Z

...runtime/src/main/java/org/apache/kafka/connect/runtime/rest/entities/ConnectorStateInfo.java

@@ -61,15 +61,18 @@ public ConnectorType type() {
        return type;
    }

+


nit: newline

snehashisp · 2025-03-12T16:32:06Z

Hi @gharris1727. Thanks for the review. Will address the comments as soon as I get some time. Also yes, there are more PRs which adds tests for the newly added code in #18325, #18326 and #18360. I raised them quite some time ago but have not updated them with the latest changes. Will address those once this PR is merged.

snehashisp · 2025-03-20T15:12:36Z

Hi @gharris1727. Please follow up on this review and the reviews for the subsequent test PRs, #18325, #18326 and #18360

snehashisp added 30 commits August 23, 2024 16:36

Add multiversioning apis to isolation

49b37c7

add method for header converter

1b740cc

init commit

3aba36e

update validation logic for connector versions

c96a8d1

add getter for all plugins for a class

017df62

Merge branch 'multiversioning' into mvn-validation

0f1cdaf

minor edits

855e77a

Allow super to load class but do a version check

4744f4f

Merge branch 'multiversioning' into mvn-validation

c708ade

validate the version when loaded from parent correctly

8beac31

Merge branch 'multiversioning' into mvn-validation

df16999

Tested connector multiple verions

6fb6471

revert changes to filestream sink

84ed0da

bug in loadclass, should use fullname

814ddb7

Merge branch 'multiversioning' into mvn-validation

7905708

add available version to exception and minor changes

7497ef7

Merge branch 'multiversioning' into mvn-validation

27aba58

validation supporting connector version

b9c0ab5

fix lf to cr

d9cad93

Add configs and recommendors

7df37cf

converter tested

cbbbc95

Add converter validatoins

5282a3f

Remove changes to json

7ad8a32

Validation and recommendors for versioning transformations and predic…

2cdf67c

…ates

update recommendation logic

53068f8

minor fixes

7676408

Add default version getter

2119a91

Merge branch 'multiversioning' into mvn-validation

eb3e693

Merge branch 'mvn-validation' into mvn-validation-transforms

9d83a1c

add defaults

dd150a5

snehashisp added 15 commits December 30, 2024 16:00

more changes

4a560fa

checkstyle fix

7b66579

update tests

8c140a0

Merge remote-tracking branch 'origin' into mvn-status

395a722

Merge remote-tracking branch 'ak/trunk' into mvn-runtime

f52032b

Merge branch 'trunk' of https://github.com/apache/kafka into mvn-runtime

51263ed

remove unnecessary changes

c942703

merge upstream

6b9c6cd

Remove old transformation method

ef24f30

Merge remote-tracking branch 'ak/trunk' into mvn-runtime

c5f614e

resolve comments

cccd5cd

Don't instantiate connector for tasks

45c4687

Merge branch 'mvn-runtime' into mvn-status

dfc074d

Merge branch 'trunk' of https://github.com/apache/kafka into mvn-status

6e1b8ee

Minor changes

82cebeb

gharris1727 reviewed Mar 10, 2025

View reviewed changes

snehashisp added 3 commits March 14, 2025 22:33

Initial Comments

3fe0613

address comments

336ed44

checkstyle fix

5a96b20

snehashisp changed the title ~~[WIP] KIP-891: Connect Multiversion Support (Updates to status and metrics)~~ KAFKA-18988: Connect Multiversion Support (Updates to status and metrics) Mar 14, 2025

snehashisp added 7 commits March 15, 2025 01:29

spotless fix

419947d

fix tests

899fbef

this-escape error

e8c7229

fix error

cf4318c

fix ordering error in conntransformation creation

66387cb

add plugins metadata to the constructor

d576ee9

update filter for version

5065bdf

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

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

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

snehashisp commented Nov 29, 2024 •

edited

Loading

snehashisp commented Mar 9, 2025

gharris1727 left a comment

gharris1727 Mar 10, 2025

snehashisp Mar 14, 2025

gharris1727 Mar 10, 2025

gharris1727 Mar 10, 2025

snehashisp Mar 14, 2025

gharris1727 Mar 10, 2025

gharris1727 Mar 10, 2025

gharris1727 Mar 10, 2025

snehashisp Mar 14, 2025 •

edited

Loading

gharris1727 Mar 14, 2025

snehashisp Mar 14, 2025

gharris1727 Mar 10, 2025

snehashisp Mar 14, 2025 •

edited

Loading

gharris1727 Mar 10, 2025

gharris1727 Mar 10, 2025

snehashisp Mar 14, 2025 •

edited

Loading

gharris1727 Mar 10, 2025

snehashisp commented Mar 12, 2025 •

edited

Loading

snehashisp commented Mar 20, 2025

		@@ -61,15 +61,18 @@ public ConnectorType type() {
		return type;
		}

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

Are you sure you want to change the base?

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

Conversation

snehashisp commented Nov 29, 2024 • edited Loading

Committer Checklist (excluded from commit message)

snehashisp commented Mar 9, 2025

gharris1727 left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

snehashisp Mar 14, 2025 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

snehashisp Mar 14, 2025 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

snehashisp Mar 14, 2025 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

snehashisp commented Mar 12, 2025 • edited Loading

snehashisp commented Mar 20, 2025

snehashisp commented Nov 29, 2024 •

edited

Loading

snehashisp Mar 14, 2025 •

edited

Loading

snehashisp Mar 14, 2025 •

edited

Loading

snehashisp Mar 14, 2025 •

edited

Loading

snehashisp commented Mar 12, 2025 •

edited

Loading