KAFKA-17150: Use Utils.loadClass instead of Class.forName to resolve aliases correctly

[gharris1727]

There are a handful of places where we call Class.forName in the code base. Previously, the ConfigDef call-site required a patch to resolve aliased classes (a feature in Connect). A second call-site via AbstractHerder -> Utils.newInstance has appeared as a regression in the current release. We can apply the same fix to all other call sites for Class.forName to prevent this from happening accidentally again. If aliases are not in-use, this should be a no-op change.

I added a PluginsTest case that exercises all of the ways we could construct Converter objects and ensures that they don't throw a ClassNotFoundException. This test failed before making the fix.

Committer Checklist (excluded from commit message)

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

[gharris1727]

Hi @C0urante @jlprat PTAL, thanks!

[gharris1727]

Also @yashmayya @mimaison and anyone else interested in reviewing!

[gharris1727]

Alternatives to this PR are:

• Revert KAFKA-13328, KAFKA-13329 (2): Add custom preflight validation support for connector header, key, and value converters #14309 -> viable but removes the new validation functionality
• Change the validateConverterConfig to call into plugins -> more involved, and risks breaking classloading-order-preference.

[C0urante]

LGTM pending CI to demonstrate that there are no (further) regressions. Great find, thanks Greg!

[chia7712]

@gharris1727 this patch is good to me. one small comment is left. PTAL

[chia7712]

Those comments about "alias" seem to valid for connector only, right? I mean the plugin-able configs in other module (core, storage, and tool) have no mechanism about "alias". If the answer is YES, maybe we can tweak the comments so as to avoid incorrect assumption that "all" plugin-able configs can use "alias".

[gharris1727]

Aliasing is a general concept, with one particular implementation in connect. I don't think the word alias is incorrect to use here. The comment also doesn't mention plugins at all.

You're right that the only time we use aliases is in connect, via the DelegatingClassLoader, so most of the time this change has no effect, loadedClass.getName().equals(klass) is true.

[chia7712]

I assumed the "alias" are applied to all plugin configs when reading those comments, so the origin comment is "we have some deprecated implementations having same alias as new one. Bababa..."

There is no strong reason to tweak the comments, because the root cause is probably my poor reading 😢

[chia7712]

LGTM and thanks for this good fix!

[jlprat]

I'm retriggering CI as 2 jobs failed in the first run.

[jlprat]

@gharris1727 The build wasn't working because of a spotless violation:

* What went wrong:

[2024-07-16T22:42:43.930Z] Execution failed for task ':connect:runtime:spotlessJavaCheck'.

[2024-07-16T22:42:43.930Z] > The following files had format violations:

[2024-07-16T22:42:43.930Z]       src/test/java/org/apache/kafka/connect/runtime/isolation/SynchronizationTest.java

[2024-07-16T22:42:43.930Z]           @@ -24,8 +24,8 @@

[2024-07-16T22:42:43.930Z]            import·org.apache.kafka.common.config.ConfigDef.Type;

[2024-07-16T22:42:43.930Z]            import·org.apache.kafka.common.utils.Utils;

[2024-07-16T22:42:43.930Z]            import·org.apache.kafka.connect.runtime.WorkerConfig;

[2024-07-16T22:42:43.930Z]           +import·org.apache.kafka.connect.storage.Converter;

[2024-07-16T22:42:43.930Z]            

[2024-07-16T22:42:43.930Z]           -import·org.apache.kafka.connect.storage.Converter;

[2024-07-16T22:42:43.930Z]            import·org.junit.jupiter.api.AfterEach;

[2024-07-16T22:42:43.930Z]            import·org.junit.jupiter.api.BeforeEach;

[2024-07-16T22:42:43.930Z]            import·org.junit.jupiter.api.Test;

[2024-07-16T22:42:43.930Z]   Run './gradlew :connect:runtime:spotlessApply' to fix these violations.

[C0urante]

@gharris1727 FYI, I've opened gharris1727#1 to add an integration test for plugin aliases. You are welcome to include it in this PR, or if you'd prefer I can file a PR against trunk to add the test instead.

[gharris1727]

@C0urante Thanks for implementing an integration test! I think one may be warranted here because it's regressed twice now, and an integration test would have caught this.

I'd prefer to unblock the release quickly, so please retarget that PR at upstream/trunk, thanks!

[gharris1727]

Test failures look unrelated, and the tests pass for me locally.