Skip to content

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

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
gharris1727 merged 3 commits into from
Jul 17, 2024
Merged

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

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
6b8ff28
KAFKA-17150: Use Utils.loadClass instead of Class.forName to resolve …
gharris1727 Jul 16, 2024
7652c9b
fixup: spotless
gharris1727 Jul 17, 2024
240fd64
fixup: scala type reference
gharris1727 Jul 17, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion clients/src/main/java/org/apache/kafka/clients/consumer/ConsumerPartitionAssignor.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -298,7 +298,7 @@ static List<ConsumerPartitionAssignor> getAssignorInstances(List<String> assigno
// first try to get the class if passed in as a string
if (klass instanceof String) {
try {
klass = Class.forName((String) klass, true, Utils.getContextOrKafkaClassLoader());
klass = Utils.loadClass((String) klass, Object.class);
} catch (ClassNotFoundException classNotFound) {
throw new KafkaException(klass + " ClassNotFoundException exception occurred", classNotFound);
}
Expand Down
9 changes: 1 addition & 8 deletions clients/src/main/java/org/apache/kafka/common/config/ConfigDef.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -766,14 +766,7 @@ else if (value instanceof String)
if (value instanceof Class)
return value;
else if (value instanceof String) {
ClassLoader contextOrKafkaClassLoader = Utils.getContextOrKafkaClassLoader();
// Use loadClass here instead of Class.forName because the name we use here may be an alias
// and not match the name of the class that gets loaded. If that happens, Class.forName can
// throw an exception.
Class<?> klass = contextOrKafkaClassLoader.loadClass(trimmed);
// Invoke forName here with the true name of the requested class to cause class
// initialization to take place.
return Class.forName(klass.getName(), true, contextOrKafkaClassLoader);
return Utils.loadClass(trimmed, Object.class);
} else
throw new ConfigException(name, value, "Expected a Class instance or class name.");
default:
Expand Down
11 changes: 9 additions & 2 deletions clients/src/main/java/org/apache/kafka/common/utils/Utils.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -425,7 +425,14 @@ public static <T> T newInstance(String klass, Class<T> base) throws ClassNotFoun
* @return the new class
*/
public static <T> Class<? extends T> loadClass(String klass, Class<T> base) throws ClassNotFoundException {
return Class.forName(klass, true, Utils.getContextOrKafkaClassLoader()).asSubclass(base);
ClassLoader contextOrKafkaClassLoader = Utils.getContextOrKafkaClassLoader();
// Use loadClass here instead of Class.forName because the name we use here may be an alias

@chia7712 chia7712 Jul 17, 2024

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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 gharris1727 Jul 17, 2024

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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 chia7712 Jul 17, 2024

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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 😢

// and not match the name of the class that gets loaded. If that happens, Class.forName can
// throw an exception.
Class<?> loadedClass = contextOrKafkaClassLoader.loadClass(klass);
// Invoke forName here with the true name of the requested class to cause class
// initialization to take place.
return Class.forName(loadedClass.getName(), true, contextOrKafkaClassLoader).asSubclass(base);
}

/**
Expand Down Expand Up @@ -454,7 +461,7 @@ public static <T> T newParameterizedInstance(String className, Object... params)
Class<?>[] argTypes = new Class<?>[params.length / 2];
Object[] args = new Object[params.length / 2];
try {
Class<?> c = Class.forName(className, true, Utils.getContextOrKafkaClassLoader());
Class<?> c = Utils.loadClass(className, Object.class);
for (int i = 0; i < params.length / 2; i++) {
argTypes[i] = (Class<?>) params[2 * i];
args[i] = params[(2 * i) + 1];
Expand Down
20 changes: 20 additions & 0 deletions connect/runtime/src/test/java/org/apache/kafka/connect/runtime/isolation/PluginsTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -23,6 +23,7 @@
import org.apache.kafka.common.config.ConfigException;
import org.apache.kafka.common.config.provider.ConfigProvider;
import org.apache.kafka.common.utils.LogCaptureAppender;
import org.apache.kafka.common.utils.Utils;
import org.apache.kafka.connect.components.Versioned;
import org.apache.kafka.connect.connector.Connector;
import org.apache.kafka.connect.connector.policy.AllConnectorClientConfigOverridePolicy;
Expand Down Expand Up @@ -596,6 +597,25 @@ public void testServiceLoadWithPlugins() {
}
}

@Test
public void testAliasesInConverters() throws ClassNotFoundException {
ClassLoader connectorLoader = plugins.connectorLoader(TestPlugin.SAMPLING_CONNECTOR.className());
try (LoaderSwap loaderSwap = plugins.withClassLoader(connectorLoader)) {
String configKey = "config.key";
String alias = "SamplingConverter";
assertTrue(TestPlugin.SAMPLING_CONVERTER.className().contains(alias));
ConfigDef def = new ConfigDef().define(configKey, ConfigDef.Type.CLASS, ConfigDef.Importance.HIGH, "docstring");
AbstractConfig config = new AbstractConfig(def, Collections.singletonMap(configKey, alias));

assertNotNull(config.getClass(configKey));
assertNotNull(config.getConfiguredInstance(configKey, Converter.class));
assertNotNull(plugins.newConverter(config, configKey, ClassLoaderUsage.CURRENT_CLASSLOADER));
assertNotNull(plugins.newConverter(config, configKey, ClassLoaderUsage.PLUGINS));

assertNotNull(Utils.newInstance(alias, Converter.class));
}
}

private void assertClassLoaderReadsVersionFromResource(
TestPlugin parentResource, TestPlugin childResource, String className, String... expectedVersions) {
URL[] systemPath = TestPlugins.pluginPath(parentResource)
Expand Down
4 changes: 3 additions & 1 deletion ...runtime/src/test/java/org/apache/kafka/connect/runtime/isolation/SynchronizationTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -22,7 +22,9 @@
import org.apache.kafka.common.config.ConfigDef;
import org.apache.kafka.common.config.ConfigDef.Importance;
import org.apache.kafka.common.config.ConfigDef.Type;
import org.apache.kafka.common.utils.Utils;
import org.apache.kafka.connect.runtime.WorkerConfig;
import org.apache.kafka.connect.storage.Converter;
Comment thread
jlprat marked this conversation as resolved.

import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
Expand Down Expand Up @@ -328,7 +330,7 @@ public void testPluginClassLoaderDoesntHoldMonitorLock()
synchronized (externalTestLock) {
try {
progress.await(null);
Class.forName(TestPlugins.TestPlugin.SAMPLING_CONVERTER.className(), true, connectorLoader);
Utils.loadClass(TestPlugins.TestPlugin.SAMPLING_CONVERTER.className(), Converter.class);
} catch (ClassNotFoundException e) {
throw new RuntimeException("Failed to load test plugin", e);
}
Expand Down
2 changes: 1 addition & 1 deletion core/src/main/scala/kafka/utils/CoreUtils.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -114,7 +114,7 @@ object CoreUtils {
* Create an instance of the class with the given class name
*/
def createObject[T <: AnyRef](className: String, args: AnyRef*): T = {
val klass = Class.forName(className, true, Utils.getContextOrKafkaClassLoader).asInstanceOf[Class[T]]
val klass = Utils.loadClass(className, classOf[Object]).asInstanceOf[Class[T]]
val constructor = klass.getConstructor(args.map(_.getClass): _*)
constructor.newInstance(args: _*)
}
Expand Down