This is an automated email from the ASF dual-hosted git repository.
mimaison pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/kafka.git
The following commit(s) were added to refs/heads/trunk by this push:
new 2310e991c06 KAFKA-20751: Preserve available versions for cached
connector plugin validation failures (#22712)
2310e991c06 is described below
commit 2310e991c0642ebc150c63c23a352c6d5c756dd8
Author: majialong <[email protected]>
AuthorDate: Wed Jul 8 22:31:52 2026 +0800
KAFKA-20751: Preserve available versions for cached connector plugin
validation failures (#22712)
Preserve the available plugin versions stored in cached connector plugin
validation failures.
This keeps repeated validation of an unavailable
`connector.plugin.version` consistent with the initial validation
response and avoids an NPE in the cached failure path.
Reviewers: Mickael Maison <[email protected]>
---
checkstyle/suppressions.xml | 2 +-
.../kafka/connect/runtime/CachedConnectors.java | 3 +-
.../kafka/connect/runtime/AbstractHerderTest.java | 32 ++++++++++
.../connect/runtime/CachedConnectorsTest.java | 74 ++++++++++++++++++++++
4 files changed, 109 insertions(+), 2 deletions(-)
diff --git a/checkstyle/suppressions.xml b/checkstyle/suppressions.xml
index 4f252572fb2..4c0ec0f8ecd 100644
--- a/checkstyle/suppressions.xml
+++ b/checkstyle/suppressions.xml
@@ -137,7 +137,7 @@
<!-- connect tests-->
<suppress checks="ClassDataAbstractionCoupling"
-
files="(DistributedHerder|KafkaBasedLog|WorkerSourceTask)Test.java"/>
+
files="(AbstractHerder|DistributedHerder|KafkaBasedLog|WorkerSourceTask)Test.java"/>
<suppress checks="ClassFanOutComplexity"
files="(WorkerSink|WorkerSource|ErrorHandling)TaskTest.java"/>
diff --git
a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/CachedConnectors.java
b/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/CachedConnectors.java
index ae9fe3b99e0..aa56f2d0387 100644
---
a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/CachedConnectors.java
+++
b/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/CachedConnectors.java
@@ -50,7 +50,8 @@ public class CachedConnectors {
String version = range == null ? LATEST_VERSION : range.toString();
if (invalidVersions.containsKey(connectorName) &&
invalidVersions.get(connectorName).containsKey(version)) {
- throw new
VersionedPluginLoadingException(invalidVersions.get(connectorName).get(version).getMessage());
+ VersionedPluginLoadingException exception =
invalidVersions.get(connectorName).get(version);
+ throw new VersionedPluginLoadingException(exception.getMessage(),
exception.availableVersions());
}
}
diff --git
a/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/AbstractHerderTest.java
b/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/AbstractHerderTest.java
index 02caa21812f..d1851138ecd 100644
---
a/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/AbstractHerderTest.java
+++
b/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/AbstractHerderTest.java
@@ -43,6 +43,7 @@ import org.apache.kafka.connect.runtime.isolation.LoaderSwap;
import org.apache.kafka.connect.runtime.isolation.PluginDesc;
import org.apache.kafka.connect.runtime.isolation.PluginType;
import org.apache.kafka.connect.runtime.isolation.Plugins;
+import
org.apache.kafka.connect.runtime.isolation.VersionedPluginLoadingException;
import org.apache.kafka.connect.runtime.rest.entities.ConfigInfo;
import org.apache.kafka.connect.runtime.rest.entities.ConfigInfos;
import org.apache.kafka.connect.runtime.rest.entities.ConfigKeyInfo;
@@ -405,6 +406,37 @@ public class AbstractHerderTest {
verifyValidationIsolation();
}
+ @Test
+ public void
testRepeatedConfigValidationOfInvalidVersionSurfacesAvailableVersions() {
+ Class<? extends Connector> connectorClass =
SampleSourceConnector.class;
+ String requestedVersion = "2.0.0";
+ List<String> availableVersions = List.of("1.0.0");
+
+ when(worker.getPlugins()).thenReturn(plugins);
+ when(plugins.newConnector(anyString(), any()))
+ .thenThrow(new VersionedPluginLoadingException("no matching
version", availableVersions));
+
+ AbstractHerder herder = testHerder();
+
+ Map<String, String> config = new HashMap<>();
+ config.put(ConnectorConfig.CONNECTOR_CLASS_CONFIG,
connectorClass.getName());
+ config.put(ConnectorConfig.CONNECTOR_VERSION, requestedVersion);
+
+ ConfigInfos first = herder.validateConnectorConfig(config, s -> null,
false);
+ ConfigInfos second = herder.validateConnectorConfig(config, s -> null,
false);
+
+ ConfigInfo firstVersionInfo = findInfo(first,
ConnectorConfig.CONNECTOR_VERSION);
+ assertNotNull(firstVersionInfo);
+ assertEquals(availableVersions,
firstVersionInfo.configValue().recommendedValues());
+
+ ConfigInfo secondVersionInfo = findInfo(second,
ConnectorConfig.CONNECTOR_VERSION);
+ assertNotNull(secondVersionInfo);
+ assertEquals(availableVersions,
secondVersionInfo.configValue().recommendedValues());
+
+ // The second validation comes from the cache, so the loader is only
invoked once.
+ verify(plugins, times(1)).newConnector(eq(connectorClass.getName()),
any());
+ }
+
@Test
public void testBuildRestartPlanForConnectorAndTasks() {
RestartRequest restartRequest = new RestartRequest(connectorName,
false, true);
diff --git
a/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/CachedConnectorsTest.java
b/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/CachedConnectorsTest.java
new file mode 100644
index 00000000000..903eb69d50c
--- /dev/null
+++
b/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/CachedConnectorsTest.java
@@ -0,0 +1,74 @@
+/*
+ * 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.runtime;
+
+import org.apache.kafka.connect.runtime.isolation.PluginUtils;
+import org.apache.kafka.connect.runtime.isolation.Plugins;
+import
org.apache.kafka.connect.runtime.isolation.VersionedPluginLoadingException;
+
+import org.apache.maven.artifact.versioning.VersionRange;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.mockito.Mock;
+import org.mockito.junit.jupiter.MockitoExtension;
+
+import java.util.List;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+@ExtendWith(MockitoExtension.class)
+public class CachedConnectorsTest {
+
+ @Mock
+ private Plugins plugins;
+
+ @Test
+ public void testCachedInvalidVersionFailurePreservesAvailableVersions()
throws Exception {
+ String requestedVersion = "2.0.0";
+ String connectorClass =
"org.apache.kafka.connect.runtime.SomeConnector";
+ List<String> availableVersions = List.of("1.0.0");
+ VersionedPluginLoadingException loadingException =
+ new VersionedPluginLoadingException("no matching version",
availableVersions);
+ when(plugins.newConnector(anyString(),
any())).thenThrow(loadingException);
+
+ CachedConnectors cachedConnectors = new CachedConnectors(plugins);
+ VersionRange versionRange =
PluginUtils.connectorVersionRequirement(requestedVersion);
+
+ VersionedPluginLoadingException firstException = assertThrows(
+ VersionedPluginLoadingException.class,
+ () -> cachedConnectors.getConnector(connectorClass, versionRange)
+ );
+ VersionedPluginLoadingException cachedException = assertThrows(
+ VersionedPluginLoadingException.class,
+ () -> cachedConnectors.getConnector(connectorClass, versionRange)
+ );
+
+ // A cached version loading failure should preserve its exception
details.
+ assertEquals(firstException.getMessage(),
cachedException.getMessage());
+ assertEquals(availableVersions, cachedException.availableVersions());
+
+ // The second lookup comes from the cache, so the loader is only
invoked once.
+ verify(plugins, times(1)).newConnector(connectorClass, versionRange);
+ }
+}
