MINOR: Add plugin.path config non-empty check in connect-plugin-path.sh

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

@@ -209,9 +209,6 @@ 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);
-                }
                 if (!Files.exists(pluginPathElement)) {
                     throw new FileNotFoundException(pluginPathElement.toString());
                 }

tools/src/main/java/org/apache/kafka/tools/ConnectPluginPath.java

@@ -52,6 +52,7 @@
 import java.util.Objects;
 import java.util.Properties;
 import java.util.Set;
+import java.util.regex.Pattern;
 import java.util.stream.Collectors;
 import java.util.stream.Stream;
 
@@ -68,6 +69,8 @@ public class ConnectPluginPath {
     };
     public static final String NO_ALIAS = "N/A";
 
+    private static final Pattern COMMA_WITH_WHITESPACE = Pattern.compile("\\s*,\\s*");
+
     public static void main(String[] args) {
         Exit.exit(mainNoExit(args, System.out, System.err));
     }
@@ -162,6 +165,9 @@ private static Set<Path> parseLocations(ArgumentParser parser, Namespace namespa
         if (rawLocations.isEmpty() && rawPluginPaths.isEmpty() && rawWorkerConfigs.isEmpty()) {
             throw new ArgumentParserException("Must specify at least one --plugin-location, --plugin-path, or --worker-config", parser);
         }
+        for (String pluginPath : rawPluginPaths) {
+            validatePluginPath(pluginPath, "--plugin-path");
+        }
         Set<Path> pluginLocations = new LinkedHashSet<>();
         for (String rawWorkerConfig : rawWorkerConfigs) {
             Properties properties;
@@ -172,6 +178,7 @@ private static Set<Path> parseLocations(ArgumentParser parser, Namespace namespa
             }
             String pluginPath = properties.getProperty(WorkerConfig.PLUGIN_PATH_CONFIG);
             if (pluginPath != null) {
+                validatePluginPath(pluginPath, WorkerConfig.PLUGIN_PATH_CONFIG);
                 rawPluginPaths.add(pluginPath);
             }
         }
@@ -192,6 +199,21 @@ private static Set<Path> parseLocations(ArgumentParser parser, Namespace namespa
         return pluginLocations;
     }
 
+    private static void validatePluginPath(String pluginPath, String configName) throws TerseException {
+        String trimmed = pluginPath.trim();
+        if (trimmed.isEmpty()) {
+            throw new TerseException("'" + configName + "' must not be empty.");
+        }
+
+        String[] pluginPathElements = COMMA_WITH_WHITESPACE.split(trimmed, -1);
+
+        for (String path : pluginPathElements) {
+            if (path.isEmpty()) {
+                throw new TerseException("'" + configName + "' values must not be empty.");
+            }
+        }
+    }
+
     enum Command {
         LIST, SYNC_MANIFESTS
     }

tools/src/test/java/org/apache/kafka/tools/ConnectPluginPathTest.java

@@ -333,6 +333,63 @@ public void testSyncManifestsKeepNotFound(PluginLocationType type) {
         assertBadPackagingPluginsStatus(table, false);
     }
 
+    @Test
+    public void testListEmptyPluginPathArg() {
+        CommandResult res = runCommand(
+                "list",
+                "--plugin-path",
+                ""
+        );
+        assertNotEquals(0, res.returnCode);
+        assertTrue("'--plugin-path' must not be empty.\n".equals(res.err));
+    }
+
+    @Test
+    public void testListEmptyPluginPathElementArg() {
+        CommandResult res = runCommand(
+                "list",
+                "--plugin-path",
+                "location-a,,location-b"
+        );
+        assertNotEquals(0, res.returnCode);
+        assertTrue("'--plugin-path' values must not be empty.\n".equals(res.err));
+    }
+
+    @Test
+    public void testListEmptyPluginPathInWorkerConfig() {
+        Path configPath = workspace.resolve("worker-empty.properties");
+        try {
+            Files.writeString(configPath, "plugin.path=", StandardCharsets.UTF_8);
+        } catch (IOException e) {
+            fail("Failed to create test worker config: " + e.getMessage());
+        }
+
+        CommandResult res = runCommand(
+                "list",
+                "--worker-config",
+                configPath.toString()
+        );
+        assertNotEquals(0, res.returnCode);
+        assertTrue("'plugin.path' must not be empty.\n".equals(res.err));
+    }
+
+    @Test
+    public void testListEmptyPluginPathElementInWorkerConfig() {
+        Path configPath = workspace.resolve("worker-empty-element.properties");
+        try {
+            Files.writeString(configPath, "plugin.path=location-a,,location-b", StandardCharsets.UTF_8);
+        } catch (IOException e) {
+            fail("Failed to create test worker config: " + e.getMessage());
+        }
+
+        CommandResult res = runCommand(
+                "list",
+                "--worker-config",
+                configPath.toString()
+        );
+        assertNotEquals(0, res.returnCode);
+        assertTrue("'plugin.path' values must not be empty.\n".equals(res.err));
+    }
 
     private static Map<String, List<String[]>> assertListSuccess(CommandResult result) {
         assertEquals(0, result.returnCode);