KAFKA-18215: KIP-891 Connect Multiversioning Support (Configs and Validation changes for Connectors and Converters) #17741

snehashisp · 2024-11-10T16:34:24Z

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#2 and contains changes to connector-config and connector validation endpoints to do appropriate validation considering the version provided and provide accurate defaults. This is primarily for connectors and converter PR#3 is for transformations.

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
[WIP] KIP-891: Connect Multiversion Support (Transformation and Predicate Changes) #17742 - Transformation and Predicate Support
[WIP] KIP-891: Connect Multiversion Support (Versioned Connector Creation and related changes) #17743 - Versioned Connector Creation
[WIP] KIP-891: 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)

gharris1727

@snehashisp Thanks for reworking the PluginVersionUtils. I think the current model is much more reasonable.

.gitignore

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

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

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

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

gharris1727 · 2024-12-08T18:34:04Z

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

@@ -98,6 +118,12 @@ public class ConnectorConfig extends AbstractConfig {
            new InstantiableClassValidator()


The KIP includes this detail:

If the .version property is left empty, or if there are no installed plugins with a matching name, errors will only be attributed to the plugin class property, not to the .version property.

If a plugin class name is valid (at least one version is installed) but the non-empty .version property doesn't include any of the installed versions, then the error will be attributed to both the .version property and the plugin class property.

I don't think the validators for the class or version configs are doing this right now. Is that part of this PR or a later PR?

I have addressed this for converters and transformations (PR # 3), but it's not that straightforward for connector and connector versions. The return on validation is a set of config Infos which is a combination of some predefined ones and ones defined in the connector. If the version is incorrect then the classloading of the connector itself will fail, so it is unclear as to which set of config infos we show. If we validate and show the config infos from the latest version of the connector, with the error attributed to the version config, the set of configs returned may not be accurate for the version requested.

One possible way would be to return the common ConnectorConfig config def, but that is not something the API does atm, for instance we could do the same thing for invalid connector class, but we just throw an exception right now, which is why I have kept it the same here.

I have addressed this for converters and transformations (PR # 3)

Do you mean transformations and predicates? I see that there's some validation happening there, but I didn't see anything similar for converters in this PR.

but it's not that straightforward for connector and connector versions

The KIP includes an exception for connector.class; we don't have to provide validation or recommendations for that.

We should provide validation results for connector.plugin.version, because that's going to be the single most useful configuration in this new feature.

Converter errors are added to the config infos here

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

Line 471 in b980c03

pluginConfigValue.addErrorMessage(e.getMessage());

.

I will see how we can return the validation result with the common connector configuration. This will have the config defs provided in ConnectorConfig but will not have the additional ones in the subclasses or the actual connector.

snehashisp · 2024-12-08T20:26:19Z

Thanks for the review @gharris1727. Will get to work on the soon, in my daytime.

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

gharris1727 · 2024-12-09T16:42:53Z

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

@@ -357,6 +357,10 @@ private <T> Set<PluginDesc<T>> pluginsOfClass(String classNameOrAlias, Set<Plugi
        return plugins;
    }

+    public PluginsRecommenders recommender() {


If we keep this as part of Plugins, then we can avoid the new PluginsRecommendors(plugins) call on every validate. Since the plugins information, once scanned remains static we don't really need to initialize all the recommenders again and again.

This is premature optimization, please don't do it. When someone has a benchmark that shows the benefit of this, it is trivial to implement.

The constructor of PluginsRecommenders does not perform any substantial pre-computation.
Validation already constructs hundreds of objects (ConfigValues, etc) each time it is called.

Also, if the plugins can be improved to be mutable in future it would be easier to revoke the current instance of recommenders in plugins and generate the new recommender, and have it reflected across all the uses.

YAGNI. When/if we get around to mutable Plugins, this class will look very different.

connect/runtime/src/test/java/org/apache/kafka/connect/runtime/SampleSinkConnector.java

gharris1727 · 2024-12-09T16:55:02Z

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

+            VersionRange range = PluginUtils.connectorVersionRequirement(connectorVersion);
+            ClassLoader connectorLoader = plugins.pluginLoader(connectorClass, range);
+            try (LoaderSwap loaderSwap = plugins.withClassLoader(connectorLoader)) {
+                T plugin = (T) plugins.newPlugin(pluginName, pluginClass, null);
+                if (plugin instanceof Versioned) {
+                    return ((Versioned) plugin).version();
+                }
+            }


If two different version of the same plugin is bundled under the same plugin classloader

This is impossible from Connect's perspective. Once a ClassLoader returns a Class, it is cached and cannot be redefined. If two versions of a plugin are both present within the same ClassLoader, one will be picked arbitrarily, and it will always be picked (on the local machine). Effectively, Connect cant even know the second class file is there, which is why we don't try and detect it.

snehashisp · 2024-12-11T02:21:11Z

Hi @gharris1727. Please take another pass. I have made the requested changes, lmk if I have missed anything. Some tests are failing, will look into those soon.

There is another unrelated issue I came across while testing this PR. There might be a regression in #16604, which seems to break reflections based isolation for plugins that are already defined in the classpath. Looks like the new classgraph scanner is always loading the classpath plugins even if there is a another version in an isolated plugin.path. This is causing the runtime to ignore the plugin due to this check. In my setup, I have built the following plugin hierarchy for testing, and I found that the different versions of json converter are never identified due to the one being present in the classpath.

plugins3
├── connect-file-1.2.1-T-0.9.0-P-3.1
│   └── lib
│       └── connect-file-1.2.1-T-0.9.0-P-3.1.jar
├── connect-file-2.2.1-T-1.0.0-P-3.1-extra
│   └── lib
│       ├── connect-file-2.2.1-T-1.0.0-P-3.1-extra.jar
│       └── connect-json-11.jar
├── connect-file-2.2.1-T-1.1.0-P-3.1
│   └── lib
│       └── connect-file-2.2.1-T-1.1.0-P-3.1.jar
├── connect-file-2.2.11-T-2.1.0-P-3.5
│   └── lib
│       └── connect-file-2.2.11-T-2.1.0-P-3.5.jar
├── json-11
│   └── lib
│       └── connect-json-11.jar
└── json-17
    └── libs
        └── connect-json-17.jar

I changed the debug logs to error in the class and observed the following errors during scanning.

[2024-12-11 07:29:28,968] ERROR class org.apache.kafka.connect.json.JsonConverter from other classloader jdk.internal.loader.ClassLoaders$AppClassLoader@c387f44 is visible from C:\Users\user\Desktop\confluent\testing\plugins3\connect-file-1.2.1-T-0.9.0-P-3.1, excluding to prevent isolated loading (org.apache.kafka.connect.runtime.isolation.ReflectionScanner:135)
[2024-12-11 07:29:28,969] ERROR class org.apache.kafka.connect.json.JsonConverter from other classloader jdk.internal.loader.ClassLoaders$AppClassLoader@c387f44 is visible from C:\Users\user\Desktop\confluent\testing\plugins3\connect-file-1.2.1-T-0.9.0-P-3.1, excluding to prevent isolated loading (org.apache.kafka.connect.runtime.isolation.ReflectionScanner:135)

The fix I made was to use overrideClassLoaders(source.loaders() instead of addClassLoader here. Based on this code, addClassLoader by default includes the jdk classloaders first and only then scans the provided loader. I think overriding the loader here is the correct approach, but have not dug too deep here. This is probably not the correct place but wanted to bring it to your attention and see if I am missing something. LMK if I should create a bug ticket for this.

gharris1727 · 2024-12-11T03:16:02Z

There is another unrelated issue I came across while testing this PR. There might be a regression in #16604

Great catch! That regression hasn't been released yet and would definitely be a blocker for the release. Please open a blocker ticket for that we can fix it.

The fix I made was to use overrideClassLoaders(source.loaders() instead of addClassLoader

I think this is the right fix. It looks like Classgraph by default expects parent-first loaders, while our loaders are child-first. When the same class is defined multiple times, it masks all later copies, which would hide all of the other versions of the plugins.

gharris1727

Please also fix the build/tests.

gharris1727 · 2024-12-11T03:21:24Z

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

@@ -1,149 +1,149 @@
-/*


nit: large diff.

How does this happen? I don't see what the change is...

It was happening when I was rolling back changes via IDE. I think it somehow adding carriage returns to the lines since its on windows. Does not normally happen If I do operation with WSL.

Ah yeah the line endings could cause a big diff. Thanks for looking into it!

.gitignore

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

snehashisp · 2024-12-11T11:59:57Z

Created https://issues.apache.org/jira/browse/KAFKA-18211 for the issue.

gharris1727

LGTM, thanks @snehashisp

chia7712 · 2024-12-15T02:34:28Z

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

@@ -82,6 +87,11 @@ public class ConnectorConfig extends AbstractConfig {
                    " or use \"FileStreamSink\" or \"FileStreamSinkConnector\" to make the configuration a bit shorter";
    private static final String CONNECTOR_CLASS_DISPLAY = "Connector class";

+    public static final String CONNECTOR_VERSION = "connector." + WorkerConfig.PLUGIN_VERSION_SUFFIX;


@snehashisp Thanks for this great patch. I have one small comment: It seems we are using connector.plugin.version instead of connector.version, correct? If so, could you please update the KIP accordingly?

…idation changes for Connectors and Converters) (apache#17741) Reviewers: Greg Harris <[email protected]>

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

Add default version getter

2119a91

Merge branch 'multiversioning' into mvn-validation

eb3e693

add a null check

f973fd0

Merge branch 'multiversioning' into mvn-validation

cce53be

Update log and comment

baea3cf

Merge branch 'multiversioning' into mvn-validation

5238dd0

adress backwards compatiblity

efbe73b

gharris1727 reviewed Dec 8, 2024

View reviewed changes

snehashisp added 2 commits December 9, 2024 01:30

fix tests for herders

47d4991

attempt to fix large diff in distributed herder

b0aee4c

snehashisp added 2 commits December 9, 2024 11:40

spotlessJavaFix

e0c3b16

updates based on comments

9a8fc1b

gharris1727 reviewed Dec 9, 2024

View reviewed changes

snehashisp added 7 commits December 10, 2024 01:13

remove plugins.recommendor, use header converter defaults

7efacbf

minor changes

b980c03

Do validations for invalid versions

871ae0a

Update plugin loading logic

f7d964d

test fix and minor adjustments

cba6924

bug fixes

520681c

revert unwanted changes

dc4f686

gharris1727 reviewed Dec 11, 2024

View reviewed changes

snehashisp added 7 commits December 11, 2024 23:11

add recommendor for invalid version, attempt to remove unwanted update

18b8d16

remove changes in connector config

978e84b

more unwanted change removal

b29b373

fix test

a0f2bd0

checkstyle fix and remove gitignore changes

72d86aa

remove unwanted changes

14cf633

newline in gitignore

357d461

gharris1727 approved these changes Dec 11, 2024

View reviewed changes

gharris1727 changed the title ~~[WIP] KIP-891: Connect Multiversioning Support (Configs and Validation changes for Connectors and Converters)~~ KAFKA-18215: KIP-891 Connect Multiversioning Support (Configs and Validation changes for Connectors and Converters) Dec 11, 2024

gharris1727 merged commit f4fe606 into apache:trunk Dec 11, 2024
11 checks passed

chia7712 reviewed Dec 15, 2024

View reviewed changes

peterxcli pushed a commit to peterxcli/kafka that referenced this pull request Dec 18, 2024

KAFKA-18215: KIP-891 Connect Multiversioning Support (Configs and Val…

fd6ccdb

…idation changes for Connectors and Converters) (apache#17741) Reviewers: Greg Harris <[email protected]>

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

KAFKA-18215: KIP-891 Connect Multiversioning Support (Configs and Validation changes for Connectors and Converters) #17741

KAFKA-18215: KIP-891 Connect Multiversioning Support (Configs and Validation changes for Connectors and Converters) #17741

snehashisp commented Nov 10, 2024 •

edited

Loading

gharris1727 left a comment

gharris1727 Dec 8, 2024

snehashisp Dec 9, 2024 •

edited

Loading

gharris1727 Dec 9, 2024 •

edited

Loading

snehashisp Dec 9, 2024 •

edited

Loading

snehashisp commented Dec 8, 2024

gharris1727 Dec 9, 2024

gharris1727 Dec 9, 2024

snehashisp commented Dec 11, 2024 •

edited

Loading

gharris1727 commented Dec 11, 2024

gharris1727 left a comment

gharris1727 Dec 11, 2024

snehashisp Dec 11, 2024

gharris1727 Dec 11, 2024

snehashisp commented Dec 11, 2024

gharris1727 left a comment

chia7712 Dec 15, 2024

		@@ -98,6 +118,12 @@ public class ConnectorConfig extends AbstractConfig {
		new InstantiableClassValidator()

KAFKA-18215: KIP-891 Connect Multiversioning Support (Configs and Validation changes for Connectors and Converters) #17741

KAFKA-18215: KIP-891 Connect Multiversioning Support (Configs and Validation changes for Connectors and Converters) #17741

Conversation

snehashisp commented Nov 10, 2024 • edited Loading

Committer Checklist (excluded from commit message)

gharris1727 left a comment

Choose a reason for hiding this comment

gharris1727 Dec 8, 2024

Choose a reason for hiding this comment

snehashisp Dec 9, 2024 • edited Loading

Choose a reason for hiding this comment

gharris1727 Dec 9, 2024 • edited Loading

Choose a reason for hiding this comment

snehashisp Dec 9, 2024 • edited Loading

Choose a reason for hiding this comment

snehashisp commented Dec 8, 2024

gharris1727 Dec 9, 2024

Choose a reason for hiding this comment

gharris1727 Dec 9, 2024

Choose a reason for hiding this comment

snehashisp commented Dec 11, 2024 • edited Loading

gharris1727 commented Dec 11, 2024

gharris1727 left a comment

Choose a reason for hiding this comment

gharris1727 Dec 11, 2024

Choose a reason for hiding this comment

snehashisp Dec 11, 2024

Choose a reason for hiding this comment

gharris1727 Dec 11, 2024

Choose a reason for hiding this comment

snehashisp commented Dec 11, 2024

gharris1727 left a comment

Choose a reason for hiding this comment

chia7712 Dec 15, 2024

Choose a reason for hiding this comment

snehashisp commented Nov 10, 2024 •

edited

Loading

snehashisp Dec 9, 2024 •

edited

Loading

gharris1727 Dec 9, 2024 •

edited

Loading

snehashisp Dec 9, 2024 •

edited

Loading

snehashisp commented Dec 11, 2024 •

edited

Loading