You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "majialoong (via GitHub)" <gi...@apache.org> on 2025/11/15 10:55:47 UTC
[PR] MINOR: Revert timing change for creating connect config [kafka]
majialoong opened a new pull request, #20891:
URL: https://github.com/apache/kafka/pull/20891
[This PR](https://github.com/apache/kafka/pull/20612) adjusted the creation order of Connect configurations to ensure that the `plugin.path` config was validated before use.
However, this change caused issues loading classes that only exist in the `plugin.path` path. This PR reverts the previous changes and adds additional unit tests to prevent this issue from recurring.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] MINOR: Revert timing change for creating connect config [kafka]
Posted by "majialoong (via GitHub)" <gi...@apache.org>.
majialoong commented on code in PR #20891:
URL: https://github.com/apache/kafka/pull/20891#discussion_r2529816374
##########
connect/runtime/src/main/java/org/apache/kafka/connect/runtime/isolation/PluginUtils.java:
##########
@@ -209,6 +209,9 @@ public static Set<Path> pluginLocations(String pluginPath, boolean failFast) {
for (String path : pluginPathElements) {
try {
Path pluginPathElement = Paths.get(path).toAbsolutePath();
+ if (pluginPath.isEmpty()) {
+ log.warn("Plugin path element is empty, evaluating to {}.", pluginPathElement);
Review Comment:
This part of the code is also related to recovery, because empty states can still occur here.
However, I had a question: why does it only check whether plugin.path is empty, instead of checking the elements within plugin.path? The logs mention "plugin path element" as a keyword, which is what I initially wanted to address. Wouldn't it be better to check whether each path element is empty instead?
@gharris1727 If you have time, could you please help me understand this behavior? Thank you very much for your help.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] MINOR: Revert timing change for creating connect config [kafka]
Posted by "majialoong (via GitHub)" <gi...@apache.org>.
majialoong commented on code in PR #20891:
URL: https://github.com/apache/kafka/pull/20891#discussion_r2529816374
##########
connect/runtime/src/main/java/org/apache/kafka/connect/runtime/isolation/PluginUtils.java:
##########
@@ -209,6 +209,9 @@ public static Set<Path> pluginLocations(String pluginPath, boolean failFast) {
for (String path : pluginPathElements) {
try {
Path pluginPathElement = Paths.get(path).toAbsolutePath();
+ if (pluginPath.isEmpty()) {
+ log.warn("Plugin path element is empty, evaluating to {}.", pluginPathElement);
Review Comment:
This part of the code is also related to recovery, because empty states can still occur here.
I have a question about this part: why does it only check whether `plugin.path` is empty, instead of checking the elements within `plugin.path`? The logs mention "plugin path element", which is what I originally wanted to address. Would it make sense to check whether each path element is empty instead?
@gharris1727 If you have time, could you please help me understand the reasoning behind this behavior? Thank you very much for your help.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] MINOR: Revert timing change for creating connect config [kafka]
Posted by "majialoong (via GitHub)" <gi...@apache.org>.
majialoong commented on code in PR #20891:
URL: https://github.com/apache/kafka/pull/20891#discussion_r2546692360
##########
connect/runtime/src/test/java/org/apache/kafka/connect/cli/AbstractConnectCliTest.java:
##########
@@ -0,0 +1,174 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.connect.cli;
+
+import org.apache.kafka.common.config.ConfigException;
+import org.apache.kafka.connect.connector.policy.ConnectorClientConfigOverridePolicy;
+import org.apache.kafka.connect.runtime.Herder;
+import org.apache.kafka.connect.runtime.WorkerConfig;
+import org.apache.kafka.connect.runtime.distributed.DistributedConfig;
+import org.apache.kafka.connect.runtime.isolation.Plugins;
+import org.apache.kafka.connect.runtime.isolation.TestPlugins;
+import org.apache.kafka.connect.runtime.rest.ConnectRestServer;
+import org.apache.kafka.connect.runtime.rest.RestClient;
+import org.apache.kafka.connect.runtime.rest.RestServer;
+
+import org.junit.jupiter.api.Test;
+import org.mockito.MockedConstruction;
+
+import java.net.URI;
+import java.nio.file.Path;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Set;
+
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.mockito.Mockito.doNothing;
+import static org.mockito.Mockito.doReturn;
+import static org.mockito.Mockito.mockConstruction;
+import static org.mockito.Mockito.spy;
+
+public class AbstractConnectCliTest {
+
+ /**
+ * Verifies that createConfig is called after compareAndSwapWithDelegatingLoader in startConnect.
+ * If the order is wrong, ConfigProvider classes in plugin.path cannot be loaded.
+ */
+ @Test
+ public void testStartConnectEnforcesCorrectOrder() {
+ ClassLoader originalTCCL = Thread.currentThread().getContextClassLoader();
+
+ try {
+ // Create worker props with ConfigProvider that's only in plugin.path
+ Set<Path> pluginPaths = TestPlugins.pluginPath(TestPlugins.TestPlugin.SAMPLING_CONFIG_PROVIDER);
+ String pluginPath = String.join(",", pluginPaths.stream().map(Path::toString).toList());
+
+ Map<String, String> workerProps = createBaseWorkerProps(pluginPath);
+ workerProps.put(WorkerConfig.CONFIG_PROVIDERS_CONFIG, "testProvider");
+ String providerClassName = TestPlugins.TestPlugin.SAMPLING_CONFIG_PROVIDER.className();
+ workerProps.put(WorkerConfig.CONFIG_PROVIDERS_CONFIG + ".testProvider.class", providerClassName);
+
+ // Use a restricted classloader that cannot find the ConfigProvider class
+ ClassLoader restrictedClassLoader = new RestrictedClassLoader(providerClassName);
+ Thread.currentThread().setContextClassLoader(restrictedClassLoader);
+
+ // Verify the restricted classloader cannot load the ConfigProvider class
+ assertThrows(ClassNotFoundException.class, () ->
+ restrictedClassLoader.loadClass(providerClassName));
+
+ // Config creation should fail when ConfigProvider class cannot be loaded
+ assertThrows(ConfigException.class, () -> new DistributedConfig(workerProps));
+
+ // Call startConnect and verify the order is correct
+ TestConnectCli testConnectCli = new TestConnectCli();
+
+ // Mock ConnectRestServer to avoid actual server initialization
+ try (MockedConstruction<ConnectRestServer> mockRestServer = mockConstruction(
+ ConnectRestServer.class,
+ (mock, context) -> {
+ doReturn(URI.create("http://localhost:8083")).when(mock).advertisedUrl();
+ doNothing().when(mock).initializeServer();
+ })) {
+
+ // If order is correct, createConfig succeeds and we reach createHerder which throws ExpectedException
+ assertThrows(ExpectedException.class, () -> testConnectCli.startConnect(workerProps));
+ }
+ } finally {
+ Thread.currentThread().setContextClassLoader(originalTCCL);
+ }
+ }
+
+ private Map<String, String> createBaseWorkerProps(String pluginPath) {
+ Map<String, String> props = new HashMap<>();
+ props.put(WorkerConfig.PLUGIN_PATH_CONFIG, pluginPath);
+ props.put("bootstrap.servers", "localhost:9092");
+ props.put(DistributedConfig.GROUP_ID_CONFIG, "test-connect-cluster");
+ props.put(DistributedConfig.CONFIG_TOPIC_CONFIG, "connect-configs");
+ props.put(DistributedConfig.OFFSET_STORAGE_TOPIC_CONFIG, "connect-offsets");
+ props.put(DistributedConfig.STATUS_STORAGE_TOPIC_CONFIG, "connect-status");
+ props.put(WorkerConfig.KEY_CONVERTER_CLASS_CONFIG, "org.apache.kafka.connect.json.JsonConverter");
+ props.put(WorkerConfig.VALUE_CONVERTER_CLASS_CONFIG, "org.apache.kafka.connect.json.JsonConverter");
+ return props;
+ }
+
+ /**
+ * Test implementation that calls the parent's startConnect to verify correct order.
+ */
+ private static class TestConnectCli extends AbstractConnectCli<Herder, DistributedConfig> {
+ TestConnectCli() {
+ super();
+ }
+
+ @Override
+ protected String usage() {
+ return "test";
+ }
+
+ @Override
+ protected Herder createHerder(DistributedConfig config, String workerId, Plugins plugins,
+ ConnectorClientConfigOverridePolicy connectorClientConfigOverridePolicy,
+ RestServer restServer,
+ RestClient restClient) {
+ // Reaching createHerder means createConfig succeeded, indicating correct order was maintained
+ throw new ExpectedException();
+ }
+
+ @Override
+ protected DistributedConfig createConfig(Map<String, String> workerProps) {
+ DistributedConfig config = new DistributedConfig(workerProps);
+ // Mock kafkaClusterId() to avoid connecting to Kafka broker
+ DistributedConfig spyConfig = spy(config);
+ doReturn("test-cluster-id").when(spyConfig).kafkaClusterId();
+ return spyConfig;
+ }
+ }
+
+ /**
+ * ClassLoader that cannot load a specific class, simulating plugin classes only in plugin.path.
+ */
+ private static class RestrictedClassLoader extends ClassLoader {
+ private final String restrictedClassName;
+ private final ClassLoader systemLoader;
+
+ RestrictedClassLoader(String restrictedClassName) {
+ super(null); // No parent to prevent delegation
+ this.restrictedClassName = restrictedClassName;
+ this.systemLoader = ClassLoader.getSystemClassLoader();
+ }
+
+ @Override
+ protected Class<?> loadClass(String name, boolean resolve) throws ClassNotFoundException {
+ // Block the restricted class and test plugin classes
+ if (name.equals(restrictedClassName)
+ || (name.startsWith("test.plugins") && name.contains("Sampling"))) {
Review Comment:
Thanks for the review. Yes, you are right.
The additional check was defensive, but this test only needs to ensure the ConfigProvider class declared in the configuration cannot be loaded by the current TCCL. I removed the extra logic and kept only the class name check.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] MINOR: Revert timing change for creating connect config [kafka]
Posted by "majialoong (via GitHub)" <gi...@apache.org>.
majialoong commented on PR #20891:
URL: https://github.com/apache/kafka/pull/20891#issuecomment-3536353622
@chia7712 @gharris1727 When you have some free time, could you please review this PR, Thanks!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] MINOR: Revert timing change for creating connect config [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 merged PR #20891:
URL: https://github.com/apache/kafka/pull/20891
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] MINOR: Revert timing change for creating connect config [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #20891:
URL: https://github.com/apache/kafka/pull/20891#discussion_r2545950418
##########
connect/runtime/src/test/java/org/apache/kafka/connect/cli/AbstractConnectCliTest.java:
##########
@@ -0,0 +1,174 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.connect.cli;
+
+import org.apache.kafka.common.config.ConfigException;
+import org.apache.kafka.connect.connector.policy.ConnectorClientConfigOverridePolicy;
+import org.apache.kafka.connect.runtime.Herder;
+import org.apache.kafka.connect.runtime.WorkerConfig;
+import org.apache.kafka.connect.runtime.distributed.DistributedConfig;
+import org.apache.kafka.connect.runtime.isolation.Plugins;
+import org.apache.kafka.connect.runtime.isolation.TestPlugins;
+import org.apache.kafka.connect.runtime.rest.ConnectRestServer;
+import org.apache.kafka.connect.runtime.rest.RestClient;
+import org.apache.kafka.connect.runtime.rest.RestServer;
+
+import org.junit.jupiter.api.Test;
+import org.mockito.MockedConstruction;
+
+import java.net.URI;
+import java.nio.file.Path;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Set;
+
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.mockito.Mockito.doNothing;
+import static org.mockito.Mockito.doReturn;
+import static org.mockito.Mockito.mockConstruction;
+import static org.mockito.Mockito.spy;
+
+public class AbstractConnectCliTest {
+
+ /**
+ * Verifies that createConfig is called after compareAndSwapWithDelegatingLoader in startConnect.
+ * If the order is wrong, ConfigProvider classes in plugin.path cannot be loaded.
+ */
+ @Test
+ public void testStartConnectEnforcesCorrectOrder() {
+ ClassLoader originalTCCL = Thread.currentThread().getContextClassLoader();
+
+ try {
+ // Create worker props with ConfigProvider that's only in plugin.path
+ Set<Path> pluginPaths = TestPlugins.pluginPath(TestPlugins.TestPlugin.SAMPLING_CONFIG_PROVIDER);
+ String pluginPath = String.join(",", pluginPaths.stream().map(Path::toString).toList());
+
+ Map<String, String> workerProps = createBaseWorkerProps(pluginPath);
+ workerProps.put(WorkerConfig.CONFIG_PROVIDERS_CONFIG, "testProvider");
+ String providerClassName = TestPlugins.TestPlugin.SAMPLING_CONFIG_PROVIDER.className();
+ workerProps.put(WorkerConfig.CONFIG_PROVIDERS_CONFIG + ".testProvider.class", providerClassName);
+
+ // Use a restricted classloader that cannot find the ConfigProvider class
+ ClassLoader restrictedClassLoader = new RestrictedClassLoader(providerClassName);
+ Thread.currentThread().setContextClassLoader(restrictedClassLoader);
+
+ // Verify the restricted classloader cannot load the ConfigProvider class
+ assertThrows(ClassNotFoundException.class, () ->
+ restrictedClassLoader.loadClass(providerClassName));
+
+ // Config creation should fail when ConfigProvider class cannot be loaded
+ assertThrows(ConfigException.class, () -> new DistributedConfig(workerProps));
+
+ // Call startConnect and verify the order is correct
+ TestConnectCli testConnectCli = new TestConnectCli();
+
+ // Mock ConnectRestServer to avoid actual server initialization
+ try (MockedConstruction<ConnectRestServer> mockRestServer = mockConstruction(
+ ConnectRestServer.class,
+ (mock, context) -> {
+ doReturn(URI.create("http://localhost:8083")).when(mock).advertisedUrl();
+ doNothing().when(mock).initializeServer();
+ })) {
+
+ // If order is correct, createConfig succeeds and we reach createHerder which throws ExpectedException
+ assertThrows(ExpectedException.class, () -> testConnectCli.startConnect(workerProps));
+ }
+ } finally {
+ Thread.currentThread().setContextClassLoader(originalTCCL);
+ }
+ }
+
+ private Map<String, String> createBaseWorkerProps(String pluginPath) {
+ Map<String, String> props = new HashMap<>();
+ props.put(WorkerConfig.PLUGIN_PATH_CONFIG, pluginPath);
+ props.put("bootstrap.servers", "localhost:9092");
+ props.put(DistributedConfig.GROUP_ID_CONFIG, "test-connect-cluster");
+ props.put(DistributedConfig.CONFIG_TOPIC_CONFIG, "connect-configs");
+ props.put(DistributedConfig.OFFSET_STORAGE_TOPIC_CONFIG, "connect-offsets");
+ props.put(DistributedConfig.STATUS_STORAGE_TOPIC_CONFIG, "connect-status");
+ props.put(WorkerConfig.KEY_CONVERTER_CLASS_CONFIG, "org.apache.kafka.connect.json.JsonConverter");
+ props.put(WorkerConfig.VALUE_CONVERTER_CLASS_CONFIG, "org.apache.kafka.connect.json.JsonConverter");
+ return props;
+ }
+
+ /**
+ * Test implementation that calls the parent's startConnect to verify correct order.
+ */
+ private static class TestConnectCli extends AbstractConnectCli<Herder, DistributedConfig> {
+ TestConnectCli() {
+ super();
+ }
+
+ @Override
+ protected String usage() {
+ return "test";
+ }
+
+ @Override
+ protected Herder createHerder(DistributedConfig config, String workerId, Plugins plugins,
+ ConnectorClientConfigOverridePolicy connectorClientConfigOverridePolicy,
+ RestServer restServer,
+ RestClient restClient) {
+ // Reaching createHerder means createConfig succeeded, indicating correct order was maintained
+ throw new ExpectedException();
+ }
+
+ @Override
+ protected DistributedConfig createConfig(Map<String, String> workerProps) {
+ DistributedConfig config = new DistributedConfig(workerProps);
+ // Mock kafkaClusterId() to avoid connecting to Kafka broker
+ DistributedConfig spyConfig = spy(config);
+ doReturn("test-cluster-id").when(spyConfig).kafkaClusterId();
+ return spyConfig;
+ }
+ }
+
+ /**
+ * ClassLoader that cannot load a specific class, simulating plugin classes only in plugin.path.
+ */
+ private static class RestrictedClassLoader extends ClassLoader {
+ private final String restrictedClassName;
+ private final ClassLoader systemLoader;
+
+ RestrictedClassLoader(String restrictedClassName) {
+ super(null); // No parent to prevent delegation
+ this.restrictedClassName = restrictedClassName;
+ this.systemLoader = ClassLoader.getSystemClassLoader();
+ }
+
+ @Override
+ protected Class<?> loadClass(String name, boolean resolve) throws ClassNotFoundException {
+ // Block the restricted class and test plugin classes
+ if (name.equals(restrictedClassName)
+ || (name.startsWith("test.plugins") && name.contains("Sampling"))) {
Review Comment:
Why add this extra check? the goal is simply to reject `test.plugins.SamplingConfigProvider`, right?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] MINOR: Revert timing change for creating connect config [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on PR #20891:
URL: https://github.com/apache/kafka/pull/20891#issuecomment-3564955425
cherry-pick to 4.2. This patch resolves the regression caused by #20612
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org