Repository: apache/kafka
File: tools/src/main/java/org/apache/kafka/tools/ConnectPluginPath.java (Java)

Code before:
            default -> throw new ArgumentParserException("Unrecognized subcommand: '" + subcommand + "'", parser);
        };
    }

    private static Set<Path> parseLocations(ArgumentParser parser, Namespace namespace) throws ArgumentParserException, TerseException {
        List<String> rawLocations = new ArrayList<>(namespace.getList("plugin_location"));
        List<String> rawPluginPaths = new ArrayList<>(namespace.getList("plugin_path"));
        List<String> rawWorkerConfigs = new ArrayList<>(namespace.getList("worker_config"));
        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;
            try {
                properties = Utils.loadProps(rawWorkerConfig);
            } catch (IOException e) {
                throw new TerseException("Unable to read worker config at " + rawWorkerConfig);
            }
            String pluginPath = properties.getProperty(WorkerConfig.PLUGIN_PATH_CONFIG);
            if (pluginPath != null) {
                validatePluginPath(pluginPath, WorkerConfig.PLUGIN_PATH_CONFIG);
                rawPluginPaths.add(pluginPath);
            }
        }
        for (String rawPluginPath : rawPluginPaths) {
            try {
                pluginLocations.addAll(PluginUtils.pluginLocations(rawPluginPath, true));
            } catch (UncheckedIOException e) {
                throw new TerseException("Unable to parse plugin path " + rawPluginPath + ": " + e.getMessage());
            }
        }
        for (String rawLocation : rawLocations) {
            Path pluginLocation = Paths.get(rawLocation);
            if (!pluginLocation.toFile().exists()) {
                throw new TerseException("Specified location " + pluginLocation + " does not exist");
            }
            pluginLocations.add(pluginLocation);
        }
        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.");
        }


Review comment (suggestion):
I think we should throw a `ConfigException` when validating `WorkerConfig.PLUGIN_PATH_CONFIG` at this stage. This would preserve the same behavior as when `WorkerConfig.PLUGIN_PATH_CONFIG` is set to an empty value in the config.

Code after:
                new Config(Command.SYNC_MANIFESTS, locations, namespace.getBoolean("dry_run"), namespace.getBoolean("keep_not_found"), out, err);
            default -> throw new ArgumentParserException("Unrecognized subcommand: '" + subcommand + "'", parser);
        };
    }

    private static Set<Path> parseLocations(ArgumentParser parser, Namespace namespace) throws ArgumentParserException, TerseException {
        List<String> rawLocations = new ArrayList<>(namespace.getList("plugin_location"));
        List<String> rawPluginPaths = new ArrayList<>(namespace.getList("plugin_path"));
        List<String> rawWorkerConfigs = new ArrayList<>(namespace.getList("worker_config"));
        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;
            try {
                properties = Utils.loadProps(rawWorkerConfig);
            } catch (IOException e) {
                throw new TerseException("Unable to read worker config at " + rawWorkerConfig);
            }
            String pluginPath = properties.getProperty(WorkerConfig.PLUGIN_PATH_CONFIG);
            if (pluginPath != null) {
                validatePluginPath(pluginPath, WorkerConfig.PLUGIN_PATH_CONFIG);
                rawPluginPaths.add(pluginPath);
            }
        }
        for (String rawPluginPath : rawPluginPaths) {
            try {
                pluginLocations.addAll(PluginUtils.pluginLocations(rawPluginPath, true));
            } catch (UncheckedIOException e) {
                throw new TerseException("Unable to parse plugin path " + rawPluginPath + ": " + e.getMessage());
            }
        }
        for (String rawLocation : rawLocations) {
            Path pluginLocation = Paths.get(rawLocation);
            if (!pluginLocation.toFile().exists()) {
                throw new TerseException("Specified location " + pluginLocation + " does not exist");
            }
            pluginLocations.add(pluginLocation);
        }
        return pluginLocations;
    }

    private static void validatePluginPath(String pluginPath, String configName) throws ConfigException {
        String trimmed = pluginPath.trim();
        if (trimmed.isEmpty()) {
            throw new ConfigException("'" + configName + "' must not be empty.");
        }