[GitHub] [kafka] gharris1727 opened a new pull request, #14195: KAFKA-15228: Add sync-manifests command to connect-plugin-path (KIP-898)

["gharris1727 (via GitHub)" <gi...@apache.org>]

gharris1727 opened a new pull request, #14195:
URL: https://github.com/apache/kafka/pull/14195

   This is the secondary component of KIP-898, a command to migrate released artifacts to include ServiceLoader manifests.
   
   In order to provide dry-run functionality and to reduce the number of write operations to the filesystem, the script will accumulate the changes to the manifests in-memory. Once the changes to manifests are completely known, the script can then generate the new manifest files, dry run the sync, and if the dry-run flag is omitted, actually write the changes to disk.
   
   In order to model the state of the manifests in-memory and the operations on manifest files, replace the existing ManifestEntry and Map system with the `ManifestWorkspace`. Each overall workspace has several `SourceWorkspace` objects, one corresponding to each PluginSource. The `SourceWorkspace` is an abstraction over the several different types of plugin sources, which each have different semantics around reading and writing manifests.
   
   The actual JAR editing is done by opening the JAR files as a FileSystem, and then using normal file write operations on the internal URL. This allows code reuse between the Class Hierarchy and Single/Multi jar cases, and means that we're using the existing zip/jar libraries instead of binary editing.
   
   ### Committer Checklist (excluded from commit message)
   - [ ] Verify design and implementation 
   - [ ] Verify test coverage and CI build status
   - [ ] Verify documentation (including upgrade notes)
   


-- 
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

[GitHub] [kafka] gharris1727 commented on a diff in pull request #14195: KAFKA-15228: Add sync-manifests command to connect-plugin-path (KIP-898)

["gharris1727 (via GitHub)" <gi...@apache.org>]

gharris1727 commented on code in PR #14195:
URL: https://github.com/apache/kafka/pull/14195#discussion_r1294058501


##########
tools/src/main/java/org/apache/kafka/tools/ManifestWorkspace.java:
##########
@@ -0,0 +1,573 @@
+/*
+ * 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.tools;
+
+import org.apache.kafka.connect.runtime.isolation.PluginSource;
+import org.apache.kafka.connect.runtime.isolation.PluginType;
+
+import java.io.BufferedOutputStream;
+import java.io.BufferedReader;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.InputStreamReader;
+import java.io.OutputStream;
+import java.io.PrintStream;
+import java.net.MalformedURLException;
+import java.net.URI;
+import java.net.URISyntaxException;
+import java.net.URL;
+import java.net.URLConnection;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.FileSystem;
+import java.nio.file.FileSystems;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+import java.nio.file.StandardCopyOption;
+import java.nio.file.StandardOpenOption;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.EnumMap;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.LinkedHashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Set;
+import java.util.function.BiConsumer;
+import java.util.zip.ZipInputStream;
+import java.util.zip.ZipOutputStream;
+
+/**
+ * An in-memory workspace for manipulating {@link java.util.ServiceLoader} manifest files.
+ * <p>Use {@link #forSource(PluginSource)} to get a workspace scoped to a single plugin location, which is able
+ * to accept simulated reads and writes of manifests.
+ * Write the simulated changes to disk via {@link #commit(boolean)}.
+ */
+public class ManifestWorkspace {
+
+    private static final String MANIFEST_PREFIX = "META-INF/services/";
+    private static final Path MANAGED_PATH = Paths.get("connect-plugin-path-shim-0.0.1-SNAPSHOT.jar");

Review Comment:
   I think the only constraint on this name is that it ends in `.jar`, so this name is extremely arbitrary. Just in case other tools happen to assume that all JARs have versions, I wanted to include one here.
   
   Since I expect people to use this JAR in production environments, the SNAPSHOT suffix is certainly likely to raise attention. 1.0.0 is much more appropriate here.
   
   
   



-- 
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

[GitHub] [kafka] gharris1727 commented on a diff in pull request #14195: KAFKA-15228: Add sync-manifests command to connect-plugin-path (KIP-898)

["gharris1727 (via GitHub)" <gi...@apache.org>]

gharris1727 commented on code in PR #14195:
URL: https://github.com/apache/kafka/pull/14195#discussion_r1293958574


##########
tools/src/main/java/org/apache/kafka/tools/ManifestWorkspace.java:
##########
@@ -0,0 +1,573 @@
+/*
+ * 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.tools;
+
+import org.apache.kafka.connect.runtime.isolation.PluginSource;
+import org.apache.kafka.connect.runtime.isolation.PluginType;
+
+import java.io.BufferedOutputStream;
+import java.io.BufferedReader;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.InputStreamReader;
+import java.io.OutputStream;
+import java.io.PrintStream;
+import java.net.MalformedURLException;
+import java.net.URI;
+import java.net.URISyntaxException;
+import java.net.URL;
+import java.net.URLConnection;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.FileSystem;
+import java.nio.file.FileSystems;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+import java.nio.file.StandardCopyOption;
+import java.nio.file.StandardOpenOption;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.EnumMap;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.LinkedHashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Set;
+import java.util.function.BiConsumer;
+import java.util.zip.ZipInputStream;
+import java.util.zip.ZipOutputStream;
+
+/**
+ * An in-memory workspace for manipulating {@link java.util.ServiceLoader} manifest files.
+ * <p>Use {@link #forSource(PluginSource)} to get a workspace scoped to a single plugin location, which is able
+ * to accept simulated reads and writes of manifests.
+ * Write the simulated changes to disk via {@link #commit(boolean)}.
+ */
+public class ManifestWorkspace {
+
+    private static final String MANIFEST_PREFIX = "META-INF/services/";
+    private static final Path MANAGED_PATH = Paths.get("connect-plugin-path-shim-0.0.1-SNAPSHOT.jar");
+    private final PrintStream out;
+    private final List<SourceWorkspace<?>> workspaces;
+    private final Map<Path, Path> temporaryOverlayFiles;
+
+    public ManifestWorkspace(PrintStream out) {
+        this.out = out;
+        workspaces = new ArrayList<>();
+        temporaryOverlayFiles = new HashMap<>();
+    }
+
+    public SourceWorkspace<?> forSource(PluginSource source) throws IOException {
+        SourceWorkspace<?> sourceWorkspace;
+        switch (source.type()) {
+            case CLASSPATH:
+                sourceWorkspace = new ClasspathWorkspace(source);
+                break;
+            case MULTI_JAR:
+                sourceWorkspace = new MultiJarWorkspace(source);
+                break;
+            case SINGLE_JAR:
+                sourceWorkspace = new SingleJarWorkspace(source);
+                break;
+            case CLASS_HIERARCHY:
+                sourceWorkspace = new ClassHierarchyWorkspace(source);
+                break;
+            default:
+                throw new IllegalStateException("Unknown source type " + source.type());
+        }
+        workspaces.add(sourceWorkspace);
+        return sourceWorkspace;
+    }
+
+    /**
+     * Commits all queued changes to disk
+     * @return true if any workspace wrote changes to disk, false if all workspaces did not have writes to apply
+     * @throws IOException if an error occurs reading or writing to the filesystem
+     * @throws TerseException if a path is not writable on disk and should be.
+     */
+    public boolean commit(boolean dryRun) throws IOException, TerseException {
+        boolean changed = false;
+        for (SourceWorkspace<?> workspace : workspaces) {
+            changed |= workspace.commit(dryRun);
+        }
+        return changed;
+    }
+
+    /**
+     * A workspace scoped to a single plugin source.
+     * <p>Buffers simulated reads and writes to the plugin path before they can be written to disk.
+     * @param <T> The data structure used by the workspace to store in-memory manifests internally.
+     */
+    public static abstract class SourceWorkspace<T> {
+        private final Path location;
+        private final PluginSource.Type type;
+        protected final T initial;
+        protected final T manifests;
+
+        private SourceWorkspace(PluginSource source) throws IOException {
+            this.location = source.location();
+            this.type = source.type();
+            this.initial = load(source);
+            this.manifests = load(source);
+        }
+
+        public Path location() {
+            return location;
+        }
+
+        public PluginSource.Type type() {
+            return type;
+        }
+
+        protected abstract T load(PluginSource source) throws IOException;
+
+        public abstract boolean hasManifest(PluginType type, String className);
+
+        public abstract void forEach(BiConsumer<String, PluginType> consumer);
+
+        public abstract void addManifest(PluginType type, String pluginClass);
+
+        public abstract void removeManifest(PluginType type, String pluginClass);
+
+        protected abstract boolean commit(boolean dryRun) throws TerseException, IOException;
+
+        protected static Map<PluginType, Set<String>> loadManifest(URL baseUrl) throws MalformedURLException {
+            Map<PluginType, Set<String>> manifests = new EnumMap<>(PluginType.class);
+            for (PluginType type : PluginType.values()) {
+                Set<String> result;
+                try {
+                    URL u = new URL(baseUrl, MANIFEST_PREFIX + type.superClass().getName());
+                    result = parse(u);
+                } catch (RuntimeException e) {
+                    result = new LinkedHashSet<>();
+                }
+                manifests.put(type, result);
+            }
+            return manifests;
+        }
+
+        protected static URL jarBaseUrl(URL fileUrl) throws MalformedURLException {
+            return new URL("jar", "", -1, fileUrl + "!/", null);
+        }
+
+        protected static void forEach(Map<PluginType, Set<String>> manifests, BiConsumer<String, PluginType> consumer) {
+            manifests.forEach((type, classNames) -> classNames.forEach(className -> consumer.accept(className, type)));
+        }
+    }
+
+    /**
+     * A single jar can only contain one manifest per plugin type.
+     */
+    private class SingleJarWorkspace extends SourceWorkspace<Map<PluginType, Set<String>>> {
+
+        private SingleJarWorkspace(PluginSource source) throws IOException {
+            super(source);
+        }
+
+        @Override
+        protected Map<PluginType, Set<String>> load(PluginSource source) throws IOException {
+            return loadManifest(jarBaseUrl(source.urls()[0]));
+        }
+
+        @Override
+        public boolean hasManifest(PluginType type, String className) {
+            return manifests.get(type).contains(className);
+        }
+
+        @Override
+        public void forEach(BiConsumer<String, PluginType> consumer) {
+            forEach(manifests, consumer);
+        }
+
+        @Override
+        public void addManifest(PluginType type, String pluginClass) {
+            manifests.get(type).add(pluginClass);
+        }
+
+        @Override
+        public void removeManifest(PluginType type, String pluginClass) {
+            manifests.get(type).remove(pluginClass);
+        }
+
+        @Override
+        protected boolean commit(boolean dryRun) throws IOException, TerseException {
+            if (startSync(dryRun, location(), initial, manifests)) {
+                rewriteJar(dryRun, location(), manifests);
+                return true;
+            }
+            return false;
+        }
+    }
+
+    /**
+     * A classpath workspace is backed by multiple jars, and that is not writable.
+     * The in-memory format is a map from jar path to the manifests contained in that jar.
+     * The control flow of the caller should not perform writes, so these exceptions indicate a bug in the program.
+     */
+    private class ClasspathWorkspace extends SourceWorkspace<Map<Path, Map<PluginType, Set<String>>>> {
+
+        private ClasspathWorkspace(PluginSource source) throws IOException {
+            super(source);
+        }
+
+        @Override
+        protected Map<Path, Map<PluginType, Set<String>>> load(PluginSource source) throws IOException {
+            Map<Path, Map<PluginType, Set<String>>> manifestsBySubLocation = new HashMap<>();
+            for (URL url : source.urls()) {
+                Path jarPath = Paths.get(url.getPath());
+                manifestsBySubLocation.put(jarPath, loadManifest(jarBaseUrl(url)));
+            }
+            return manifestsBySubLocation;
+        }
+
+        public boolean hasManifest(PluginType type, String className) {
+            return manifests.values()
+                    .stream()
+                    .map(m -> m.get(type))
+                    .anyMatch(s -> s.contains(className));
+        }
+
+        public void forEach(BiConsumer<String, PluginType> consumer) {
+            manifests.values().forEach(m -> forEach(m, consumer));
+        }
+
+        @Override
+        public void addManifest(PluginType type, String pluginClass) {
+            throw new UnsupportedOperationException("Cannot change the contents of the classpath");
+        }
+
+        @Override
+        public void removeManifest(PluginType type, String pluginClass) {
+            throw new UnsupportedOperationException("Cannot change the contents of the classpath");
+        }
+
+        @Override
+        protected boolean commit(boolean dryRun) throws IOException, TerseException {
+            // There is never anything to commit for the classpath
+            return false;
+        }
+    }
+
+    /**
+     * A multi-jar workspace is similar to the classpath workspace because it has multiple jars.
+     * However, the multi-jar workspace is writable, and injects a managed jar where it writes added manifests.
+     */
+    private class MultiJarWorkspace extends ClasspathWorkspace {
+
+        private MultiJarWorkspace(PluginSource source) throws IOException {
+            super(source);
+        }
+
+        @Override
+        protected Map<Path, Map<PluginType, Set<String>>> load(PluginSource source) throws IOException {
+            Map<Path, Map<PluginType, Set<String>>> manifests = super.load(source);
+            // In addition to the normal multi-jar paths, inject a managed jar where we can add manifests.
+            Path managedPath = source.location().resolve(MANAGED_PATH);
+            URL url = managedPath.toUri().toURL();
+            manifests.put(managedPath, loadManifest(jarBaseUrl(url)));
+            return manifests;
+        }
+
+        @Override
+        public void addManifest(PluginType type, String pluginClass) {
+            // Add plugins to the managed manifest
+            manifests.get(location().resolve(MANAGED_PATH)).get(type).add(pluginClass);
+        }
+
+        @Override
+        public void removeManifest(PluginType type, String pluginClass) {
+            // If a plugin appears in multiple manifests, remove it from all of them.
+            for (Map<PluginType, Set<String>> manifestState : manifests.values()) {
+                manifestState.get(type).remove(pluginClass);
+            }
+        }
+
+        @Override
+        public boolean commit(boolean dryRun) throws IOException, TerseException {
+            boolean changed = false;
+            for (Map.Entry<Path, Map<PluginType, Set<String>>> manifestSource : manifests.entrySet()) {
+                Path jarPath = manifestSource.getKey();
+                Map<PluginType, Set<String>> before = initial.get(jarPath);
+                Map<PluginType, Set<String>> after = manifestSource.getValue();
+                if (startSync(dryRun, jarPath, before, after)) {
+                    rewriteJar(dryRun, jarPath, after);
+                    changed = true;
+                }
+            }
+            return changed;
+        }
+    }
+
+    /**
+     * The class hierarchy is similar to the single-jar because there can only be one manifest per type.
+     * However, the path to that single manifest is accessed via the pluginLocation.
+     */
+    private class ClassHierarchyWorkspace extends SingleJarWorkspace {
+
+        private ClassHierarchyWorkspace(PluginSource source) throws IOException {
+            super(source);
+        }
+
+        @Override
+        protected Map<PluginType, Set<String>> load(PluginSource source) throws IOException {
+            return loadManifest(source.location().toUri().toURL());
+        }
+
+        protected boolean commit(boolean dryRun) throws IOException, TerseException {
+            if (startSync(dryRun, location(), initial, manifests)) {
+                rewriteClassHierarchyManifest(dryRun, location(), manifests);
+                return true;
+            }
+            return false;
+        }
+    }
+
+    private boolean startSync(boolean dryRun, Path syncLocation, Map<PluginType, Set<String>> before, Map<PluginType, Set<String>> after) {
+        Objects.requireNonNull(syncLocation, "syncLocation must be non-null");
+        Objects.requireNonNull(before, "before must be non-null");
+        Objects.requireNonNull(after, "after must be non-null");
+        if (before.equals(after)) {
+            return false;
+        }
+        Set<String> added = new HashSet<>();
+        after.values().forEach(added::addAll);
+        before.values().forEach(added::removeAll);
+        Set<String> removed = new HashSet<>();
+        before.values().forEach(removed::addAll);
+        after.values().forEach(removed::removeAll);
+        out.printf("%sSync\t\t%s Add %s Remove %s%n", dryRun ? "Dry Run " : "", syncLocation, added, removed);
+        return true;
+    }
+
+    /**
+     * Rewrite a jar on disk to contain a manifests containing the specified entries.
+     * Will create the jar file and manifests if they do not exist.
+     * Will delete the jar file and manifests if they become empty.
+     *
+     * @param dryRun        True if the rewrite should be applied, false if it should be simulated.
+     * @param jarPath       Path to a jar file for a plugin
+     * @param manifestState Map from plugin type to Class names of plugins which should appear in that manifest
+     */
+    private void rewriteJar(boolean dryRun, Path jarPath, Map<PluginType, Set<String>> manifestState) throws IOException, TerseException {
+        Objects.requireNonNull(jarPath, "jarPath must be non-null");
+        Objects.requireNonNull(manifestState, "manifestState must be non-null");
+        Path writableJar = getWritablePath(dryRun, jarPath);
+        if (!manifestState.values().stream().allMatch(Collection::isEmpty) && !Files.exists(writableJar)) {
+            out.printf("\tCreate\t%s%n", jarPath);
+            createJar(writableJar);
+        }
+        try (FileSystem jar = FileSystems.newFileSystem(
+                new URI("jar", writableJar.toUri().toString(), ""),
+                Collections.emptyMap()
+        )) {
+            Path zipRoot = jar.getRootDirectories().iterator().next();
+            rewriteClassHierarchyManifest(dryRun, zipRoot, manifestState);
+        } catch (URISyntaxException e) {
+            throw new IOException(e);
+        }
+        if (Files.exists(writableJar) && jarIsEmpty(writableJar)) {
+            Files.delete(writableJar);
+        }
+    }
+
+    private void createJar(Path path) throws IOException {
+        Objects.requireNonNull(path, "path must be non-null");
+        try (ZipOutputStream stream = new ZipOutputStream(Files.newOutputStream(
+                path,
+                StandardOpenOption.CREATE,
+                StandardOpenOption.WRITE,
+                StandardOpenOption.TRUNCATE_EXISTING
+        ))) {
+            stream.closeEntry();
+        }
+    }
+
+    private boolean jarIsEmpty(Path path) throws IOException {
+        Objects.requireNonNull(path, "path must be non-null");
+        try (ZipInputStream stream = new ZipInputStream(Files.newInputStream(
+                path,
+                StandardOpenOption.READ
+        ))) {
+            return stream.getNextEntry() == null;
+        }
+    }
+
+    /**
+     * Rewrite a multiple manifest files contained inside a class hierarchy.
+     * Will create the files and parent directories if they not exist.
+     * Will delete the files and parent directories within the plugin location if they become empty.
+     *
+     * @param dryRun         True if the rewrite should be applied, false if it should be simulated.
+     * @param pluginLocation Path to top-level of class hierarchy for a plugin
+     * @param manifestState  Map from plugin type to Class names of plugins which should appear in that manifest
+     */
+    private void rewriteClassHierarchyManifest(boolean dryRun, Path pluginLocation, Map<PluginType, Set<String>> manifestState) throws IOException, TerseException {
+        Objects.requireNonNull(pluginLocation, "pluginLocation must be non-null");
+        Objects.requireNonNull(manifestState, "manifestState must be non-null");
+        if (!Files.exists(pluginLocation)) {
+            throw new TerseException(pluginLocation + " does not exist");
+        }
+        if (!Files.isWritable(pluginLocation)) {
+            throw new TerseException(pluginLocation + " is not writable");
+        }
+        for (Map.Entry<PluginType, Set<String>> manifest : manifestState.entrySet()) {
+            PluginType type = manifest.getKey();
+            Set<String> elements = manifest.getValue();
+            Path manifestPath = pluginLocation.resolve(MANIFEST_PREFIX).resolve(type.superClass().getName());
+            rewriteManifestFile(dryRun, manifestPath, elements);
+        }
+    }
+
+    /**
+     * Rewrite a file, creating it and any parent directories if it does not exist, deleting it if specified
+     *
+     * @param dryRun              True if the rewrite should be applied, false if it should be simulated.
+     * @param filePath            Path to file which should be rewritten.
+     * @param elements            Class names of plugins which should appear in the manifest
+     */
+    private void rewriteManifestFile(boolean dryRun, Path filePath, Set<String> elements) throws IOException, TerseException {
+        Objects.requireNonNull(filePath, "filePath must be non-null");
+        Objects.requireNonNull(elements, "elements must be non-null");
+        Path writableFile = getWritablePath(dryRun, filePath);
+        if (elements.isEmpty()) {
+            if (Files.exists(filePath)) {
+                out.printf("\tDelete\t%s%n", filePath);
+                if (!dryRun) {
+                    Files.delete(writableFile);
+                }
+            }
+            // TODO: delete parent directories if empty up to the containing directory
+        } else {
+            Path parent = filePath.getParent();
+            if (parent != null) {
+                if (Files.exists(parent)) {
+                    if (!Files.isWritable(parent)) {
+                        throw new TerseException(parent + " is not writable");
+                    }
+                } else {
+                    out.printf("\tCreate\t%s%n", parent);
+                    if (!dryRun) {
+                        Files.createDirectories(parent);
+                    }
+                }
+            }
+            if (!Files.exists(filePath)) {
+                out.printf("\tCreate\t%s%n", filePath);
+            }
+            out.printf("\tWrite\t%s with content %s%n", filePath, elements);
+            if (!dryRun) {
+                try (OutputStream stream = new BufferedOutputStream(Files.newOutputStream(
+                        writableFile,
+                        StandardOpenOption.CREATE,
+                        StandardOpenOption.WRITE,
+                        StandardOpenOption.TRUNCATE_EXISTING

Review Comment:
   Yes everything you mentioned is correct.



-- 
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

[GitHub] [kafka] C0urante commented on a diff in pull request #14195: KAFKA-15228: Add sync-manifests command to connect-plugin-path (KIP-898)

["C0urante (via GitHub)" <gi...@apache.org>]

C0urante commented on code in PR #14195:
URL: https://github.com/apache/kafka/pull/14195#discussion_r1294779885


##########
tools/src/test/java/org/apache/kafka/tools/ConnectPluginPathTest.java:
##########
@@ -204,24 +353,25 @@ private static void assertPluginsAreCompatible(Map<String, List<String[]>> table
         assertPluginMigrationStatus(table, true, true, plugins);
     }
 
-    private static void assertNonMigratedPluginsPresent(Map<String, List<String[]>> table) {
-        assertPluginMigrationStatus(table, true, false,
+    private static void assertNonMigratedPluginsStatus(Map<String, List<String[]>> table, boolean migrated) {
+        assertPluginMigrationStatus(table, true, migrated,
                 TestPlugins.TestPlugin.NON_MIGRATED_CONVERTER,
                 TestPlugins.TestPlugin.NON_MIGRATED_HEADER_CONVERTER,
                 TestPlugins.TestPlugin.NON_MIGRATED_PREDICATE,
                 TestPlugins.TestPlugin.NON_MIGRATED_SINK_CONNECTOR,
                 TestPlugins.TestPlugin.NON_MIGRATED_SOURCE_CONNECTOR,
                 TestPlugins.TestPlugin.NON_MIGRATED_TRANSFORMATION);
         // This plugin is partially compatible
-        assertPluginMigrationStatus(table, true, null,
+        assertPluginMigrationStatus(table, true, migrated ? true : null,
                 TestPlugins.TestPlugin.NON_MIGRATED_MULTI_PLUGIN);
     }
 
-    private static void assertBadPackagingPluginsPresent(Map<String, List<String[]>> table) {
+    private static void assertBadPackaginPluginsStatus(Map<String, List<String[]>> table, boolean migrated) {

Review Comment:
   Nit:
   ```suggestion
       private static void assertBadPackagingPluginsStatus(Map<String, List<String[]>> table, boolean migrated) {
   ```



##########
tools/src/test/java/org/apache/kafka/tools/ConnectPluginPathTest.java:
##########
@@ -204,24 +353,25 @@ private static void assertPluginsAreCompatible(Map<String, List<String[]>> table
         assertPluginMigrationStatus(table, true, true, plugins);
     }
 
-    private static void assertNonMigratedPluginsPresent(Map<String, List<String[]>> table) {
-        assertPluginMigrationStatus(table, true, false,
+    private static void assertNonMigratedPluginsStatus(Map<String, List<String[]>> table, boolean migrated) {
+        assertPluginMigrationStatus(table, true, migrated,
                 TestPlugins.TestPlugin.NON_MIGRATED_CONVERTER,
                 TestPlugins.TestPlugin.NON_MIGRATED_HEADER_CONVERTER,
                 TestPlugins.TestPlugin.NON_MIGRATED_PREDICATE,
                 TestPlugins.TestPlugin.NON_MIGRATED_SINK_CONNECTOR,
                 TestPlugins.TestPlugin.NON_MIGRATED_SOURCE_CONNECTOR,
                 TestPlugins.TestPlugin.NON_MIGRATED_TRANSFORMATION);
         // This plugin is partially compatible
-        assertPluginMigrationStatus(table, true, null,
+        assertPluginMigrationStatus(table, true, migrated ? true : null,
                 TestPlugins.TestPlugin.NON_MIGRATED_MULTI_PLUGIN);
     }
 
-    private static void assertBadPackagingPluginsPresent(Map<String, List<String[]>> table) {
+    private static void assertBadPackaginPluginsStatus(Map<String, List<String[]>> table, boolean migrated) {
         assertPluginsAreCompatible(table,
                 TestPlugins.TestPlugin.BAD_PACKAGING_CO_LOCATED,
                 TestPlugins.TestPlugin.BAD_PACKAGING_VERSION_METHOD_THROWS_CONNECTOR);
-        assertPluginMigrationStatus(table, false, true,
+        // These plugin

Review Comment:
   Incomplete comment?



##########
tools/src/main/java/org/apache/kafka/tools/ConnectPluginPath.java:
##########
@@ -368,6 +391,30 @@ private static void endCommand(
             config.out.printf("Total plugins:      \t%d%n", totalPlugins);
             config.out.printf("Loadable plugins:   \t%d%n", loadablePlugins);
             config.out.printf("Compatible plugins: \t%d%n", compatiblePlugins);
+        } else if (config.command == Command.SYNC_MANIFESTS) {
+            if (workspace.commit(true)) {
+                if (config.dryRun) {
+                    config.out.println("Dry run passed: All above changes can be committed to disk if re-run with dry run disabled.");
+                } else {
+                    config.out.println("Writing changes to plugins...");
+                    workspace.commit(false);
+                    config.out.println("All plugins have accurate ServiceLoader manifests");
+                }
+            } else {
+                config.out.println("No changes required.");
+            }
+        }
+    }
+
+    private static void failCommand(Config config, Throwable e) {
+        if (config.command == Command.LIST) {
+            throw new RuntimeException("Unexpected error occurred while listing plugins", e);
+        } else if (config.command == Command.SYNC_MANIFESTS) {
+            if (config.dryRun) {
+                throw new RuntimeException("Unexpected error occurred while dry-running sync", e);
+            } else {
+                config.out.println("Connect plugin path now in unexpected state: Clear your plugin path and retry with dry run enabled");

Review Comment:
   I think we can do something like this:
   
   1. Reintroduce the `PrintStream err` field to the `Config` class
   2. In `ConnectPluginPath::endCommand`, in the `catch (Throwable t)` block after calling `workspace.commit(true)`, print the full stack trace of that error immediately with `t.printStackTrace(config.err);`
   3. Instead of throwing `t`, throw a `TerseException` with the last message we want the CLI to emit (e.g., "Sync incomplete, plugin path may be corrupted...")



##########
tools/src/test/java/org/apache/kafka/tools/ConnectPluginPathTest.java:
##########
@@ -192,6 +194,51 @@ public void testListMultipleWorkerConfigs(PluginLocationType type) {
                 TestPlugins.TestPlugin.SERVICE_LOADER);
     }
 
+    @ParameterizedTest
+    @EnumSource
+    public void testSyncManifests(PluginLocationType type) {
+        CommandResult res = runCommand(
+                "sync-manifests",
+                "--plugin-location",
+                setupLocation(workspace.resolve("location-a"), type, TestPlugins.TestPlugin.NON_MIGRATED_CONVERTER)
+        );
+        assertEquals(0, res.returnCode);
+        assertScanResult(true, TestPlugins.TestPlugin.NON_MIGRATED_CONVERTER, res.reflective);
+        assertScanResult(true, TestPlugins.TestPlugin.NON_MIGRATED_CONVERTER, res.serviceLoading);
+    }
+
+    @ParameterizedTest
+    @EnumSource
+    public void testSyncManifestsDryRun(PluginLocationType type) {
+        CommandResult res = runCommand(
+                "sync-manifests",
+                "--plugin-location",
+                setupLocation(workspace.resolve("location-a"), type, TestPlugins.TestPlugin.NON_MIGRATED_CONVERTER),
+                "--dry-run"
+        );
+        assertEquals(0, res.returnCode);
+        assertScanResult(true, TestPlugins.TestPlugin.NON_MIGRATED_CONVERTER, res.reflective);
+        assertScanResult(false, TestPlugins.TestPlugin.NON_MIGRATED_CONVERTER, res.serviceLoading);
+    }
+
+    @ParameterizedTest
+    @EnumSource
+    public void testSyncManifestsKeepNotFound(PluginLocationType type) {
+        CommandResult res = runCommand(
+                "sync-manifests",
+                "--plugin-location",
+                setupLocation(workspace.resolve("location-a"), type, TestPlugins.TestPlugin.BAD_PACKAGING_STATIC_INITIALIZER_THROWS_REST_EXTENSION),
+                "--plugin-location",
+                setupLocation(workspace.resolve("location-b"), type, TestPlugins.TestPlugin.NON_MIGRATED_CONVERTER),
+                "--keep-not-found"
+        );
+        assertEquals(0, res.returnCode);
+        assertScanResult(true, TestPlugins.TestPlugin.NON_MIGRATED_CONVERTER, res.reflective);
+        assertScanResult(true, TestPlugins.TestPlugin.NON_MIGRATED_CONVERTER, res.serviceLoading);
+        assertScanResult(false, TestPlugins.TestPlugin.BAD_PACKAGING_STATIC_INITIALIZER_THROWS_REST_EXTENSION, res.reflective);
+        assertScanResult(false, TestPlugins.TestPlugin.BAD_PACKAGING_STATIC_INITIALIZER_THROWS_REST_EXTENSION, res.serviceLoading);

Review Comment:
   Very nice!



-- 
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

[GitHub] [kafka] gharris1727 commented on a diff in pull request #14195: KAFKA-15228: Add sync-manifests command to connect-plugin-path (KIP-898)

["gharris1727 (via GitHub)" <gi...@apache.org>]

gharris1727 commented on code in PR #14195:
URL: https://github.com/apache/kafka/pull/14195#discussion_r1294053484


##########
tools/src/main/java/org/apache/kafka/tools/ManifestWorkspace.java:
##########
@@ -0,0 +1,573 @@
+/*
+ * 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.tools;
+
+import org.apache.kafka.connect.runtime.isolation.PluginSource;
+import org.apache.kafka.connect.runtime.isolation.PluginType;
+
+import java.io.BufferedOutputStream;
+import java.io.BufferedReader;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.InputStreamReader;
+import java.io.OutputStream;
+import java.io.PrintStream;
+import java.net.MalformedURLException;
+import java.net.URI;
+import java.net.URISyntaxException;
+import java.net.URL;
+import java.net.URLConnection;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.FileSystem;
+import java.nio.file.FileSystems;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+import java.nio.file.StandardCopyOption;
+import java.nio.file.StandardOpenOption;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.EnumMap;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.LinkedHashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Set;
+import java.util.function.BiConsumer;
+import java.util.zip.ZipInputStream;
+import java.util.zip.ZipOutputStream;
+
+/**
+ * An in-memory workspace for manipulating {@link java.util.ServiceLoader} manifest files.
+ * <p>Use {@link #forSource(PluginSource)} to get a workspace scoped to a single plugin location, which is able
+ * to accept simulated reads and writes of manifests.
+ * Write the simulated changes to disk via {@link #commit(boolean)}.
+ */
+public class ManifestWorkspace {
+
+    private static final String MANIFEST_PREFIX = "META-INF/services/";
+    private static final Path MANAGED_PATH = Paths.get("connect-plugin-path-shim-0.0.1-SNAPSHOT.jar");
+    private final PrintStream out;
+    private final List<SourceWorkspace<?>> workspaces;
+    private final Map<Path, Path> temporaryOverlayFiles;
+
+    public ManifestWorkspace(PrintStream out) {
+        this.out = out;
+        workspaces = new ArrayList<>();
+        temporaryOverlayFiles = new HashMap<>();
+    }
+
+    public SourceWorkspace<?> forSource(PluginSource source) throws IOException {
+        SourceWorkspace<?> sourceWorkspace;
+        switch (source.type()) {
+            case CLASSPATH:
+                sourceWorkspace = new ClasspathWorkspace(source);
+                break;
+            case MULTI_JAR:
+                sourceWorkspace = new MultiJarWorkspace(source);
+                break;
+            case SINGLE_JAR:
+                sourceWorkspace = new SingleJarWorkspace(source);
+                break;
+            case CLASS_HIERARCHY:
+                sourceWorkspace = new ClassHierarchyWorkspace(source);
+                break;
+            default:
+                throw new IllegalStateException("Unknown source type " + source.type());
+        }
+        workspaces.add(sourceWorkspace);
+        return sourceWorkspace;
+    }
+
+    /**
+     * Commits all queued changes to disk
+     * @return true if any workspace wrote changes to disk, false if all workspaces did not have writes to apply
+     * @throws IOException if an error occurs reading or writing to the filesystem
+     * @throws TerseException if a path is not writable on disk and should be.
+     */
+    public boolean commit(boolean dryRun) throws IOException, TerseException {
+        boolean changed = false;
+        for (SourceWorkspace<?> workspace : workspaces) {
+            changed |= workspace.commit(dryRun);
+        }
+        return changed;
+    }
+
+    /**
+     * A workspace scoped to a single plugin source.
+     * <p>Buffers simulated reads and writes to the plugin path before they can be written to disk.
+     * @param <T> The data structure used by the workspace to store in-memory manifests internally.
+     */
+    public static abstract class SourceWorkspace<T> {
+        private final Path location;
+        private final PluginSource.Type type;
+        protected final T initial;
+        protected final T manifests;
+
+        private SourceWorkspace(PluginSource source) throws IOException {
+            this.location = source.location();
+            this.type = source.type();
+            this.initial = load(source);
+            this.manifests = load(source);
+        }
+
+        public Path location() {
+            return location;
+        }
+
+        public PluginSource.Type type() {
+            return type;
+        }
+
+        protected abstract T load(PluginSource source) throws IOException;
+
+        public abstract boolean hasManifest(PluginType type, String className);
+
+        public abstract void forEach(BiConsumer<String, PluginType> consumer);
+
+        public abstract void addManifest(PluginType type, String pluginClass);
+
+        public abstract void removeManifest(PluginType type, String pluginClass);
+
+        protected abstract boolean commit(boolean dryRun) throws TerseException, IOException;
+
+        protected static Map<PluginType, Set<String>> loadManifest(URL baseUrl) throws MalformedURLException {
+            Map<PluginType, Set<String>> manifests = new EnumMap<>(PluginType.class);
+            for (PluginType type : PluginType.values()) {
+                Set<String> result;
+                try {
+                    URL u = new URL(baseUrl, MANIFEST_PREFIX + type.superClass().getName());
+                    result = parse(u);
+                } catch (RuntimeException e) {
+                    result = new LinkedHashSet<>();
+                }
+                manifests.put(type, result);
+            }
+            return manifests;
+        }
+
+        protected static URL jarBaseUrl(URL fileUrl) throws MalformedURLException {
+            return new URL("jar", "", -1, fileUrl + "!/", null);
+        }
+
+        protected static void forEach(Map<PluginType, Set<String>> manifests, BiConsumer<String, PluginType> consumer) {
+            manifests.forEach((type, classNames) -> classNames.forEach(className -> consumer.accept(className, type)));
+        }
+    }
+
+    /**
+     * A single jar can only contain one manifest per plugin type.
+     */
+    private class SingleJarWorkspace extends SourceWorkspace<Map<PluginType, Set<String>>> {
+
+        private SingleJarWorkspace(PluginSource source) throws IOException {
+            super(source);
+        }
+
+        @Override
+        protected Map<PluginType, Set<String>> load(PluginSource source) throws IOException {
+            return loadManifest(jarBaseUrl(source.urls()[0]));
+        }
+
+        @Override
+        public boolean hasManifest(PluginType type, String className) {
+            return manifests.get(type).contains(className);
+        }
+
+        @Override
+        public void forEach(BiConsumer<String, PluginType> consumer) {
+            forEach(manifests, consumer);
+        }
+
+        @Override
+        public void addManifest(PluginType type, String pluginClass) {
+            manifests.get(type).add(pluginClass);
+        }
+
+        @Override
+        public void removeManifest(PluginType type, String pluginClass) {
+            manifests.get(type).remove(pluginClass);
+        }
+
+        @Override
+        protected boolean commit(boolean dryRun) throws IOException, TerseException {
+            if (startSync(dryRun, location(), initial, manifests)) {
+                rewriteJar(dryRun, location(), manifests);
+                return true;
+            }
+            return false;
+        }
+    }
+
+    /**
+     * A classpath workspace is backed by multiple jars, and that is not writable.
+     * The in-memory format is a map from jar path to the manifests contained in that jar.
+     * The control flow of the caller should not perform writes, so these exceptions indicate a bug in the program.
+     */
+    private class ClasspathWorkspace extends SourceWorkspace<Map<Path, Map<PluginType, Set<String>>>> {
+
+        private ClasspathWorkspace(PluginSource source) throws IOException {
+            super(source);
+        }
+
+        @Override
+        protected Map<Path, Map<PluginType, Set<String>>> load(PluginSource source) throws IOException {
+            Map<Path, Map<PluginType, Set<String>>> manifestsBySubLocation = new HashMap<>();
+            for (URL url : source.urls()) {
+                Path jarPath = Paths.get(url.getPath());
+                manifestsBySubLocation.put(jarPath, loadManifest(jarBaseUrl(url)));
+            }
+            return manifestsBySubLocation;
+        }
+
+        public boolean hasManifest(PluginType type, String className) {
+            return manifests.values()
+                    .stream()
+                    .map(m -> m.get(type))
+                    .anyMatch(s -> s.contains(className));
+        }
+
+        public void forEach(BiConsumer<String, PluginType> consumer) {
+            manifests.values().forEach(m -> forEach(m, consumer));
+        }
+
+        @Override
+        public void addManifest(PluginType type, String pluginClass) {
+            throw new UnsupportedOperationException("Cannot change the contents of the classpath");
+        }
+
+        @Override
+        public void removeManifest(PluginType type, String pluginClass) {
+            throw new UnsupportedOperationException("Cannot change the contents of the classpath");
+        }
+
+        @Override
+        protected boolean commit(boolean dryRun) throws IOException, TerseException {
+            // There is never anything to commit for the classpath
+            return false;
+        }
+    }
+
+    /**
+     * A multi-jar workspace is similar to the classpath workspace because it has multiple jars.
+     * However, the multi-jar workspace is writable, and injects a managed jar where it writes added manifests.
+     */
+    private class MultiJarWorkspace extends ClasspathWorkspace {
+
+        private MultiJarWorkspace(PluginSource source) throws IOException {
+            super(source);
+        }
+
+        @Override
+        protected Map<Path, Map<PluginType, Set<String>>> load(PluginSource source) throws IOException {
+            Map<Path, Map<PluginType, Set<String>>> manifests = super.load(source);
+            // In addition to the normal multi-jar paths, inject a managed jar where we can add manifests.
+            Path managedPath = source.location().resolve(MANAGED_PATH);
+            URL url = managedPath.toUri().toURL();
+            manifests.put(managedPath, loadManifest(jarBaseUrl(url)));
+            return manifests;
+        }
+
+        @Override
+        public void addManifest(PluginType type, String pluginClass) {
+            // Add plugins to the managed manifest
+            manifests.get(location().resolve(MANAGED_PATH)).get(type).add(pluginClass);
+        }
+
+        @Override
+        public void removeManifest(PluginType type, String pluginClass) {
+            // If a plugin appears in multiple manifests, remove it from all of them.
+            for (Map<PluginType, Set<String>> manifestState : manifests.values()) {
+                manifestState.get(type).remove(pluginClass);
+            }
+        }
+
+        @Override
+        public boolean commit(boolean dryRun) throws IOException, TerseException {
+            boolean changed = false;
+            for (Map.Entry<Path, Map<PluginType, Set<String>>> manifestSource : manifests.entrySet()) {
+                Path jarPath = manifestSource.getKey();
+                Map<PluginType, Set<String>> before = initial.get(jarPath);
+                Map<PluginType, Set<String>> after = manifestSource.getValue();
+                if (startSync(dryRun, jarPath, before, after)) {
+                    rewriteJar(dryRun, jarPath, after);
+                    changed = true;
+                }
+            }
+            return changed;
+        }
+    }
+
+    /**
+     * The class hierarchy is similar to the single-jar because there can only be one manifest per type.
+     * However, the path to that single manifest is accessed via the pluginLocation.
+     */
+    private class ClassHierarchyWorkspace extends SingleJarWorkspace {
+
+        private ClassHierarchyWorkspace(PluginSource source) throws IOException {
+            super(source);
+        }
+
+        @Override
+        protected Map<PluginType, Set<String>> load(PluginSource source) throws IOException {
+            return loadManifest(source.location().toUri().toURL());
+        }
+
+        protected boolean commit(boolean dryRun) throws IOException, TerseException {
+            if (startSync(dryRun, location(), initial, manifests)) {
+                rewriteClassHierarchyManifest(dryRun, location(), manifests);
+                return true;
+            }
+            return false;
+        }
+    }
+
+    private boolean startSync(boolean dryRun, Path syncLocation, Map<PluginType, Set<String>> before, Map<PluginType, Set<String>> after) {
+        Objects.requireNonNull(syncLocation, "syncLocation must be non-null");
+        Objects.requireNonNull(before, "before must be non-null");
+        Objects.requireNonNull(after, "after must be non-null");
+        if (before.equals(after)) {
+            return false;
+        }
+        Set<String> added = new HashSet<>();
+        after.values().forEach(added::addAll);
+        before.values().forEach(added::removeAll);
+        Set<String> removed = new HashSet<>();
+        before.values().forEach(removed::addAll);
+        after.values().forEach(removed::removeAll);
+        out.printf("%sSync\t\t%s Add %s Remove %s%n", dryRun ? "Dry Run " : "", syncLocation, added, removed);
+        return true;
+    }
+
+    /**
+     * Rewrite a jar on disk to contain a manifests containing the specified entries.
+     * Will create the jar file and manifests if they do not exist.
+     * Will delete the jar file and manifests if they become empty.
+     *
+     * @param dryRun        True if the rewrite should be applied, false if it should be simulated.
+     * @param jarPath       Path to a jar file for a plugin
+     * @param manifestState Map from plugin type to Class names of plugins which should appear in that manifest
+     */
+    private void rewriteJar(boolean dryRun, Path jarPath, Map<PluginType, Set<String>> manifestState) throws IOException, TerseException {
+        Objects.requireNonNull(jarPath, "jarPath must be non-null");
+        Objects.requireNonNull(manifestState, "manifestState must be non-null");
+        Path writableJar = getWritablePath(dryRun, jarPath);
+        if (!manifestState.values().stream().allMatch(Collection::isEmpty) && !Files.exists(writableJar)) {
+            out.printf("\tCreate\t%s%n", jarPath);
+            createJar(writableJar);
+        }
+        try (FileSystem jar = FileSystems.newFileSystem(
+                new URI("jar", writableJar.toUri().toString(), ""),
+                Collections.emptyMap()
+        )) {
+            Path zipRoot = jar.getRootDirectories().iterator().next();
+            rewriteClassHierarchyManifest(dryRun, zipRoot, manifestState);
+        } catch (URISyntaxException e) {
+            throw new IOException(e);
+        }
+        if (Files.exists(writableJar) && jarIsEmpty(writableJar)) {
+            Files.delete(writableJar);
+        }
+    }
+
+    private void createJar(Path path) throws IOException {
+        Objects.requireNonNull(path, "path must be non-null");
+        try (ZipOutputStream stream = new ZipOutputStream(Files.newOutputStream(
+                path,
+                StandardOpenOption.CREATE,
+                StandardOpenOption.WRITE,
+                StandardOpenOption.TRUNCATE_EXISTING
+        ))) {
+            stream.closeEntry();
+        }
+    }
+
+    private boolean jarIsEmpty(Path path) throws IOException {
+        Objects.requireNonNull(path, "path must be non-null");
+        try (ZipInputStream stream = new ZipInputStream(Files.newInputStream(
+                path,
+                StandardOpenOption.READ
+        ))) {
+            return stream.getNextEntry() == null;
+        }
+    }
+
+    /**
+     * Rewrite a multiple manifest files contained inside a class hierarchy.
+     * Will create the files and parent directories if they not exist.
+     * Will delete the files and parent directories within the plugin location if they become empty.
+     *
+     * @param dryRun         True if the rewrite should be applied, false if it should be simulated.
+     * @param pluginLocation Path to top-level of class hierarchy for a plugin
+     * @param manifestState  Map from plugin type to Class names of plugins which should appear in that manifest
+     */
+    private void rewriteClassHierarchyManifest(boolean dryRun, Path pluginLocation, Map<PluginType, Set<String>> manifestState) throws IOException, TerseException {
+        Objects.requireNonNull(pluginLocation, "pluginLocation must be non-null");
+        Objects.requireNonNull(manifestState, "manifestState must be non-null");
+        if (!Files.exists(pluginLocation)) {
+            throw new TerseException(pluginLocation + " does not exist");
+        }
+        if (!Files.isWritable(pluginLocation)) {
+            throw new TerseException(pluginLocation + " is not writable");
+        }
+        for (Map.Entry<PluginType, Set<String>> manifest : manifestState.entrySet()) {
+            PluginType type = manifest.getKey();
+            Set<String> elements = manifest.getValue();
+            Path manifestPath = pluginLocation.resolve(MANIFEST_PREFIX).resolve(type.superClass().getName());
+            rewriteManifestFile(dryRun, manifestPath, elements);
+        }
+    }
+
+    /**
+     * Rewrite a file, creating it and any parent directories if it does not exist, deleting it if specified

Review Comment:
   This was nonsense. This (and other places where deletes happen) should only delete something if the script would otherwise leave it empty. An empty manifest will be removed, an empty directory will be removed, and an empty jar file will be removed.
   
   I changed it so that `rewriteManifestFile` is only responsible for creating and deleting manifest files, `rewriteClassHierarchyManifest` is responsible for creating and deleting META-INF and services directories. `rewriteJar` was already responsible for creating and deleting jar files.



-- 
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

[GitHub] [kafka] gharris1727 commented on a diff in pull request #14195: KAFKA-15228: Add sync-manifests command to connect-plugin-path (KIP-898)

["gharris1727 (via GitHub)" <gi...@apache.org>]

gharris1727 commented on code in PR #14195:
URL: https://github.com/apache/kafka/pull/14195#discussion_r1294055068


##########
tools/src/test/java/org/apache/kafka/tools/ConnectPluginPathTest.java:
##########
@@ -192,6 +194,51 @@ public void testListMultipleWorkerConfigs(PluginLocationType type) {
                 TestPlugins.TestPlugin.SERVICE_LOADER);
     }
 
+    @ParameterizedTest
+    @EnumSource
+    public void testSyncManifests(PluginLocationType type) {
+        CommandResult res = runCommand(
+                "sync-manifests",
+                "--plugin-location",
+                setupLocation(workspace.resolve("location-a"), type, TestPlugins.TestPlugin.NON_MIGRATED_CONVERTER)
+        );
+        assertEquals(0, res.returnCode);
+        assertScanResult(true, TestPlugins.TestPlugin.NON_MIGRATED_CONVERTER, res.reflective);
+        assertScanResult(true, TestPlugins.TestPlugin.NON_MIGRATED_CONVERTER, res.serviceLoading);
+    }
+
+    @ParameterizedTest
+    @EnumSource
+    public void testSyncManifestsDryRun(PluginLocationType type) {
+        CommandResult res = runCommand(
+                "sync-manifests",
+                "--plugin-location",
+                setupLocation(workspace.resolve("location-a"), type, TestPlugins.TestPlugin.NON_MIGRATED_CONVERTER),
+                "--dry-run"
+        );
+        assertEquals(0, res.returnCode);
+        assertScanResult(true, TestPlugins.TestPlugin.NON_MIGRATED_CONVERTER, res.reflective);
+        assertScanResult(false, TestPlugins.TestPlugin.NON_MIGRATED_CONVERTER, res.serviceLoading);
+    }
+
+    @ParameterizedTest
+    @EnumSource
+    public void testSyncManifestsKeepNotFound(PluginLocationType type) {
+        CommandResult res = runCommand(
+                "sync-manifests",
+                "--plugin-location",
+                setupLocation(workspace.resolve("location-a"), type, TestPlugins.TestPlugin.BAD_PACKAGING_STATIC_INITIALIZER_THROWS_REST_EXTENSION),
+                "--plugin-location",
+                setupLocation(workspace.resolve("location-b"), type, TestPlugins.TestPlugin.NON_MIGRATED_CONVERTER),
+                "--keep-not-found"
+        );
+        assertEquals(0, res.returnCode);
+        assertScanResult(true, TestPlugins.TestPlugin.NON_MIGRATED_CONVERTER, res.reflective);
+        assertScanResult(true, TestPlugins.TestPlugin.NON_MIGRATED_CONVERTER, res.serviceLoading);
+        assertScanResult(false, TestPlugins.TestPlugin.BAD_PACKAGING_STATIC_INITIALIZER_THROWS_REST_EXTENSION, res.reflective);
+        assertScanResult(false, TestPlugins.TestPlugin.BAD_PACKAGING_STATIC_INITIALIZER_THROWS_REST_EXTENSION, res.serviceLoading);

Review Comment:
   You're right, it was very annoying to read the manifest files for assertions due to the parameterization of the test.
   
   I added assertions to each of these `sync-manifests` tests that runs a `list` afterwards and asserts on the result, which is how users will commonly verify the result of a `sync-manifests` command.



-- 
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

[GitHub] [kafka] gharris1727 commented on a diff in pull request #14195: KAFKA-15228: Add sync-manifests command to connect-plugin-path (KIP-898)

["gharris1727 (via GitHub)" <gi...@apache.org>]

gharris1727 commented on code in PR #14195:
URL: https://github.com/apache/kafka/pull/14195#discussion_r1294083162


##########
tools/src/main/java/org/apache/kafka/tools/ConnectPluginPath.java:
##########
@@ -368,6 +391,30 @@ private static void endCommand(
             config.out.printf("Total plugins:      \t%d%n", totalPlugins);
             config.out.printf("Loadable plugins:   \t%d%n", loadablePlugins);
             config.out.printf("Compatible plugins: \t%d%n", compatiblePlugins);
+        } else if (config.command == Command.SYNC_MANIFESTS) {
+            if (workspace.commit(true)) {
+                if (config.dryRun) {
+                    config.out.println("Dry run passed: All above changes can be committed to disk if re-run with dry run disabled.");
+                } else {
+                    config.out.println("Writing changes to plugins...");
+                    workspace.commit(false);
+                    config.out.println("All plugins have accurate ServiceLoader manifests");
+                }
+            } else {
+                config.out.println("No changes required.");
+            }
+        }
+    }
+
+    private static void failCommand(Config config, Throwable e) {
+        if (config.command == Command.LIST) {
+            throw new RuntimeException("Unexpected error occurred while listing plugins", e);
+        } else if (config.command == Command.SYNC_MANIFESTS) {
+            if (config.dryRun) {
+                throw new RuntimeException("Unexpected error occurred while dry-running sync", e);
+            } else {
+                config.out.println("Connect plugin path now in unexpected state: Clear your plugin path and retry with dry run enabled");

Review Comment:
   Yes this error stacktrace should be visible. I've changed the error handling to only inject the warning about the corrupted plugin path when exceptions come from `commit(false)`. For unpredictable errors (mostly IOExceptions) I expect the stack trace (on stderr) to look like:
   
   ```
   RuntimeException: Unexpected error occurred while executing sync
   ...
   caused by:
   RuntimeException: Sync incomplete, plugin path may be corrupted. Clear your plugin path and retry with dry-run enabled
   ...
   caused by:
   IOException: filesystem broke
   ```
   
   This doesn't emphasize the corruption warning, but does make it show up under more accurate conditions.
   
   Overall I don't like the error handling here but I'm not sure what needs to change. Let me know if you have any more ideas.
   



-- 
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

[GitHub] [kafka] C0urante commented on a diff in pull request #14195: KAFKA-15228: Add sync-manifests command to connect-plugin-path (KIP-898)

["C0urante (via GitHub)" <gi...@apache.org>]

C0urante commented on code in PR #14195:
URL: https://github.com/apache/kafka/pull/14195#discussion_r1294750935


##########
tools/src/main/java/org/apache/kafka/tools/ManifestWorkspace.java:
##########
@@ -0,0 +1,573 @@
+/*
+ * 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.tools;
+
+import org.apache.kafka.connect.runtime.isolation.PluginSource;
+import org.apache.kafka.connect.runtime.isolation.PluginType;
+
+import java.io.BufferedOutputStream;
+import java.io.BufferedReader;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.InputStreamReader;
+import java.io.OutputStream;
+import java.io.PrintStream;
+import java.net.MalformedURLException;
+import java.net.URI;
+import java.net.URISyntaxException;
+import java.net.URL;
+import java.net.URLConnection;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.FileSystem;
+import java.nio.file.FileSystems;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+import java.nio.file.StandardCopyOption;
+import java.nio.file.StandardOpenOption;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.EnumMap;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.LinkedHashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Set;
+import java.util.function.BiConsumer;
+import java.util.zip.ZipInputStream;
+import java.util.zip.ZipOutputStream;
+
+/**
+ * An in-memory workspace for manipulating {@link java.util.ServiceLoader} manifest files.
+ * <p>Use {@link #forSource(PluginSource)} to get a workspace scoped to a single plugin location, which is able
+ * to accept simulated reads and writes of manifests.
+ * Write the simulated changes to disk via {@link #commit(boolean)}.
+ */
+public class ManifestWorkspace {
+
+    private static final String MANIFEST_PREFIX = "META-INF/services/";
+    private static final Path MANAGED_PATH = Paths.get("connect-plugin-path-shim-0.0.1-SNAPSHOT.jar");
+    private final PrintStream out;
+    private final List<SourceWorkspace<?>> workspaces;
+    private final Map<Path, Path> temporaryOverlayFiles;
+
+    public ManifestWorkspace(PrintStream out) {
+        this.out = out;
+        workspaces = new ArrayList<>();
+        temporaryOverlayFiles = new HashMap<>();
+    }
+
+    public SourceWorkspace<?> forSource(PluginSource source) throws IOException {
+        SourceWorkspace<?> sourceWorkspace;
+        switch (source.type()) {
+            case CLASSPATH:
+                sourceWorkspace = new ClasspathWorkspace(source);
+                break;
+            case MULTI_JAR:
+                sourceWorkspace = new MultiJarWorkspace(source);
+                break;
+            case SINGLE_JAR:
+                sourceWorkspace = new SingleJarWorkspace(source);
+                break;
+            case CLASS_HIERARCHY:
+                sourceWorkspace = new ClassHierarchyWorkspace(source);
+                break;
+            default:
+                throw new IllegalStateException("Unknown source type " + source.type());
+        }
+        workspaces.add(sourceWorkspace);
+        return sourceWorkspace;
+    }
+
+    /**
+     * Commits all queued changes to disk
+     * @return true if any workspace wrote changes to disk, false if all workspaces did not have writes to apply
+     * @throws IOException if an error occurs reading or writing to the filesystem
+     * @throws TerseException if a path is not writable on disk and should be.
+     */
+    public boolean commit(boolean dryRun) throws IOException, TerseException {
+        boolean changed = false;
+        for (SourceWorkspace<?> workspace : workspaces) {
+            changed |= workspace.commit(dryRun);
+        }
+        return changed;
+    }
+
+    /**
+     * A workspace scoped to a single plugin source.
+     * <p>Buffers simulated reads and writes to the plugin path before they can be written to disk.
+     * @param <T> The data structure used by the workspace to store in-memory manifests internally.
+     */
+    public static abstract class SourceWorkspace<T> {
+        private final Path location;
+        private final PluginSource.Type type;
+        protected final T initial;
+        protected final T manifests;
+
+        private SourceWorkspace(PluginSource source) throws IOException {
+            this.location = source.location();
+            this.type = source.type();
+            this.initial = load(source);
+            this.manifests = load(source);
+        }
+
+        public Path location() {
+            return location;
+        }
+
+        public PluginSource.Type type() {
+            return type;
+        }
+
+        protected abstract T load(PluginSource source) throws IOException;
+
+        public abstract boolean hasManifest(PluginType type, String className);
+
+        public abstract void forEach(BiConsumer<String, PluginType> consumer);
+
+        public abstract void addManifest(PluginType type, String pluginClass);
+
+        public abstract void removeManifest(PluginType type, String pluginClass);
+
+        protected abstract boolean commit(boolean dryRun) throws TerseException, IOException;
+
+        protected static Map<PluginType, Set<String>> loadManifest(URL baseUrl) throws MalformedURLException {
+            Map<PluginType, Set<String>> manifests = new EnumMap<>(PluginType.class);
+            for (PluginType type : PluginType.values()) {
+                Set<String> result;
+                try {
+                    URL u = new URL(baseUrl, MANIFEST_PREFIX + type.superClass().getName());
+                    result = parse(u);
+                } catch (RuntimeException e) {
+                    result = new LinkedHashSet<>();
+                }
+                manifests.put(type, result);
+            }
+            return manifests;
+        }
+
+        protected static URL jarBaseUrl(URL fileUrl) throws MalformedURLException {
+            return new URL("jar", "", -1, fileUrl + "!/", null);
+        }
+
+        protected static void forEach(Map<PluginType, Set<String>> manifests, BiConsumer<String, PluginType> consumer) {
+            manifests.forEach((type, classNames) -> classNames.forEach(className -> consumer.accept(className, type)));
+        }
+    }
+
+    /**
+     * A single jar can only contain one manifest per plugin type.
+     */
+    private class SingleJarWorkspace extends SourceWorkspace<Map<PluginType, Set<String>>> {
+
+        private SingleJarWorkspace(PluginSource source) throws IOException {
+            super(source);
+        }
+
+        @Override
+        protected Map<PluginType, Set<String>> load(PluginSource source) throws IOException {
+            return loadManifest(jarBaseUrl(source.urls()[0]));
+        }
+
+        @Override
+        public boolean hasManifest(PluginType type, String className) {
+            return manifests.get(type).contains(className);
+        }
+
+        @Override
+        public void forEach(BiConsumer<String, PluginType> consumer) {
+            forEach(manifests, consumer);
+        }
+
+        @Override
+        public void addManifest(PluginType type, String pluginClass) {
+            manifests.get(type).add(pluginClass);
+        }
+
+        @Override
+        public void removeManifest(PluginType type, String pluginClass) {
+            manifests.get(type).remove(pluginClass);
+        }
+
+        @Override
+        protected boolean commit(boolean dryRun) throws IOException, TerseException {
+            if (startSync(dryRun, location(), initial, manifests)) {
+                rewriteJar(dryRun, location(), manifests);
+                return true;
+            }
+            return false;
+        }
+    }
+
+    /**
+     * A classpath workspace is backed by multiple jars, and that is not writable.
+     * The in-memory format is a map from jar path to the manifests contained in that jar.
+     * The control flow of the caller should not perform writes, so these exceptions indicate a bug in the program.
+     */
+    private class ClasspathWorkspace extends SourceWorkspace<Map<Path, Map<PluginType, Set<String>>>> {
+
+        private ClasspathWorkspace(PluginSource source) throws IOException {
+            super(source);
+        }
+
+        @Override
+        protected Map<Path, Map<PluginType, Set<String>>> load(PluginSource source) throws IOException {
+            Map<Path, Map<PluginType, Set<String>>> manifestsBySubLocation = new HashMap<>();
+            for (URL url : source.urls()) {
+                Path jarPath = Paths.get(url.getPath());
+                manifestsBySubLocation.put(jarPath, loadManifest(jarBaseUrl(url)));
+            }
+            return manifestsBySubLocation;
+        }
+
+        public boolean hasManifest(PluginType type, String className) {
+            return manifests.values()
+                    .stream()
+                    .map(m -> m.get(type))
+                    .anyMatch(s -> s.contains(className));
+        }
+
+        public void forEach(BiConsumer<String, PluginType> consumer) {
+            manifests.values().forEach(m -> forEach(m, consumer));
+        }
+
+        @Override
+        public void addManifest(PluginType type, String pluginClass) {
+            throw new UnsupportedOperationException("Cannot change the contents of the classpath");
+        }
+
+        @Override
+        public void removeManifest(PluginType type, String pluginClass) {
+            throw new UnsupportedOperationException("Cannot change the contents of the classpath");
+        }
+
+        @Override
+        protected boolean commit(boolean dryRun) throws IOException, TerseException {
+            // There is never anything to commit for the classpath
+            return false;
+        }
+    }
+
+    /**
+     * A multi-jar workspace is similar to the classpath workspace because it has multiple jars.
+     * However, the multi-jar workspace is writable, and injects a managed jar where it writes added manifests.
+     */
+    private class MultiJarWorkspace extends ClasspathWorkspace {
+
+        private MultiJarWorkspace(PluginSource source) throws IOException {
+            super(source);
+        }
+
+        @Override
+        protected Map<Path, Map<PluginType, Set<String>>> load(PluginSource source) throws IOException {
+            Map<Path, Map<PluginType, Set<String>>> manifests = super.load(source);
+            // In addition to the normal multi-jar paths, inject a managed jar where we can add manifests.
+            Path managedPath = source.location().resolve(MANAGED_PATH);
+            URL url = managedPath.toUri().toURL();
+            manifests.put(managedPath, loadManifest(jarBaseUrl(url)));
+            return manifests;
+        }
+
+        @Override
+        public void addManifest(PluginType type, String pluginClass) {
+            // Add plugins to the managed manifest
+            manifests.get(location().resolve(MANAGED_PATH)).get(type).add(pluginClass);
+        }
+
+        @Override
+        public void removeManifest(PluginType type, String pluginClass) {
+            // If a plugin appears in multiple manifests, remove it from all of them.
+            for (Map<PluginType, Set<String>> manifestState : manifests.values()) {
+                manifestState.get(type).remove(pluginClass);
+            }
+        }
+
+        @Override
+        public boolean commit(boolean dryRun) throws IOException, TerseException {
+            boolean changed = false;
+            for (Map.Entry<Path, Map<PluginType, Set<String>>> manifestSource : manifests.entrySet()) {
+                Path jarPath = manifestSource.getKey();
+                Map<PluginType, Set<String>> before = initial.get(jarPath);
+                Map<PluginType, Set<String>> after = manifestSource.getValue();
+                if (startSync(dryRun, jarPath, before, after)) {
+                    rewriteJar(dryRun, jarPath, after);
+                    changed = true;
+                }
+            }
+            return changed;
+        }
+    }
+
+    /**
+     * The class hierarchy is similar to the single-jar because there can only be one manifest per type.
+     * However, the path to that single manifest is accessed via the pluginLocation.
+     */
+    private class ClassHierarchyWorkspace extends SingleJarWorkspace {
+
+        private ClassHierarchyWorkspace(PluginSource source) throws IOException {
+            super(source);
+        }
+
+        @Override
+        protected Map<PluginType, Set<String>> load(PluginSource source) throws IOException {
+            return loadManifest(source.location().toUri().toURL());
+        }
+
+        protected boolean commit(boolean dryRun) throws IOException, TerseException {
+            if (startSync(dryRun, location(), initial, manifests)) {
+                rewriteClassHierarchyManifest(dryRun, location(), manifests);
+                return true;
+            }
+            return false;
+        }
+    }
+
+    private boolean startSync(boolean dryRun, Path syncLocation, Map<PluginType, Set<String>> before, Map<PluginType, Set<String>> after) {
+        Objects.requireNonNull(syncLocation, "syncLocation must be non-null");
+        Objects.requireNonNull(before, "before must be non-null");
+        Objects.requireNonNull(after, "after must be non-null");
+        if (before.equals(after)) {
+            return false;
+        }
+        Set<String> added = new HashSet<>();
+        after.values().forEach(added::addAll);
+        before.values().forEach(added::removeAll);
+        Set<String> removed = new HashSet<>();
+        before.values().forEach(removed::addAll);
+        after.values().forEach(removed::removeAll);
+        out.printf("%sSync\t\t%s Add %s Remove %s%n", dryRun ? "Dry Run " : "", syncLocation, added, removed);
+        return true;
+    }
+
+    /**
+     * Rewrite a jar on disk to contain a manifests containing the specified entries.
+     * Will create the jar file and manifests if they do not exist.
+     * Will delete the jar file and manifests if they become empty.
+     *
+     * @param dryRun        True if the rewrite should be applied, false if it should be simulated.
+     * @param jarPath       Path to a jar file for a plugin
+     * @param manifestState Map from plugin type to Class names of plugins which should appear in that manifest
+     */
+    private void rewriteJar(boolean dryRun, Path jarPath, Map<PluginType, Set<String>> manifestState) throws IOException, TerseException {
+        Objects.requireNonNull(jarPath, "jarPath must be non-null");
+        Objects.requireNonNull(manifestState, "manifestState must be non-null");
+        Path writableJar = getWritablePath(dryRun, jarPath);
+        if (!manifestState.values().stream().allMatch(Collection::isEmpty) && !Files.exists(writableJar)) {
+            out.printf("\tCreate\t%s%n", jarPath);
+            createJar(writableJar);
+        }
+        try (FileSystem jar = FileSystems.newFileSystem(
+                new URI("jar", writableJar.toUri().toString(), ""),
+                Collections.emptyMap()
+        )) {
+            Path zipRoot = jar.getRootDirectories().iterator().next();
+            rewriteClassHierarchyManifest(dryRun, zipRoot, manifestState);
+        } catch (URISyntaxException e) {
+            throw new IOException(e);
+        }
+        if (Files.exists(writableJar) && jarIsEmpty(writableJar)) {
+            Files.delete(writableJar);
+        }
+    }
+
+    private void createJar(Path path) throws IOException {
+        Objects.requireNonNull(path, "path must be non-null");
+        try (ZipOutputStream stream = new ZipOutputStream(Files.newOutputStream(
+                path,
+                StandardOpenOption.CREATE,
+                StandardOpenOption.WRITE,
+                StandardOpenOption.TRUNCATE_EXISTING
+        ))) {
+            stream.closeEntry();
+        }
+    }
+
+    private boolean jarIsEmpty(Path path) throws IOException {
+        Objects.requireNonNull(path, "path must be non-null");
+        try (ZipInputStream stream = new ZipInputStream(Files.newInputStream(
+                path,
+                StandardOpenOption.READ
+        ))) {
+            return stream.getNextEntry() == null;
+        }
+    }
+
+    /**
+     * Rewrite a multiple manifest files contained inside a class hierarchy.
+     * Will create the files and parent directories if they not exist.
+     * Will delete the files and parent directories within the plugin location if they become empty.
+     *
+     * @param dryRun         True if the rewrite should be applied, false if it should be simulated.
+     * @param pluginLocation Path to top-level of class hierarchy for a plugin
+     * @param manifestState  Map from plugin type to Class names of plugins which should appear in that manifest
+     */
+    private void rewriteClassHierarchyManifest(boolean dryRun, Path pluginLocation, Map<PluginType, Set<String>> manifestState) throws IOException, TerseException {
+        Objects.requireNonNull(pluginLocation, "pluginLocation must be non-null");
+        Objects.requireNonNull(manifestState, "manifestState must be non-null");
+        if (!Files.exists(pluginLocation)) {
+            throw new TerseException(pluginLocation + " does not exist");
+        }
+        if (!Files.isWritable(pluginLocation)) {
+            throw new TerseException(pluginLocation + " is not writable");
+        }
+        for (Map.Entry<PluginType, Set<String>> manifest : manifestState.entrySet()) {
+            PluginType type = manifest.getKey();
+            Set<String> elements = manifest.getValue();
+            Path manifestPath = pluginLocation.resolve(MANIFEST_PREFIX).resolve(type.superClass().getName());
+            rewriteManifestFile(dryRun, manifestPath, elements);
+        }
+    }
+
+    /**
+     * Rewrite a file, creating it and any parent directories if it does not exist, deleting it if specified
+     *
+     * @param dryRun              True if the rewrite should be applied, false if it should be simulated.
+     * @param filePath            Path to file which should be rewritten.
+     * @param elements            Class names of plugins which should appear in the manifest
+     */
+    private void rewriteManifestFile(boolean dryRun, Path filePath, Set<String> elements) throws IOException, TerseException {
+        Objects.requireNonNull(filePath, "filePath must be non-null");
+        Objects.requireNonNull(elements, "elements must be non-null");
+        Path writableFile = getWritablePath(dryRun, filePath);
+        if (elements.isEmpty()) {
+            if (Files.exists(filePath)) {
+                out.printf("\tDelete\t%s%n", filePath);
+                if (!dryRun) {
+                    Files.delete(writableFile);
+                }
+            }
+            // TODO: delete parent directories if empty up to the containing directory
+        } else {
+            Path parent = filePath.getParent();
+            if (parent != null) {
+                if (Files.exists(parent)) {
+                    if (!Files.isWritable(parent)) {
+                        throw new TerseException(parent + " is not writable");
+                    }
+                } else {
+                    out.printf("\tCreate\t%s%n", parent);
+                    if (!dryRun) {
+                        Files.createDirectories(parent);
+                    }
+                }
+            }
+            if (!Files.exists(filePath)) {
+                out.printf("\tCreate\t%s%n", filePath);
+            }
+            out.printf("\tWrite\t%s with content %s%n", filePath, elements);
+            if (!dryRun) {
+                try (OutputStream stream = new BufferedOutputStream(Files.newOutputStream(
+                        writableFile,
+                        StandardOpenOption.CREATE,
+                        StandardOpenOption.WRITE,
+                        StandardOpenOption.TRUNCATE_EXISTING

Review Comment:
   Great, thanks 👍



##########
tools/src/main/java/org/apache/kafka/tools/ManifestWorkspace.java:
##########
@@ -0,0 +1,573 @@
+/*
+ * 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.tools;
+
+import org.apache.kafka.connect.runtime.isolation.PluginSource;
+import org.apache.kafka.connect.runtime.isolation.PluginType;
+
+import java.io.BufferedOutputStream;
+import java.io.BufferedReader;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.InputStreamReader;
+import java.io.OutputStream;
+import java.io.PrintStream;
+import java.net.MalformedURLException;
+import java.net.URI;
+import java.net.URISyntaxException;
+import java.net.URL;
+import java.net.URLConnection;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.FileSystem;
+import java.nio.file.FileSystems;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+import java.nio.file.StandardCopyOption;
+import java.nio.file.StandardOpenOption;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.EnumMap;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.LinkedHashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Set;
+import java.util.function.BiConsumer;
+import java.util.zip.ZipInputStream;
+import java.util.zip.ZipOutputStream;
+
+/**
+ * An in-memory workspace for manipulating {@link java.util.ServiceLoader} manifest files.
+ * <p>Use {@link #forSource(PluginSource)} to get a workspace scoped to a single plugin location, which is able
+ * to accept simulated reads and writes of manifests.
+ * Write the simulated changes to disk via {@link #commit(boolean)}.
+ */
+public class ManifestWorkspace {
+
+    private static final String MANIFEST_PREFIX = "META-INF/services/";
+    private static final Path MANAGED_PATH = Paths.get("connect-plugin-path-shim-0.0.1-SNAPSHOT.jar");
+    private final PrintStream out;
+    private final List<SourceWorkspace<?>> workspaces;
+    private final Map<Path, Path> temporaryOverlayFiles;
+
+    public ManifestWorkspace(PrintStream out) {
+        this.out = out;
+        workspaces = new ArrayList<>();
+        temporaryOverlayFiles = new HashMap<>();
+    }
+
+    public SourceWorkspace<?> forSource(PluginSource source) throws IOException {
+        SourceWorkspace<?> sourceWorkspace;
+        switch (source.type()) {
+            case CLASSPATH:
+                sourceWorkspace = new ClasspathWorkspace(source);
+                break;
+            case MULTI_JAR:
+                sourceWorkspace = new MultiJarWorkspace(source);
+                break;
+            case SINGLE_JAR:
+                sourceWorkspace = new SingleJarWorkspace(source);
+                break;
+            case CLASS_HIERARCHY:
+                sourceWorkspace = new ClassHierarchyWorkspace(source);
+                break;
+            default:
+                throw new IllegalStateException("Unknown source type " + source.type());
+        }
+        workspaces.add(sourceWorkspace);
+        return sourceWorkspace;
+    }
+
+    /**
+     * Commits all queued changes to disk
+     * @return true if any workspace wrote changes to disk, false if all workspaces did not have writes to apply
+     * @throws IOException if an error occurs reading or writing to the filesystem
+     * @throws TerseException if a path is not writable on disk and should be.
+     */
+    public boolean commit(boolean dryRun) throws IOException, TerseException {
+        boolean changed = false;
+        for (SourceWorkspace<?> workspace : workspaces) {
+            changed |= workspace.commit(dryRun);
+        }
+        return changed;
+    }
+
+    /**
+     * A workspace scoped to a single plugin source.
+     * <p>Buffers simulated reads and writes to the plugin path before they can be written to disk.
+     * @param <T> The data structure used by the workspace to store in-memory manifests internally.
+     */
+    public static abstract class SourceWorkspace<T> {
+        private final Path location;
+        private final PluginSource.Type type;
+        protected final T initial;
+        protected final T manifests;
+
+        private SourceWorkspace(PluginSource source) throws IOException {
+            this.location = source.location();
+            this.type = source.type();
+            this.initial = load(source);
+            this.manifests = load(source);
+        }
+
+        public Path location() {
+            return location;
+        }
+
+        public PluginSource.Type type() {
+            return type;
+        }
+
+        protected abstract T load(PluginSource source) throws IOException;
+
+        public abstract boolean hasManifest(PluginType type, String className);
+
+        public abstract void forEach(BiConsumer<String, PluginType> consumer);
+
+        public abstract void addManifest(PluginType type, String pluginClass);
+
+        public abstract void removeManifest(PluginType type, String pluginClass);
+
+        protected abstract boolean commit(boolean dryRun) throws TerseException, IOException;
+
+        protected static Map<PluginType, Set<String>> loadManifest(URL baseUrl) throws MalformedURLException {
+            Map<PluginType, Set<String>> manifests = new EnumMap<>(PluginType.class);
+            for (PluginType type : PluginType.values()) {
+                Set<String> result;
+                try {
+                    URL u = new URL(baseUrl, MANIFEST_PREFIX + type.superClass().getName());
+                    result = parse(u);
+                } catch (RuntimeException e) {
+                    result = new LinkedHashSet<>();
+                }
+                manifests.put(type, result);
+            }
+            return manifests;
+        }
+
+        protected static URL jarBaseUrl(URL fileUrl) throws MalformedURLException {
+            return new URL("jar", "", -1, fileUrl + "!/", null);
+        }
+
+        protected static void forEach(Map<PluginType, Set<String>> manifests, BiConsumer<String, PluginType> consumer) {
+            manifests.forEach((type, classNames) -> classNames.forEach(className -> consumer.accept(className, type)));
+        }
+    }
+
+    /**
+     * A single jar can only contain one manifest per plugin type.
+     */
+    private class SingleJarWorkspace extends SourceWorkspace<Map<PluginType, Set<String>>> {
+
+        private SingleJarWorkspace(PluginSource source) throws IOException {
+            super(source);
+        }
+
+        @Override
+        protected Map<PluginType, Set<String>> load(PluginSource source) throws IOException {
+            return loadManifest(jarBaseUrl(source.urls()[0]));
+        }
+
+        @Override
+        public boolean hasManifest(PluginType type, String className) {
+            return manifests.get(type).contains(className);
+        }
+
+        @Override
+        public void forEach(BiConsumer<String, PluginType> consumer) {
+            forEach(manifests, consumer);
+        }
+
+        @Override
+        public void addManifest(PluginType type, String pluginClass) {
+            manifests.get(type).add(pluginClass);
+        }
+
+        @Override
+        public void removeManifest(PluginType type, String pluginClass) {
+            manifests.get(type).remove(pluginClass);
+        }
+
+        @Override
+        protected boolean commit(boolean dryRun) throws IOException, TerseException {
+            if (startSync(dryRun, location(), initial, manifests)) {
+                rewriteJar(dryRun, location(), manifests);
+                return true;
+            }
+            return false;
+        }
+    }
+
+    /**
+     * A classpath workspace is backed by multiple jars, and that is not writable.
+     * The in-memory format is a map from jar path to the manifests contained in that jar.
+     * The control flow of the caller should not perform writes, so these exceptions indicate a bug in the program.
+     */
+    private class ClasspathWorkspace extends SourceWorkspace<Map<Path, Map<PluginType, Set<String>>>> {
+
+        private ClasspathWorkspace(PluginSource source) throws IOException {
+            super(source);
+        }
+
+        @Override
+        protected Map<Path, Map<PluginType, Set<String>>> load(PluginSource source) throws IOException {
+            Map<Path, Map<PluginType, Set<String>>> manifestsBySubLocation = new HashMap<>();
+            for (URL url : source.urls()) {
+                Path jarPath = Paths.get(url.getPath());
+                manifestsBySubLocation.put(jarPath, loadManifest(jarBaseUrl(url)));
+            }
+            return manifestsBySubLocation;
+        }
+
+        public boolean hasManifest(PluginType type, String className) {
+            return manifests.values()
+                    .stream()
+                    .map(m -> m.get(type))
+                    .anyMatch(s -> s.contains(className));
+        }
+
+        public void forEach(BiConsumer<String, PluginType> consumer) {
+            manifests.values().forEach(m -> forEach(m, consumer));
+        }
+
+        @Override
+        public void addManifest(PluginType type, String pluginClass) {
+            throw new UnsupportedOperationException("Cannot change the contents of the classpath");
+        }
+
+        @Override
+        public void removeManifest(PluginType type, String pluginClass) {
+            throw new UnsupportedOperationException("Cannot change the contents of the classpath");
+        }
+
+        @Override
+        protected boolean commit(boolean dryRun) throws IOException, TerseException {
+            // There is never anything to commit for the classpath
+            return false;
+        }
+    }
+
+    /**
+     * A multi-jar workspace is similar to the classpath workspace because it has multiple jars.
+     * However, the multi-jar workspace is writable, and injects a managed jar where it writes added manifests.
+     */
+    private class MultiJarWorkspace extends ClasspathWorkspace {
+
+        private MultiJarWorkspace(PluginSource source) throws IOException {
+            super(source);
+        }
+
+        @Override
+        protected Map<Path, Map<PluginType, Set<String>>> load(PluginSource source) throws IOException {
+            Map<Path, Map<PluginType, Set<String>>> manifests = super.load(source);
+            // In addition to the normal multi-jar paths, inject a managed jar where we can add manifests.
+            Path managedPath = source.location().resolve(MANAGED_PATH);
+            URL url = managedPath.toUri().toURL();
+            manifests.put(managedPath, loadManifest(jarBaseUrl(url)));
+            return manifests;
+        }
+
+        @Override
+        public void addManifest(PluginType type, String pluginClass) {
+            // Add plugins to the managed manifest
+            manifests.get(location().resolve(MANAGED_PATH)).get(type).add(pluginClass);
+        }
+
+        @Override
+        public void removeManifest(PluginType type, String pluginClass) {
+            // If a plugin appears in multiple manifests, remove it from all of them.
+            for (Map<PluginType, Set<String>> manifestState : manifests.values()) {
+                manifestState.get(type).remove(pluginClass);
+            }
+        }
+
+        @Override
+        public boolean commit(boolean dryRun) throws IOException, TerseException {
+            boolean changed = false;
+            for (Map.Entry<Path, Map<PluginType, Set<String>>> manifestSource : manifests.entrySet()) {
+                Path jarPath = manifestSource.getKey();
+                Map<PluginType, Set<String>> before = initial.get(jarPath);
+                Map<PluginType, Set<String>> after = manifestSource.getValue();
+                if (startSync(dryRun, jarPath, before, after)) {
+                    rewriteJar(dryRun, jarPath, after);
+                    changed = true;
+                }
+            }
+            return changed;
+        }
+    }
+
+    /**
+     * The class hierarchy is similar to the single-jar because there can only be one manifest per type.
+     * However, the path to that single manifest is accessed via the pluginLocation.
+     */
+    private class ClassHierarchyWorkspace extends SingleJarWorkspace {
+
+        private ClassHierarchyWorkspace(PluginSource source) throws IOException {
+            super(source);
+        }
+
+        @Override
+        protected Map<PluginType, Set<String>> load(PluginSource source) throws IOException {
+            return loadManifest(source.location().toUri().toURL());
+        }
+
+        protected boolean commit(boolean dryRun) throws IOException, TerseException {
+            if (startSync(dryRun, location(), initial, manifests)) {
+                rewriteClassHierarchyManifest(dryRun, location(), manifests);
+                return true;
+            }
+            return false;
+        }
+    }
+
+    private boolean startSync(boolean dryRun, Path syncLocation, Map<PluginType, Set<String>> before, Map<PluginType, Set<String>> after) {
+        Objects.requireNonNull(syncLocation, "syncLocation must be non-null");
+        Objects.requireNonNull(before, "before must be non-null");
+        Objects.requireNonNull(after, "after must be non-null");
+        if (before.equals(after)) {
+            return false;
+        }
+        Set<String> added = new HashSet<>();
+        after.values().forEach(added::addAll);
+        before.values().forEach(added::removeAll);
+        Set<String> removed = new HashSet<>();
+        before.values().forEach(removed::addAll);
+        after.values().forEach(removed::removeAll);
+        out.printf("%sSync\t\t%s Add %s Remove %s%n", dryRun ? "Dry Run " : "", syncLocation, added, removed);
+        return true;
+    }
+
+    /**
+     * Rewrite a jar on disk to contain a manifests containing the specified entries.
+     * Will create the jar file and manifests if they do not exist.
+     * Will delete the jar file and manifests if they become empty.
+     *
+     * @param dryRun        True if the rewrite should be applied, false if it should be simulated.
+     * @param jarPath       Path to a jar file for a plugin
+     * @param manifestState Map from plugin type to Class names of plugins which should appear in that manifest
+     */
+    private void rewriteJar(boolean dryRun, Path jarPath, Map<PluginType, Set<String>> manifestState) throws IOException, TerseException {
+        Objects.requireNonNull(jarPath, "jarPath must be non-null");
+        Objects.requireNonNull(manifestState, "manifestState must be non-null");
+        Path writableJar = getWritablePath(dryRun, jarPath);
+        if (!manifestState.values().stream().allMatch(Collection::isEmpty) && !Files.exists(writableJar)) {
+            out.printf("\tCreate\t%s%n", jarPath);
+            createJar(writableJar);
+        }
+        try (FileSystem jar = FileSystems.newFileSystem(
+                new URI("jar", writableJar.toUri().toString(), ""),
+                Collections.emptyMap()
+        )) {
+            Path zipRoot = jar.getRootDirectories().iterator().next();
+            rewriteClassHierarchyManifest(dryRun, zipRoot, manifestState);
+        } catch (URISyntaxException e) {
+            throw new IOException(e);
+        }
+        if (Files.exists(writableJar) && jarIsEmpty(writableJar)) {
+            Files.delete(writableJar);
+        }
+    }
+
+    private void createJar(Path path) throws IOException {
+        Objects.requireNonNull(path, "path must be non-null");
+        try (ZipOutputStream stream = new ZipOutputStream(Files.newOutputStream(
+                path,
+                StandardOpenOption.CREATE,
+                StandardOpenOption.WRITE,
+                StandardOpenOption.TRUNCATE_EXISTING
+        ))) {
+            stream.closeEntry();
+        }
+    }
+
+    private boolean jarIsEmpty(Path path) throws IOException {
+        Objects.requireNonNull(path, "path must be non-null");
+        try (ZipInputStream stream = new ZipInputStream(Files.newInputStream(
+                path,
+                StandardOpenOption.READ
+        ))) {
+            return stream.getNextEntry() == null;
+        }
+    }
+
+    /**
+     * Rewrite a multiple manifest files contained inside a class hierarchy.
+     * Will create the files and parent directories if they not exist.
+     * Will delete the files and parent directories within the plugin location if they become empty.
+     *
+     * @param dryRun         True if the rewrite should be applied, false if it should be simulated.
+     * @param pluginLocation Path to top-level of class hierarchy for a plugin
+     * @param manifestState  Map from plugin type to Class names of plugins which should appear in that manifest
+     */
+    private void rewriteClassHierarchyManifest(boolean dryRun, Path pluginLocation, Map<PluginType, Set<String>> manifestState) throws IOException, TerseException {
+        Objects.requireNonNull(pluginLocation, "pluginLocation must be non-null");
+        Objects.requireNonNull(manifestState, "manifestState must be non-null");
+        if (!Files.exists(pluginLocation)) {
+            throw new TerseException(pluginLocation + " does not exist");
+        }
+        if (!Files.isWritable(pluginLocation)) {
+            throw new TerseException(pluginLocation + " is not writable");
+        }
+        for (Map.Entry<PluginType, Set<String>> manifest : manifestState.entrySet()) {
+            PluginType type = manifest.getKey();
+            Set<String> elements = manifest.getValue();
+            Path manifestPath = pluginLocation.resolve(MANIFEST_PREFIX).resolve(type.superClass().getName());
+            rewriteManifestFile(dryRun, manifestPath, elements);
+        }
+    }
+
+    /**
+     * Rewrite a file, creating it and any parent directories if it does not exist, deleting it if specified
+     *
+     * @param dryRun              True if the rewrite should be applied, false if it should be simulated.
+     * @param filePath            Path to file which should be rewritten.
+     * @param elements            Class names of plugins which should appear in the manifest
+     */
+    private void rewriteManifestFile(boolean dryRun, Path filePath, Set<String> elements) throws IOException, TerseException {
+        Objects.requireNonNull(filePath, "filePath must be non-null");
+        Objects.requireNonNull(elements, "elements must be non-null");
+        Path writableFile = getWritablePath(dryRun, filePath);
+        if (elements.isEmpty()) {
+            if (Files.exists(filePath)) {
+                out.printf("\tDelete\t%s%n", filePath);
+                if (!dryRun) {
+                    Files.delete(writableFile);
+                }
+            }
+            // TODO: delete parent directories if empty up to the containing directory
+        } else {
+            Path parent = filePath.getParent();
+            if (parent != null) {
+                if (Files.exists(parent)) {
+                    if (!Files.isWritable(parent)) {
+                        throw new TerseException(parent + " is not writable");
+                    }
+                } else {
+                    out.printf("\tCreate\t%s%n", parent);
+                    if (!dryRun) {
+                        Files.createDirectories(parent);
+                    }
+                }
+            }
+            if (!Files.exists(filePath)) {
+                out.printf("\tCreate\t%s%n", filePath);
+            }
+            out.printf("\tWrite\t%s with content %s%n", filePath, elements);
+            if (!dryRun) {
+                try (OutputStream stream = new BufferedOutputStream(Files.newOutputStream(
+                        writableFile,
+                        StandardOpenOption.CREATE,
+                        StandardOpenOption.WRITE,
+                        StandardOpenOption.TRUNCATE_EXISTING

Review Comment:
   Great, 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

[GitHub] [kafka] gharris1727 commented on a diff in pull request #14195: KAFKA-15228: Add sync-manifests command to connect-plugin-path (KIP-898)

["gharris1727 (via GitHub)" <gi...@apache.org>]

gharris1727 commented on code in PR #14195:
URL: https://github.com/apache/kafka/pull/14195#discussion_r1294053643


##########
tools/src/main/java/org/apache/kafka/tools/ManifestWorkspace.java:
##########
@@ -0,0 +1,573 @@
+/*
+ * 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.tools;
+
+import org.apache.kafka.connect.runtime.isolation.PluginSource;
+import org.apache.kafka.connect.runtime.isolation.PluginType;
+
+import java.io.BufferedOutputStream;
+import java.io.BufferedReader;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.InputStreamReader;
+import java.io.OutputStream;
+import java.io.PrintStream;
+import java.net.MalformedURLException;
+import java.net.URI;
+import java.net.URISyntaxException;
+import java.net.URL;
+import java.net.URLConnection;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.FileSystem;
+import java.nio.file.FileSystems;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+import java.nio.file.StandardCopyOption;
+import java.nio.file.StandardOpenOption;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.EnumMap;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.LinkedHashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Set;
+import java.util.function.BiConsumer;
+import java.util.zip.ZipInputStream;
+import java.util.zip.ZipOutputStream;
+
+/**
+ * An in-memory workspace for manipulating {@link java.util.ServiceLoader} manifest files.
+ * <p>Use {@link #forSource(PluginSource)} to get a workspace scoped to a single plugin location, which is able
+ * to accept simulated reads and writes of manifests.
+ * Write the simulated changes to disk via {@link #commit(boolean)}.
+ */
+public class ManifestWorkspace {
+
+    private static final String MANIFEST_PREFIX = "META-INF/services/";
+    private static final Path MANAGED_PATH = Paths.get("connect-plugin-path-shim-0.0.1-SNAPSHOT.jar");
+    private final PrintStream out;
+    private final List<SourceWorkspace<?>> workspaces;
+    private final Map<Path, Path> temporaryOverlayFiles;
+
+    public ManifestWorkspace(PrintStream out) {
+        this.out = out;
+        workspaces = new ArrayList<>();
+        temporaryOverlayFiles = new HashMap<>();
+    }
+
+    public SourceWorkspace<?> forSource(PluginSource source) throws IOException {
+        SourceWorkspace<?> sourceWorkspace;
+        switch (source.type()) {
+            case CLASSPATH:
+                sourceWorkspace = new ClasspathWorkspace(source);
+                break;
+            case MULTI_JAR:
+                sourceWorkspace = new MultiJarWorkspace(source);
+                break;
+            case SINGLE_JAR:
+                sourceWorkspace = new SingleJarWorkspace(source);
+                break;
+            case CLASS_HIERARCHY:
+                sourceWorkspace = new ClassHierarchyWorkspace(source);
+                break;
+            default:
+                throw new IllegalStateException("Unknown source type " + source.type());
+        }
+        workspaces.add(sourceWorkspace);
+        return sourceWorkspace;
+    }
+
+    /**
+     * Commits all queued changes to disk
+     * @return true if any workspace wrote changes to disk, false if all workspaces did not have writes to apply
+     * @throws IOException if an error occurs reading or writing to the filesystem
+     * @throws TerseException if a path is not writable on disk and should be.
+     */
+    public boolean commit(boolean dryRun) throws IOException, TerseException {
+        boolean changed = false;
+        for (SourceWorkspace<?> workspace : workspaces) {
+            changed |= workspace.commit(dryRun);
+        }
+        return changed;
+    }
+
+    /**
+     * A workspace scoped to a single plugin source.
+     * <p>Buffers simulated reads and writes to the plugin path before they can be written to disk.
+     * @param <T> The data structure used by the workspace to store in-memory manifests internally.
+     */
+    public static abstract class SourceWorkspace<T> {
+        private final Path location;
+        private final PluginSource.Type type;
+        protected final T initial;
+        protected final T manifests;
+
+        private SourceWorkspace(PluginSource source) throws IOException {
+            this.location = source.location();
+            this.type = source.type();
+            this.initial = load(source);
+            this.manifests = load(source);
+        }
+
+        public Path location() {
+            return location;
+        }
+
+        public PluginSource.Type type() {
+            return type;
+        }
+
+        protected abstract T load(PluginSource source) throws IOException;
+
+        public abstract boolean hasManifest(PluginType type, String className);
+
+        public abstract void forEach(BiConsumer<String, PluginType> consumer);
+
+        public abstract void addManifest(PluginType type, String pluginClass);
+
+        public abstract void removeManifest(PluginType type, String pluginClass);
+
+        protected abstract boolean commit(boolean dryRun) throws TerseException, IOException;
+
+        protected static Map<PluginType, Set<String>> loadManifest(URL baseUrl) throws MalformedURLException {
+            Map<PluginType, Set<String>> manifests = new EnumMap<>(PluginType.class);
+            for (PluginType type : PluginType.values()) {
+                Set<String> result;
+                try {
+                    URL u = new URL(baseUrl, MANIFEST_PREFIX + type.superClass().getName());
+                    result = parse(u);
+                } catch (RuntimeException e) {
+                    result = new LinkedHashSet<>();
+                }
+                manifests.put(type, result);
+            }
+            return manifests;
+        }
+
+        protected static URL jarBaseUrl(URL fileUrl) throws MalformedURLException {
+            return new URL("jar", "", -1, fileUrl + "!/", null);
+        }
+
+        protected static void forEach(Map<PluginType, Set<String>> manifests, BiConsumer<String, PluginType> consumer) {
+            manifests.forEach((type, classNames) -> classNames.forEach(className -> consumer.accept(className, type)));
+        }
+    }
+
+    /**
+     * A single jar can only contain one manifest per plugin type.
+     */
+    private class SingleJarWorkspace extends SourceWorkspace<Map<PluginType, Set<String>>> {
+
+        private SingleJarWorkspace(PluginSource source) throws IOException {
+            super(source);

Review Comment:
   it looks a little out of place as the only assertion, but it should always hold.



-- 
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

[GitHub] [kafka] C0urante commented on a diff in pull request #14195: KAFKA-15228: Add sync-manifests command to connect-plugin-path (KIP-898)

["C0urante (via GitHub)" <gi...@apache.org>]

C0urante commented on code in PR #14195:
URL: https://github.com/apache/kafka/pull/14195#discussion_r1294745531


##########
tools/src/main/java/org/apache/kafka/tools/ManifestWorkspace.java:
##########
@@ -0,0 +1,573 @@
+/*
+ * 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.tools;
+
+import org.apache.kafka.connect.runtime.isolation.PluginSource;
+import org.apache.kafka.connect.runtime.isolation.PluginType;
+
+import java.io.BufferedOutputStream;
+import java.io.BufferedReader;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.InputStreamReader;
+import java.io.OutputStream;
+import java.io.PrintStream;
+import java.net.MalformedURLException;
+import java.net.URI;
+import java.net.URISyntaxException;
+import java.net.URL;
+import java.net.URLConnection;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.FileSystem;
+import java.nio.file.FileSystems;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+import java.nio.file.StandardCopyOption;
+import java.nio.file.StandardOpenOption;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.EnumMap;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.LinkedHashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Set;
+import java.util.function.BiConsumer;
+import java.util.zip.ZipInputStream;
+import java.util.zip.ZipOutputStream;
+
+/**
+ * An in-memory workspace for manipulating {@link java.util.ServiceLoader} manifest files.
+ * <p>Use {@link #forSource(PluginSource)} to get a workspace scoped to a single plugin location, which is able
+ * to accept simulated reads and writes of manifests.
+ * Write the simulated changes to disk via {@link #commit(boolean)}.
+ */
+public class ManifestWorkspace {
+
+    private static final String MANIFEST_PREFIX = "META-INF/services/";
+    private static final Path MANAGED_PATH = Paths.get("connect-plugin-path-shim-0.0.1-SNAPSHOT.jar");

Review Comment:
   Fine by me 👍 



##########
tools/src/main/java/org/apache/kafka/tools/ManifestWorkspace.java:
##########
@@ -0,0 +1,573 @@
+/*
+ * 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.tools;
+
+import org.apache.kafka.connect.runtime.isolation.PluginSource;
+import org.apache.kafka.connect.runtime.isolation.PluginType;
+
+import java.io.BufferedOutputStream;
+import java.io.BufferedReader;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.InputStreamReader;
+import java.io.OutputStream;
+import java.io.PrintStream;
+import java.net.MalformedURLException;
+import java.net.URI;
+import java.net.URISyntaxException;
+import java.net.URL;
+import java.net.URLConnection;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.FileSystem;
+import java.nio.file.FileSystems;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+import java.nio.file.StandardCopyOption;
+import java.nio.file.StandardOpenOption;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.EnumMap;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.LinkedHashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Set;
+import java.util.function.BiConsumer;
+import java.util.zip.ZipInputStream;
+import java.util.zip.ZipOutputStream;
+
+/**
+ * An in-memory workspace for manipulating {@link java.util.ServiceLoader} manifest files.
+ * <p>Use {@link #forSource(PluginSource)} to get a workspace scoped to a single plugin location, which is able
+ * to accept simulated reads and writes of manifests.
+ * Write the simulated changes to disk via {@link #commit(boolean)}.
+ */
+public class ManifestWorkspace {
+
+    private static final String MANIFEST_PREFIX = "META-INF/services/";
+    private static final Path MANAGED_PATH = Paths.get("connect-plugin-path-shim-0.0.1-SNAPSHOT.jar");

Review Comment:
   Fine by me 👍 



-- 
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

[GitHub] [kafka] gharris1727 commented on a diff in pull request #14195: KAFKA-15228: Add sync-manifests command to connect-plugin-path (KIP-898)

["gharris1727 (via GitHub)" <gi...@apache.org>]

gharris1727 commented on code in PR #14195:
URL: https://github.com/apache/kafka/pull/14195#discussion_r1294055531


##########
tools/src/main/java/org/apache/kafka/tools/ManifestWorkspace.java:
##########
@@ -0,0 +1,573 @@
+/*
+ * 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.tools;
+
+import org.apache.kafka.connect.runtime.isolation.PluginSource;
+import org.apache.kafka.connect.runtime.isolation.PluginType;
+
+import java.io.BufferedOutputStream;
+import java.io.BufferedReader;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.InputStreamReader;
+import java.io.OutputStream;
+import java.io.PrintStream;
+import java.net.MalformedURLException;
+import java.net.URI;
+import java.net.URISyntaxException;
+import java.net.URL;
+import java.net.URLConnection;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.FileSystem;
+import java.nio.file.FileSystems;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+import java.nio.file.StandardCopyOption;
+import java.nio.file.StandardOpenOption;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.EnumMap;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.LinkedHashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Set;
+import java.util.function.BiConsumer;
+import java.util.zip.ZipInputStream;
+import java.util.zip.ZipOutputStream;
+
+/**
+ * An in-memory workspace for manipulating {@link java.util.ServiceLoader} manifest files.
+ * <p>Use {@link #forSource(PluginSource)} to get a workspace scoped to a single plugin location, which is able
+ * to accept simulated reads and writes of manifests.
+ * Write the simulated changes to disk via {@link #commit(boolean)}.
+ */
+public class ManifestWorkspace {
+
+    private static final String MANIFEST_PREFIX = "META-INF/services/";
+    private static final Path MANAGED_PATH = Paths.get("connect-plugin-path-shim-0.0.1-SNAPSHOT.jar");
+    private final PrintStream out;
+    private final List<SourceWorkspace<?>> workspaces;
+    private final Map<Path, Path> temporaryOverlayFiles;
+
+    public ManifestWorkspace(PrintStream out) {
+        this.out = out;
+        workspaces = new ArrayList<>();
+        temporaryOverlayFiles = new HashMap<>();
+    }
+
+    public SourceWorkspace<?> forSource(PluginSource source) throws IOException {
+        SourceWorkspace<?> sourceWorkspace;
+        switch (source.type()) {
+            case CLASSPATH:
+                sourceWorkspace = new ClasspathWorkspace(source);
+                break;
+            case MULTI_JAR:
+                sourceWorkspace = new MultiJarWorkspace(source);
+                break;
+            case SINGLE_JAR:
+                sourceWorkspace = new SingleJarWorkspace(source);
+                break;
+            case CLASS_HIERARCHY:
+                sourceWorkspace = new ClassHierarchyWorkspace(source);
+                break;
+            default:
+                throw new IllegalStateException("Unknown source type " + source.type());
+        }
+        workspaces.add(sourceWorkspace);
+        return sourceWorkspace;
+    }
+
+    /**
+     * Commits all queued changes to disk
+     * @return true if any workspace wrote changes to disk, false if all workspaces did not have writes to apply
+     * @throws IOException if an error occurs reading or writing to the filesystem
+     * @throws TerseException if a path is not writable on disk and should be.
+     */
+    public boolean commit(boolean dryRun) throws IOException, TerseException {
+        boolean changed = false;
+        for (SourceWorkspace<?> workspace : workspaces) {
+            changed |= workspace.commit(dryRun);
+        }
+        return changed;
+    }
+
+    /**
+     * A workspace scoped to a single plugin source.
+     * <p>Buffers simulated reads and writes to the plugin path before they can be written to disk.
+     * @param <T> The data structure used by the workspace to store in-memory manifests internally.
+     */
+    public static abstract class SourceWorkspace<T> {
+        private final Path location;
+        private final PluginSource.Type type;
+        protected final T initial;
+        protected final T manifests;
+
+        private SourceWorkspace(PluginSource source) throws IOException {
+            this.location = source.location();
+            this.type = source.type();
+            this.initial = load(source);
+            this.manifests = load(source);
+        }
+
+        public Path location() {
+            return location;
+        }
+
+        public PluginSource.Type type() {
+            return type;
+        }
+
+        protected abstract T load(PluginSource source) throws IOException;
+
+        public abstract boolean hasManifest(PluginType type, String className);
+
+        public abstract void forEach(BiConsumer<String, PluginType> consumer);
+
+        public abstract void addManifest(PluginType type, String pluginClass);
+
+        public abstract void removeManifest(PluginType type, String pluginClass);
+
+        protected abstract boolean commit(boolean dryRun) throws TerseException, IOException;
+
+        protected static Map<PluginType, Set<String>> loadManifest(URL baseUrl) throws MalformedURLException {
+            Map<PluginType, Set<String>> manifests = new EnumMap<>(PluginType.class);
+            for (PluginType type : PluginType.values()) {
+                Set<String> result;
+                try {
+                    URL u = new URL(baseUrl, MANIFEST_PREFIX + type.superClass().getName());
+                    result = parse(u);
+                } catch (RuntimeException e) {
+                    result = new LinkedHashSet<>();
+                }
+                manifests.put(type, result);
+            }
+            return manifests;
+        }
+
+        protected static URL jarBaseUrl(URL fileUrl) throws MalformedURLException {
+            return new URL("jar", "", -1, fileUrl + "!/", null);
+        }
+
+        protected static void forEach(Map<PluginType, Set<String>> manifests, BiConsumer<String, PluginType> consumer) {
+            manifests.forEach((type, classNames) -> classNames.forEach(className -> consumer.accept(className, type)));
+        }
+    }
+
+    /**
+     * A single jar can only contain one manifest per plugin type.
+     */
+    private class SingleJarWorkspace extends SourceWorkspace<Map<PluginType, Set<String>>> {
+
+        private SingleJarWorkspace(PluginSource source) throws IOException {
+            super(source);
+        }
+
+        @Override
+        protected Map<PluginType, Set<String>> load(PluginSource source) throws IOException {
+            return loadManifest(jarBaseUrl(source.urls()[0]));
+        }
+
+        @Override
+        public boolean hasManifest(PluginType type, String className) {
+            return manifests.get(type).contains(className);
+        }
+
+        @Override
+        public void forEach(BiConsumer<String, PluginType> consumer) {
+            forEach(manifests, consumer);
+        }
+
+        @Override
+        public void addManifest(PluginType type, String pluginClass) {
+            manifests.get(type).add(pluginClass);
+        }
+
+        @Override
+        public void removeManifest(PluginType type, String pluginClass) {
+            manifests.get(type).remove(pluginClass);
+        }
+
+        @Override
+        protected boolean commit(boolean dryRun) throws IOException, TerseException {
+            if (startSync(dryRun, location(), initial, manifests)) {
+                rewriteJar(dryRun, location(), manifests);
+                return true;
+            }
+            return false;
+        }
+    }
+
+    /**
+     * A classpath workspace is backed by multiple jars, and that is not writable.
+     * The in-memory format is a map from jar path to the manifests contained in that jar.
+     * The control flow of the caller should not perform writes, so these exceptions indicate a bug in the program.
+     */
+    private class ClasspathWorkspace extends SourceWorkspace<Map<Path, Map<PluginType, Set<String>>>> {
+
+        private ClasspathWorkspace(PluginSource source) throws IOException {
+            super(source);
+        }
+
+        @Override
+        protected Map<Path, Map<PluginType, Set<String>>> load(PluginSource source) throws IOException {
+            Map<Path, Map<PluginType, Set<String>>> manifestsBySubLocation = new HashMap<>();
+            for (URL url : source.urls()) {
+                Path jarPath = Paths.get(url.getPath());
+                manifestsBySubLocation.put(jarPath, loadManifest(jarBaseUrl(url)));
+            }
+            return manifestsBySubLocation;
+        }
+
+        public boolean hasManifest(PluginType type, String className) {
+            return manifests.values()
+                    .stream()
+                    .map(m -> m.get(type))
+                    .anyMatch(s -> s.contains(className));
+        }
+
+        public void forEach(BiConsumer<String, PluginType> consumer) {
+            manifests.values().forEach(m -> forEach(m, consumer));
+        }
+
+        @Override
+        public void addManifest(PluginType type, String pluginClass) {
+            throw new UnsupportedOperationException("Cannot change the contents of the classpath");
+        }
+
+        @Override
+        public void removeManifest(PluginType type, String pluginClass) {
+            throw new UnsupportedOperationException("Cannot change the contents of the classpath");
+        }
+
+        @Override
+        protected boolean commit(boolean dryRun) throws IOException, TerseException {
+            // There is never anything to commit for the classpath
+            return false;
+        }
+    }
+
+    /**
+     * A multi-jar workspace is similar to the classpath workspace because it has multiple jars.
+     * However, the multi-jar workspace is writable, and injects a managed jar where it writes added manifests.
+     */
+    private class MultiJarWorkspace extends ClasspathWorkspace {
+
+        private MultiJarWorkspace(PluginSource source) throws IOException {
+            super(source);
+        }
+
+        @Override
+        protected Map<Path, Map<PluginType, Set<String>>> load(PluginSource source) throws IOException {
+            Map<Path, Map<PluginType, Set<String>>> manifests = super.load(source);
+            // In addition to the normal multi-jar paths, inject a managed jar where we can add manifests.
+            Path managedPath = source.location().resolve(MANAGED_PATH);
+            URL url = managedPath.toUri().toURL();
+            manifests.put(managedPath, loadManifest(jarBaseUrl(url)));
+            return manifests;
+        }
+
+        @Override
+        public void addManifest(PluginType type, String pluginClass) {
+            // Add plugins to the managed manifest
+            manifests.get(location().resolve(MANAGED_PATH)).get(type).add(pluginClass);
+        }
+
+        @Override
+        public void removeManifest(PluginType type, String pluginClass) {
+            // If a plugin appears in multiple manifests, remove it from all of them.
+            for (Map<PluginType, Set<String>> manifestState : manifests.values()) {
+                manifestState.get(type).remove(pluginClass);
+            }
+        }
+
+        @Override
+        public boolean commit(boolean dryRun) throws IOException, TerseException {
+            boolean changed = false;
+            for (Map.Entry<Path, Map<PluginType, Set<String>>> manifestSource : manifests.entrySet()) {
+                Path jarPath = manifestSource.getKey();
+                Map<PluginType, Set<String>> before = initial.get(jarPath);
+                Map<PluginType, Set<String>> after = manifestSource.getValue();
+                if (startSync(dryRun, jarPath, before, after)) {
+                    rewriteJar(dryRun, jarPath, after);
+                    changed = true;
+                }
+            }
+            return changed;
+        }
+    }
+
+    /**
+     * The class hierarchy is similar to the single-jar because there can only be one manifest per type.
+     * However, the path to that single manifest is accessed via the pluginLocation.
+     */
+    private class ClassHierarchyWorkspace extends SingleJarWorkspace {
+
+        private ClassHierarchyWorkspace(PluginSource source) throws IOException {
+            super(source);
+        }
+
+        @Override
+        protected Map<PluginType, Set<String>> load(PluginSource source) throws IOException {
+            return loadManifest(source.location().toUri().toURL());
+        }
+
+        protected boolean commit(boolean dryRun) throws IOException, TerseException {
+            if (startSync(dryRun, location(), initial, manifests)) {
+                rewriteClassHierarchyManifest(dryRun, location(), manifests);
+                return true;
+            }
+            return false;
+        }
+    }
+
+    private boolean startSync(boolean dryRun, Path syncLocation, Map<PluginType, Set<String>> before, Map<PluginType, Set<String>> after) {
+        Objects.requireNonNull(syncLocation, "syncLocation must be non-null");
+        Objects.requireNonNull(before, "before must be non-null");
+        Objects.requireNonNull(after, "after must be non-null");
+        if (before.equals(after)) {
+            return false;
+        }
+        Set<String> added = new HashSet<>();
+        after.values().forEach(added::addAll);
+        before.values().forEach(added::removeAll);
+        Set<String> removed = new HashSet<>();
+        before.values().forEach(removed::addAll);
+        after.values().forEach(removed::removeAll);
+        out.printf("%sSync\t\t%s Add %s Remove %s%n", dryRun ? "Dry Run " : "", syncLocation, added, removed);
+        return true;
+    }
+
+    /**
+     * Rewrite a jar on disk to contain a manifests containing the specified entries.
+     * Will create the jar file and manifests if they do not exist.
+     * Will delete the jar file and manifests if they become empty.
+     *
+     * @param dryRun        True if the rewrite should be applied, false if it should be simulated.
+     * @param jarPath       Path to a jar file for a plugin
+     * @param manifestState Map from plugin type to Class names of plugins which should appear in that manifest
+     */
+    private void rewriteJar(boolean dryRun, Path jarPath, Map<PluginType, Set<String>> manifestState) throws IOException, TerseException {
+        Objects.requireNonNull(jarPath, "jarPath must be non-null");
+        Objects.requireNonNull(manifestState, "manifestState must be non-null");
+        Path writableJar = getWritablePath(dryRun, jarPath);
+        if (!manifestState.values().stream().allMatch(Collection::isEmpty) && !Files.exists(writableJar)) {
+            out.printf("\tCreate\t%s%n", jarPath);
+            createJar(writableJar);
+        }
+        try (FileSystem jar = FileSystems.newFileSystem(
+                new URI("jar", writableJar.toUri().toString(), ""),
+                Collections.emptyMap()
+        )) {
+            Path zipRoot = jar.getRootDirectories().iterator().next();
+            rewriteClassHierarchyManifest(dryRun, zipRoot, manifestState);
+        } catch (URISyntaxException e) {
+            throw new IOException(e);
+        }
+        if (Files.exists(writableJar) && jarIsEmpty(writableJar)) {
+            Files.delete(writableJar);
+        }
+    }
+
+    private void createJar(Path path) throws IOException {
+        Objects.requireNonNull(path, "path must be non-null");
+        try (ZipOutputStream stream = new ZipOutputStream(Files.newOutputStream(
+                path,
+                StandardOpenOption.CREATE,
+                StandardOpenOption.WRITE,
+                StandardOpenOption.TRUNCATE_EXISTING
+        ))) {
+            stream.closeEntry();
+        }
+    }
+
+    private boolean jarIsEmpty(Path path) throws IOException {
+        Objects.requireNonNull(path, "path must be non-null");
+        try (ZipInputStream stream = new ZipInputStream(Files.newInputStream(
+                path,
+                StandardOpenOption.READ
+        ))) {
+            return stream.getNextEntry() == null;
+        }
+    }
+
+    /**
+     * Rewrite a multiple manifest files contained inside a class hierarchy.
+     * Will create the files and parent directories if they not exist.
+     * Will delete the files and parent directories within the plugin location if they become empty.
+     *
+     * @param dryRun         True if the rewrite should be applied, false if it should be simulated.
+     * @param pluginLocation Path to top-level of class hierarchy for a plugin
+     * @param manifestState  Map from plugin type to Class names of plugins which should appear in that manifest
+     */
+    private void rewriteClassHierarchyManifest(boolean dryRun, Path pluginLocation, Map<PluginType, Set<String>> manifestState) throws IOException, TerseException {
+        Objects.requireNonNull(pluginLocation, "pluginLocation must be non-null");
+        Objects.requireNonNull(manifestState, "manifestState must be non-null");
+        if (!Files.exists(pluginLocation)) {
+            throw new TerseException(pluginLocation + " does not exist");
+        }
+        if (!Files.isWritable(pluginLocation)) {
+            throw new TerseException(pluginLocation + " is not writable");
+        }
+        for (Map.Entry<PluginType, Set<String>> manifest : manifestState.entrySet()) {
+            PluginType type = manifest.getKey();
+            Set<String> elements = manifest.getValue();
+            Path manifestPath = pluginLocation.resolve(MANIFEST_PREFIX).resolve(type.superClass().getName());
+            rewriteManifestFile(dryRun, manifestPath, elements);
+        }
+    }
+
+    /**
+     * Rewrite a file, creating it and any parent directories if it does not exist, deleting it if specified
+     *
+     * @param dryRun              True if the rewrite should be applied, false if it should be simulated.
+     * @param filePath            Path to file which should be rewritten.
+     * @param elements            Class names of plugins which should appear in the manifest
+     */
+    private void rewriteManifestFile(boolean dryRun, Path filePath, Set<String> elements) throws IOException, TerseException {
+        Objects.requireNonNull(filePath, "filePath must be non-null");
+        Objects.requireNonNull(elements, "elements must be non-null");
+        Path writableFile = getWritablePath(dryRun, filePath);
+        if (elements.isEmpty()) {
+            if (Files.exists(filePath)) {
+                out.printf("\tDelete\t%s%n", filePath);
+                if (!dryRun) {
+                    Files.delete(writableFile);
+                }
+            }
+            // TODO: delete parent directories if empty up to the containing directory
+        } else {
+            Path parent = filePath.getParent();
+            if (parent != null) {
+                if (Files.exists(parent)) {
+                    if (!Files.isWritable(parent)) {
+                        throw new TerseException(parent + " is not writable");
+                    }
+                } else {
+                    out.printf("\tCreate\t%s%n", parent);
+                    if (!dryRun) {
+                        Files.createDirectories(parent);
+                    }
+                }
+            }
+            if (!Files.exists(filePath)) {
+                out.printf("\tCreate\t%s%n", filePath);
+            }
+            out.printf("\tWrite\t%s with content %s%n", filePath, elements);
+            if (!dryRun) {
+                try (OutputStream stream = new BufferedOutputStream(Files.newOutputStream(
+                        writableFile,
+                        StandardOpenOption.CREATE,
+                        StandardOpenOption.WRITE,
+                        StandardOpenOption.TRUNCATE_EXISTING
+                ))) {
+                    byte[] newline = System.lineSeparator().getBytes(StandardCharsets.UTF_8);
+                    String header = "# Generated by connect-plugin-path.sh";

Review Comment:
   this is a fantastic suggestion, especially if it is necessary to track down bugs in this logic later.



-- 
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

[GitHub] [kafka] gharris1727 commented on pull request #14195: KAFKA-15228: Add sync-manifests command to connect-plugin-path (KIP-898)

["gharris1727 (via GitHub)" <gi...@apache.org>]

gharris1727 commented on PR #14195:
URL: https://github.com/apache/kafka/pull/14195#issuecomment-1681097973

   Test failures appear unrelated, and the ConnectPluginPathTest suite has no failures.


-- 
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

[GitHub] [kafka] gharris1727 merged pull request #14195: KAFKA-15228: Add sync-manifests command to connect-plugin-path (KIP-898)

["gharris1727 (via GitHub)" <gi...@apache.org>]

gharris1727 merged PR #14195:
URL: https://github.com/apache/kafka/pull/14195


-- 
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

[GitHub] [kafka] C0urante commented on a diff in pull request #14195: KAFKA-15228: Add sync-manifests command to connect-plugin-path (KIP-898)

["C0urante (via GitHub)" <gi...@apache.org>]

C0urante commented on code in PR #14195:
URL: https://github.com/apache/kafka/pull/14195#discussion_r1293726875


##########
tools/src/main/java/org/apache/kafka/tools/ConnectPluginPath.java:
##########
@@ -368,6 +391,30 @@ private static void endCommand(
             config.out.printf("Total plugins:      \t%d%n", totalPlugins);
             config.out.printf("Loadable plugins:   \t%d%n", loadablePlugins);
             config.out.printf("Compatible plugins: \t%d%n", compatiblePlugins);
+        } else if (config.command == Command.SYNC_MANIFESTS) {
+            if (workspace.commit(true)) {
+                if (config.dryRun) {
+                    config.out.println("Dry run passed: All above changes can be committed to disk if re-run with dry run disabled.");
+                } else {
+                    config.out.println("Writing changes to plugins...");
+                    workspace.commit(false);
+                    config.out.println("All plugins have accurate ServiceLoader manifests");

Review Comment:
   Nit (I know this may be paranoid but I still think people are going to be confused/frustrated if they see this message but the script couldn't migrate a broken plugin):
   ```suggestion
                       config.out.println("All loadable plugins have accurate ServiceLoader manifests");
   ```



##########
tools/src/main/java/org/apache/kafka/tools/ConnectPluginPath.java:
##########
@@ -368,6 +391,30 @@ private static void endCommand(
             config.out.printf("Total plugins:      \t%d%n", totalPlugins);
             config.out.printf("Loadable plugins:   \t%d%n", loadablePlugins);
             config.out.printf("Compatible plugins: \t%d%n", compatiblePlugins);
+        } else if (config.command == Command.SYNC_MANIFESTS) {
+            if (workspace.commit(true)) {
+                if (config.dryRun) {
+                    config.out.println("Dry run passed: All above changes can be committed to disk if re-run with dry run disabled.");
+                } else {
+                    config.out.println("Writing changes to plugins...");
+                    workspace.commit(false);
+                    config.out.println("All plugins have accurate ServiceLoader manifests");
+                }
+            } else {
+                config.out.println("No changes required.");
+            }
+        }
+    }
+
+    private static void failCommand(Config config, Throwable e) {
+        if (config.command == Command.LIST) {
+            throw new RuntimeException("Unexpected error occurred while listing plugins", e);
+        } else if (config.command == Command.SYNC_MANIFESTS) {
+            if (config.dryRun) {
+                throw new RuntimeException("Unexpected error occurred while dry-running sync", e);
+            } else {
+                config.out.println("Connect plugin path now in unexpected state: Clear your plugin path and retry with dry run enabled");

Review Comment:
   Shouldn't we still log the error somehow?
   
   If the goal here is to emphasize the remedial action (i.e., clear the plugin path and do NOT try to start a worker with it), one option is to log the error, then add a blank line or two, then log this message. This would make the message here the last thing that the script prints, and hopefully the first thing that users read.



##########
tools/src/main/java/org/apache/kafka/tools/ManifestWorkspace.java:
##########
@@ -0,0 +1,573 @@
+/*
+ * 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.tools;
+
+import org.apache.kafka.connect.runtime.isolation.PluginSource;
+import org.apache.kafka.connect.runtime.isolation.PluginType;
+
+import java.io.BufferedOutputStream;
+import java.io.BufferedReader;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.InputStreamReader;
+import java.io.OutputStream;
+import java.io.PrintStream;
+import java.net.MalformedURLException;
+import java.net.URI;
+import java.net.URISyntaxException;
+import java.net.URL;
+import java.net.URLConnection;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.FileSystem;
+import java.nio.file.FileSystems;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+import java.nio.file.StandardCopyOption;
+import java.nio.file.StandardOpenOption;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.EnumMap;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.LinkedHashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Set;
+import java.util.function.BiConsumer;
+import java.util.zip.ZipInputStream;
+import java.util.zip.ZipOutputStream;
+
+/**
+ * An in-memory workspace for manipulating {@link java.util.ServiceLoader} manifest files.
+ * <p>Use {@link #forSource(PluginSource)} to get a workspace scoped to a single plugin location, which is able
+ * to accept simulated reads and writes of manifests.
+ * Write the simulated changes to disk via {@link #commit(boolean)}.
+ */
+public class ManifestWorkspace {
+
+    private static final String MANIFEST_PREFIX = "META-INF/services/";
+    private static final Path MANAGED_PATH = Paths.get("connect-plugin-path-shim-0.0.1-SNAPSHOT.jar");

Review Comment:
   I know it's idiomatic for artifacts in the Java world to have versions in JAR names, but it doesn't seem especially necessary in this case.
   
   If we do want to attach a version for some reason, we may want to make it more of a "real" version by using `1`, `1.0.0`, or something to that effect (calling this a snapshot or giving it a 0.x.y version seems inaccurate).



##########
tools/src/main/java/org/apache/kafka/tools/ManifestWorkspace.java:
##########
@@ -0,0 +1,573 @@
+/*
+ * 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.tools;
+
+import org.apache.kafka.connect.runtime.isolation.PluginSource;
+import org.apache.kafka.connect.runtime.isolation.PluginType;
+
+import java.io.BufferedOutputStream;
+import java.io.BufferedReader;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.InputStreamReader;
+import java.io.OutputStream;
+import java.io.PrintStream;
+import java.net.MalformedURLException;
+import java.net.URI;
+import java.net.URISyntaxException;
+import java.net.URL;
+import java.net.URLConnection;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.FileSystem;
+import java.nio.file.FileSystems;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+import java.nio.file.StandardCopyOption;
+import java.nio.file.StandardOpenOption;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.EnumMap;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.LinkedHashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Set;
+import java.util.function.BiConsumer;
+import java.util.zip.ZipInputStream;
+import java.util.zip.ZipOutputStream;
+
+/**
+ * An in-memory workspace for manipulating {@link java.util.ServiceLoader} manifest files.
+ * <p>Use {@link #forSource(PluginSource)} to get a workspace scoped to a single plugin location, which is able
+ * to accept simulated reads and writes of manifests.
+ * Write the simulated changes to disk via {@link #commit(boolean)}.
+ */
+public class ManifestWorkspace {
+
+    private static final String MANIFEST_PREFIX = "META-INF/services/";
+    private static final Path MANAGED_PATH = Paths.get("connect-plugin-path-shim-0.0.1-SNAPSHOT.jar");
+    private final PrintStream out;
+    private final List<SourceWorkspace<?>> workspaces;
+    private final Map<Path, Path> temporaryOverlayFiles;
+
+    public ManifestWorkspace(PrintStream out) {
+        this.out = out;
+        workspaces = new ArrayList<>();
+        temporaryOverlayFiles = new HashMap<>();
+    }
+
+    public SourceWorkspace<?> forSource(PluginSource source) throws IOException {
+        SourceWorkspace<?> sourceWorkspace;
+        switch (source.type()) {
+            case CLASSPATH:
+                sourceWorkspace = new ClasspathWorkspace(source);
+                break;
+            case MULTI_JAR:
+                sourceWorkspace = new MultiJarWorkspace(source);
+                break;
+            case SINGLE_JAR:
+                sourceWorkspace = new SingleJarWorkspace(source);
+                break;
+            case CLASS_HIERARCHY:
+                sourceWorkspace = new ClassHierarchyWorkspace(source);
+                break;
+            default:
+                throw new IllegalStateException("Unknown source type " + source.type());
+        }
+        workspaces.add(sourceWorkspace);
+        return sourceWorkspace;
+    }
+
+    /**
+     * Commits all queued changes to disk
+     * @return true if any workspace wrote changes to disk, false if all workspaces did not have writes to apply
+     * @throws IOException if an error occurs reading or writing to the filesystem
+     * @throws TerseException if a path is not writable on disk and should be.
+     */
+    public boolean commit(boolean dryRun) throws IOException, TerseException {
+        boolean changed = false;
+        for (SourceWorkspace<?> workspace : workspaces) {
+            changed |= workspace.commit(dryRun);
+        }
+        return changed;
+    }
+
+    /**
+     * A workspace scoped to a single plugin source.
+     * <p>Buffers simulated reads and writes to the plugin path before they can be written to disk.
+     * @param <T> The data structure used by the workspace to store in-memory manifests internally.
+     */
+    public static abstract class SourceWorkspace<T> {
+        private final Path location;
+        private final PluginSource.Type type;
+        protected final T initial;
+        protected final T manifests;
+
+        private SourceWorkspace(PluginSource source) throws IOException {
+            this.location = source.location();
+            this.type = source.type();
+            this.initial = load(source);
+            this.manifests = load(source);
+        }
+
+        public Path location() {
+            return location;
+        }
+
+        public PluginSource.Type type() {
+            return type;
+        }
+
+        protected abstract T load(PluginSource source) throws IOException;
+
+        public abstract boolean hasManifest(PluginType type, String className);
+
+        public abstract void forEach(BiConsumer<String, PluginType> consumer);
+
+        public abstract void addManifest(PluginType type, String pluginClass);
+
+        public abstract void removeManifest(PluginType type, String pluginClass);
+
+        protected abstract boolean commit(boolean dryRun) throws TerseException, IOException;
+
+        protected static Map<PluginType, Set<String>> loadManifest(URL baseUrl) throws MalformedURLException {
+            Map<PluginType, Set<String>> manifests = new EnumMap<>(PluginType.class);
+            for (PluginType type : PluginType.values()) {
+                Set<String> result;
+                try {
+                    URL u = new URL(baseUrl, MANIFEST_PREFIX + type.superClass().getName());
+                    result = parse(u);
+                } catch (RuntimeException e) {
+                    result = new LinkedHashSet<>();
+                }
+                manifests.put(type, result);
+            }
+            return manifests;
+        }
+
+        protected static URL jarBaseUrl(URL fileUrl) throws MalformedURLException {
+            return new URL("jar", "", -1, fileUrl + "!/", null);
+        }
+
+        protected static void forEach(Map<PluginType, Set<String>> manifests, BiConsumer<String, PluginType> consumer) {
+            manifests.forEach((type, classNames) -> classNames.forEach(className -> consumer.accept(className, type)));
+        }
+    }
+
+    /**
+     * A single jar can only contain one manifest per plugin type.
+     */
+    private class SingleJarWorkspace extends SourceWorkspace<Map<PluginType, Set<String>>> {
+
+        private SingleJarWorkspace(PluginSource source) throws IOException {
+            super(source);
+        }
+
+        @Override
+        protected Map<PluginType, Set<String>> load(PluginSource source) throws IOException {
+            return loadManifest(jarBaseUrl(source.urls()[0]));
+        }
+
+        @Override
+        public boolean hasManifest(PluginType type, String className) {
+            return manifests.get(type).contains(className);
+        }
+
+        @Override
+        public void forEach(BiConsumer<String, PluginType> consumer) {
+            forEach(manifests, consumer);
+        }
+
+        @Override
+        public void addManifest(PluginType type, String pluginClass) {
+            manifests.get(type).add(pluginClass);
+        }
+
+        @Override
+        public void removeManifest(PluginType type, String pluginClass) {
+            manifests.get(type).remove(pluginClass);
+        }
+
+        @Override
+        protected boolean commit(boolean dryRun) throws IOException, TerseException {
+            if (startSync(dryRun, location(), initial, manifests)) {
+                rewriteJar(dryRun, location(), manifests);
+                return true;
+            }
+            return false;
+        }
+    }
+
+    /**
+     * A classpath workspace is backed by multiple jars, and that is not writable.
+     * The in-memory format is a map from jar path to the manifests contained in that jar.
+     * The control flow of the caller should not perform writes, so these exceptions indicate a bug in the program.
+     */
+    private class ClasspathWorkspace extends SourceWorkspace<Map<Path, Map<PluginType, Set<String>>>> {
+
+        private ClasspathWorkspace(PluginSource source) throws IOException {
+            super(source);
+        }
+
+        @Override
+        protected Map<Path, Map<PluginType, Set<String>>> load(PluginSource source) throws IOException {
+            Map<Path, Map<PluginType, Set<String>>> manifestsBySubLocation = new HashMap<>();
+            for (URL url : source.urls()) {
+                Path jarPath = Paths.get(url.getPath());
+                manifestsBySubLocation.put(jarPath, loadManifest(jarBaseUrl(url)));
+            }
+            return manifestsBySubLocation;
+        }
+
+        public boolean hasManifest(PluginType type, String className) {
+            return manifests.values()
+                    .stream()
+                    .map(m -> m.get(type))
+                    .anyMatch(s -> s.contains(className));
+        }
+
+        public void forEach(BiConsumer<String, PluginType> consumer) {
+            manifests.values().forEach(m -> forEach(m, consumer));
+        }
+
+        @Override
+        public void addManifest(PluginType type, String pluginClass) {
+            throw new UnsupportedOperationException("Cannot change the contents of the classpath");
+        }
+
+        @Override
+        public void removeManifest(PluginType type, String pluginClass) {
+            throw new UnsupportedOperationException("Cannot change the contents of the classpath");
+        }
+
+        @Override
+        protected boolean commit(boolean dryRun) throws IOException, TerseException {
+            // There is never anything to commit for the classpath
+            return false;
+        }
+    }
+
+    /**
+     * A multi-jar workspace is similar to the classpath workspace because it has multiple jars.
+     * However, the multi-jar workspace is writable, and injects a managed jar where it writes added manifests.
+     */
+    private class MultiJarWorkspace extends ClasspathWorkspace {
+
+        private MultiJarWorkspace(PluginSource source) throws IOException {
+            super(source);
+        }
+
+        @Override
+        protected Map<Path, Map<PluginType, Set<String>>> load(PluginSource source) throws IOException {
+            Map<Path, Map<PluginType, Set<String>>> manifests = super.load(source);
+            // In addition to the normal multi-jar paths, inject a managed jar where we can add manifests.
+            Path managedPath = source.location().resolve(MANAGED_PATH);
+            URL url = managedPath.toUri().toURL();
+            manifests.put(managedPath, loadManifest(jarBaseUrl(url)));
+            return manifests;
+        }
+
+        @Override
+        public void addManifest(PluginType type, String pluginClass) {
+            // Add plugins to the managed manifest
+            manifests.get(location().resolve(MANAGED_PATH)).get(type).add(pluginClass);
+        }
+
+        @Override
+        public void removeManifest(PluginType type, String pluginClass) {
+            // If a plugin appears in multiple manifests, remove it from all of them.
+            for (Map<PluginType, Set<String>> manifestState : manifests.values()) {
+                manifestState.get(type).remove(pluginClass);
+            }
+        }
+
+        @Override
+        public boolean commit(boolean dryRun) throws IOException, TerseException {
+            boolean changed = false;
+            for (Map.Entry<Path, Map<PluginType, Set<String>>> manifestSource : manifests.entrySet()) {
+                Path jarPath = manifestSource.getKey();
+                Map<PluginType, Set<String>> before = initial.get(jarPath);
+                Map<PluginType, Set<String>> after = manifestSource.getValue();
+                if (startSync(dryRun, jarPath, before, after)) {
+                    rewriteJar(dryRun, jarPath, after);
+                    changed = true;
+                }
+            }
+            return changed;
+        }
+    }
+
+    /**
+     * The class hierarchy is similar to the single-jar because there can only be one manifest per type.
+     * However, the path to that single manifest is accessed via the pluginLocation.
+     */
+    private class ClassHierarchyWorkspace extends SingleJarWorkspace {
+
+        private ClassHierarchyWorkspace(PluginSource source) throws IOException {
+            super(source);
+        }
+
+        @Override
+        protected Map<PluginType, Set<String>> load(PluginSource source) throws IOException {
+            return loadManifest(source.location().toUri().toURL());
+        }
+
+        protected boolean commit(boolean dryRun) throws IOException, TerseException {
+            if (startSync(dryRun, location(), initial, manifests)) {
+                rewriteClassHierarchyManifest(dryRun, location(), manifests);
+                return true;
+            }
+            return false;
+        }
+    }
+
+    private boolean startSync(boolean dryRun, Path syncLocation, Map<PluginType, Set<String>> before, Map<PluginType, Set<String>> after) {
+        Objects.requireNonNull(syncLocation, "syncLocation must be non-null");
+        Objects.requireNonNull(before, "before must be non-null");
+        Objects.requireNonNull(after, "after must be non-null");
+        if (before.equals(after)) {
+            return false;
+        }
+        Set<String> added = new HashSet<>();
+        after.values().forEach(added::addAll);
+        before.values().forEach(added::removeAll);
+        Set<String> removed = new HashSet<>();
+        before.values().forEach(removed::addAll);
+        after.values().forEach(removed::removeAll);
+        out.printf("%sSync\t\t%s Add %s Remove %s%n", dryRun ? "Dry Run " : "", syncLocation, added, removed);
+        return true;
+    }
+
+    /**
+     * Rewrite a jar on disk to contain a manifests containing the specified entries.
+     * Will create the jar file and manifests if they do not exist.
+     * Will delete the jar file and manifests if they become empty.
+     *
+     * @param dryRun        True if the rewrite should be applied, false if it should be simulated.
+     * @param jarPath       Path to a jar file for a plugin
+     * @param manifestState Map from plugin type to Class names of plugins which should appear in that manifest
+     */
+    private void rewriteJar(boolean dryRun, Path jarPath, Map<PluginType, Set<String>> manifestState) throws IOException, TerseException {
+        Objects.requireNonNull(jarPath, "jarPath must be non-null");
+        Objects.requireNonNull(manifestState, "manifestState must be non-null");
+        Path writableJar = getWritablePath(dryRun, jarPath);
+        if (!manifestState.values().stream().allMatch(Collection::isEmpty) && !Files.exists(writableJar)) {
+            out.printf("\tCreate\t%s%n", jarPath);
+            createJar(writableJar);
+        }
+        try (FileSystem jar = FileSystems.newFileSystem(
+                new URI("jar", writableJar.toUri().toString(), ""),
+                Collections.emptyMap()
+        )) {
+            Path zipRoot = jar.getRootDirectories().iterator().next();
+            rewriteClassHierarchyManifest(dryRun, zipRoot, manifestState);
+        } catch (URISyntaxException e) {
+            throw new IOException(e);
+        }
+        if (Files.exists(writableJar) && jarIsEmpty(writableJar)) {
+            Files.delete(writableJar);
+        }
+    }
+
+    private void createJar(Path path) throws IOException {
+        Objects.requireNonNull(path, "path must be non-null");
+        try (ZipOutputStream stream = new ZipOutputStream(Files.newOutputStream(
+                path,
+                StandardOpenOption.CREATE,
+                StandardOpenOption.WRITE,
+                StandardOpenOption.TRUNCATE_EXISTING
+        ))) {
+            stream.closeEntry();
+        }
+    }
+
+    private boolean jarIsEmpty(Path path) throws IOException {
+        Objects.requireNonNull(path, "path must be non-null");
+        try (ZipInputStream stream = new ZipInputStream(Files.newInputStream(
+                path,
+                StandardOpenOption.READ
+        ))) {
+            return stream.getNextEntry() == null;
+        }
+    }
+
+    /**
+     * Rewrite a multiple manifest files contained inside a class hierarchy.
+     * Will create the files and parent directories if they not exist.
+     * Will delete the files and parent directories within the plugin location if they become empty.
+     *
+     * @param dryRun         True if the rewrite should be applied, false if it should be simulated.
+     * @param pluginLocation Path to top-level of class hierarchy for a plugin
+     * @param manifestState  Map from plugin type to Class names of plugins which should appear in that manifest
+     */
+    private void rewriteClassHierarchyManifest(boolean dryRun, Path pluginLocation, Map<PluginType, Set<String>> manifestState) throws IOException, TerseException {
+        Objects.requireNonNull(pluginLocation, "pluginLocation must be non-null");
+        Objects.requireNonNull(manifestState, "manifestState must be non-null");
+        if (!Files.exists(pluginLocation)) {
+            throw new TerseException(pluginLocation + " does not exist");
+        }
+        if (!Files.isWritable(pluginLocation)) {
+            throw new TerseException(pluginLocation + " is not writable");
+        }
+        for (Map.Entry<PluginType, Set<String>> manifest : manifestState.entrySet()) {
+            PluginType type = manifest.getKey();
+            Set<String> elements = manifest.getValue();
+            Path manifestPath = pluginLocation.resolve(MANIFEST_PREFIX).resolve(type.superClass().getName());
+            rewriteManifestFile(dryRun, manifestPath, elements);
+        }
+    }
+
+    /**
+     * Rewrite a file, creating it and any parent directories if it does not exist, deleting it if specified
+     *
+     * @param dryRun              True if the rewrite should be applied, false if it should be simulated.
+     * @param filePath            Path to file which should be rewritten.
+     * @param elements            Class names of plugins which should appear in the manifest
+     */
+    private void rewriteManifestFile(boolean dryRun, Path filePath, Set<String> elements) throws IOException, TerseException {
+        Objects.requireNonNull(filePath, "filePath must be non-null");
+        Objects.requireNonNull(elements, "elements must be non-null");
+        Path writableFile = getWritablePath(dryRun, filePath);
+        if (elements.isEmpty()) {
+            if (Files.exists(filePath)) {
+                out.printf("\tDelete\t%s%n", filePath);
+                if (!dryRun) {
+                    Files.delete(writableFile);
+                }
+            }
+            // TODO: delete parent directories if empty up to the containing directory
+        } else {
+            Path parent = filePath.getParent();
+            if (parent != null) {
+                if (Files.exists(parent)) {
+                    if (!Files.isWritable(parent)) {
+                        throw new TerseException(parent + " is not writable");
+                    }
+                } else {
+                    out.printf("\tCreate\t%s%n", parent);
+                    if (!dryRun) {
+                        Files.createDirectories(parent);
+                    }
+                }
+            }
+            if (!Files.exists(filePath)) {
+                out.printf("\tCreate\t%s%n", filePath);
+            }
+            out.printf("\tWrite\t%s with content %s%n", filePath, elements);
+            if (!dryRun) {
+                try (OutputStream stream = new BufferedOutputStream(Files.newOutputStream(
+                        writableFile,
+                        StandardOpenOption.CREATE,
+                        StandardOpenOption.WRITE,
+                        StandardOpenOption.TRUNCATE_EXISTING
+                ))) {
+                    byte[] newline = System.lineSeparator().getBytes(StandardCharsets.UTF_8);
+                    String header = "# Generated by connect-plugin-path.sh";

Review Comment:
   Do you think it'd help to include the [version](https://github.com/apache/kafka/blob/67b527460e155f1f5e850bb7fd65c7c373367b48/clients/src/main/java/org/apache/kafka/common/utils/AppInfoParser.java#L52-L54) of the script here?



##########
tools/src/main/java/org/apache/kafka/tools/ManifestWorkspace.java:
##########
@@ -0,0 +1,573 @@
+/*
+ * 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.tools;
+
+import org.apache.kafka.connect.runtime.isolation.PluginSource;
+import org.apache.kafka.connect.runtime.isolation.PluginType;
+
+import java.io.BufferedOutputStream;
+import java.io.BufferedReader;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.InputStreamReader;
+import java.io.OutputStream;
+import java.io.PrintStream;
+import java.net.MalformedURLException;
+import java.net.URI;
+import java.net.URISyntaxException;
+import java.net.URL;
+import java.net.URLConnection;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.FileSystem;
+import java.nio.file.FileSystems;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+import java.nio.file.StandardCopyOption;
+import java.nio.file.StandardOpenOption;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.EnumMap;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.LinkedHashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Set;
+import java.util.function.BiConsumer;
+import java.util.zip.ZipInputStream;
+import java.util.zip.ZipOutputStream;
+
+/**
+ * An in-memory workspace for manipulating {@link java.util.ServiceLoader} manifest files.
+ * <p>Use {@link #forSource(PluginSource)} to get a workspace scoped to a single plugin location, which is able
+ * to accept simulated reads and writes of manifests.
+ * Write the simulated changes to disk via {@link #commit(boolean)}.
+ */
+public class ManifestWorkspace {
+
+    private static final String MANIFEST_PREFIX = "META-INF/services/";
+    private static final Path MANAGED_PATH = Paths.get("connect-plugin-path-shim-0.0.1-SNAPSHOT.jar");
+    private final PrintStream out;
+    private final List<SourceWorkspace<?>> workspaces;
+    private final Map<Path, Path> temporaryOverlayFiles;
+
+    public ManifestWorkspace(PrintStream out) {
+        this.out = out;
+        workspaces = new ArrayList<>();
+        temporaryOverlayFiles = new HashMap<>();
+    }
+
+    public SourceWorkspace<?> forSource(PluginSource source) throws IOException {
+        SourceWorkspace<?> sourceWorkspace;
+        switch (source.type()) {
+            case CLASSPATH:
+                sourceWorkspace = new ClasspathWorkspace(source);
+                break;
+            case MULTI_JAR:
+                sourceWorkspace = new MultiJarWorkspace(source);
+                break;
+            case SINGLE_JAR:
+                sourceWorkspace = new SingleJarWorkspace(source);
+                break;
+            case CLASS_HIERARCHY:
+                sourceWorkspace = new ClassHierarchyWorkspace(source);
+                break;
+            default:
+                throw new IllegalStateException("Unknown source type " + source.type());
+        }
+        workspaces.add(sourceWorkspace);
+        return sourceWorkspace;
+    }
+
+    /**
+     * Commits all queued changes to disk
+     * @return true if any workspace wrote changes to disk, false if all workspaces did not have writes to apply
+     * @throws IOException if an error occurs reading or writing to the filesystem
+     * @throws TerseException if a path is not writable on disk and should be.
+     */
+    public boolean commit(boolean dryRun) throws IOException, TerseException {
+        boolean changed = false;
+        for (SourceWorkspace<?> workspace : workspaces) {
+            changed |= workspace.commit(dryRun);
+        }
+        return changed;
+    }
+
+    /**
+     * A workspace scoped to a single plugin source.
+     * <p>Buffers simulated reads and writes to the plugin path before they can be written to disk.
+     * @param <T> The data structure used by the workspace to store in-memory manifests internally.
+     */
+    public static abstract class SourceWorkspace<T> {
+        private final Path location;
+        private final PluginSource.Type type;
+        protected final T initial;
+        protected final T manifests;
+
+        private SourceWorkspace(PluginSource source) throws IOException {
+            this.location = source.location();
+            this.type = source.type();
+            this.initial = load(source);
+            this.manifests = load(source);
+        }
+
+        public Path location() {
+            return location;
+        }
+
+        public PluginSource.Type type() {
+            return type;
+        }
+
+        protected abstract T load(PluginSource source) throws IOException;
+
+        public abstract boolean hasManifest(PluginType type, String className);
+
+        public abstract void forEach(BiConsumer<String, PluginType> consumer);
+
+        public abstract void addManifest(PluginType type, String pluginClass);
+
+        public abstract void removeManifest(PluginType type, String pluginClass);
+
+        protected abstract boolean commit(boolean dryRun) throws TerseException, IOException;
+
+        protected static Map<PluginType, Set<String>> loadManifest(URL baseUrl) throws MalformedURLException {
+            Map<PluginType, Set<String>> manifests = new EnumMap<>(PluginType.class);
+            for (PluginType type : PluginType.values()) {
+                Set<String> result;
+                try {
+                    URL u = new URL(baseUrl, MANIFEST_PREFIX + type.superClass().getName());
+                    result = parse(u);
+                } catch (RuntimeException e) {
+                    result = new LinkedHashSet<>();
+                }
+                manifests.put(type, result);
+            }
+            return manifests;
+        }
+
+        protected static URL jarBaseUrl(URL fileUrl) throws MalformedURLException {
+            return new URL("jar", "", -1, fileUrl + "!/", null);
+        }
+
+        protected static void forEach(Map<PluginType, Set<String>> manifests, BiConsumer<String, PluginType> consumer) {
+            manifests.forEach((type, classNames) -> classNames.forEach(className -> consumer.accept(className, type)));
+        }
+    }
+
+    /**
+     * A single jar can only contain one manifest per plugin type.
+     */
+    private class SingleJarWorkspace extends SourceWorkspace<Map<PluginType, Set<String>>> {
+
+        private SingleJarWorkspace(PluginSource source) throws IOException {
+            super(source);

Review Comment:
   Worth adding an assertion that `source::urls` contains exactly one element?



##########
tools/src/main/java/org/apache/kafka/tools/ManifestWorkspace.java:
##########
@@ -0,0 +1,573 @@
+/*
+ * 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.tools;
+
+import org.apache.kafka.connect.runtime.isolation.PluginSource;
+import org.apache.kafka.connect.runtime.isolation.PluginType;
+
+import java.io.BufferedOutputStream;
+import java.io.BufferedReader;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.InputStreamReader;
+import java.io.OutputStream;
+import java.io.PrintStream;
+import java.net.MalformedURLException;
+import java.net.URI;
+import java.net.URISyntaxException;
+import java.net.URL;
+import java.net.URLConnection;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.FileSystem;
+import java.nio.file.FileSystems;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+import java.nio.file.StandardCopyOption;
+import java.nio.file.StandardOpenOption;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.EnumMap;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.LinkedHashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Set;
+import java.util.function.BiConsumer;
+import java.util.zip.ZipInputStream;
+import java.util.zip.ZipOutputStream;
+
+/**
+ * An in-memory workspace for manipulating {@link java.util.ServiceLoader} manifest files.
+ * <p>Use {@link #forSource(PluginSource)} to get a workspace scoped to a single plugin location, which is able
+ * to accept simulated reads and writes of manifests.
+ * Write the simulated changes to disk via {@link #commit(boolean)}.
+ */
+public class ManifestWorkspace {
+
+    private static final String MANIFEST_PREFIX = "META-INF/services/";
+    private static final Path MANAGED_PATH = Paths.get("connect-plugin-path-shim-0.0.1-SNAPSHOT.jar");
+    private final PrintStream out;
+    private final List<SourceWorkspace<?>> workspaces;
+    private final Map<Path, Path> temporaryOverlayFiles;
+
+    public ManifestWorkspace(PrintStream out) {
+        this.out = out;
+        workspaces = new ArrayList<>();
+        temporaryOverlayFiles = new HashMap<>();
+    }
+
+    public SourceWorkspace<?> forSource(PluginSource source) throws IOException {
+        SourceWorkspace<?> sourceWorkspace;
+        switch (source.type()) {
+            case CLASSPATH:
+                sourceWorkspace = new ClasspathWorkspace(source);
+                break;
+            case MULTI_JAR:
+                sourceWorkspace = new MultiJarWorkspace(source);
+                break;
+            case SINGLE_JAR:
+                sourceWorkspace = new SingleJarWorkspace(source);
+                break;
+            case CLASS_HIERARCHY:
+                sourceWorkspace = new ClassHierarchyWorkspace(source);
+                break;
+            default:
+                throw new IllegalStateException("Unknown source type " + source.type());
+        }
+        workspaces.add(sourceWorkspace);
+        return sourceWorkspace;
+    }
+
+    /**
+     * Commits all queued changes to disk
+     * @return true if any workspace wrote changes to disk, false if all workspaces did not have writes to apply
+     * @throws IOException if an error occurs reading or writing to the filesystem
+     * @throws TerseException if a path is not writable on disk and should be.
+     */
+    public boolean commit(boolean dryRun) throws IOException, TerseException {
+        boolean changed = false;
+        for (SourceWorkspace<?> workspace : workspaces) {
+            changed |= workspace.commit(dryRun);
+        }
+        return changed;
+    }
+
+    /**
+     * A workspace scoped to a single plugin source.
+     * <p>Buffers simulated reads and writes to the plugin path before they can be written to disk.
+     * @param <T> The data structure used by the workspace to store in-memory manifests internally.
+     */
+    public static abstract class SourceWorkspace<T> {
+        private final Path location;
+        private final PluginSource.Type type;
+        protected final T initial;
+        protected final T manifests;
+
+        private SourceWorkspace(PluginSource source) throws IOException {
+            this.location = source.location();
+            this.type = source.type();
+            this.initial = load(source);
+            this.manifests = load(source);
+        }
+
+        public Path location() {
+            return location;
+        }
+
+        public PluginSource.Type type() {
+            return type;
+        }
+
+        protected abstract T load(PluginSource source) throws IOException;
+
+        public abstract boolean hasManifest(PluginType type, String className);
+
+        public abstract void forEach(BiConsumer<String, PluginType> consumer);
+
+        public abstract void addManifest(PluginType type, String pluginClass);
+
+        public abstract void removeManifest(PluginType type, String pluginClass);
+
+        protected abstract boolean commit(boolean dryRun) throws TerseException, IOException;
+
+        protected static Map<PluginType, Set<String>> loadManifest(URL baseUrl) throws MalformedURLException {
+            Map<PluginType, Set<String>> manifests = new EnumMap<>(PluginType.class);
+            for (PluginType type : PluginType.values()) {
+                Set<String> result;
+                try {
+                    URL u = new URL(baseUrl, MANIFEST_PREFIX + type.superClass().getName());
+                    result = parse(u);
+                } catch (RuntimeException e) {
+                    result = new LinkedHashSet<>();
+                }
+                manifests.put(type, result);
+            }
+            return manifests;
+        }
+
+        protected static URL jarBaseUrl(URL fileUrl) throws MalformedURLException {
+            return new URL("jar", "", -1, fileUrl + "!/", null);
+        }
+
+        protected static void forEach(Map<PluginType, Set<String>> manifests, BiConsumer<String, PluginType> consumer) {
+            manifests.forEach((type, classNames) -> classNames.forEach(className -> consumer.accept(className, type)));
+        }
+    }
+
+    /**
+     * A single jar can only contain one manifest per plugin type.
+     */
+    private class SingleJarWorkspace extends SourceWorkspace<Map<PluginType, Set<String>>> {
+
+        private SingleJarWorkspace(PluginSource source) throws IOException {
+            super(source);
+        }
+
+        @Override
+        protected Map<PluginType, Set<String>> load(PluginSource source) throws IOException {
+            return loadManifest(jarBaseUrl(source.urls()[0]));
+        }
+
+        @Override
+        public boolean hasManifest(PluginType type, String className) {
+            return manifests.get(type).contains(className);
+        }
+
+        @Override
+        public void forEach(BiConsumer<String, PluginType> consumer) {
+            forEach(manifests, consumer);
+        }
+
+        @Override
+        public void addManifest(PluginType type, String pluginClass) {
+            manifests.get(type).add(pluginClass);
+        }
+
+        @Override
+        public void removeManifest(PluginType type, String pluginClass) {
+            manifests.get(type).remove(pluginClass);
+        }
+
+        @Override
+        protected boolean commit(boolean dryRun) throws IOException, TerseException {
+            if (startSync(dryRun, location(), initial, manifests)) {
+                rewriteJar(dryRun, location(), manifests);
+                return true;
+            }
+            return false;
+        }
+    }
+
+    /**
+     * A classpath workspace is backed by multiple jars, and that is not writable.
+     * The in-memory format is a map from jar path to the manifests contained in that jar.
+     * The control flow of the caller should not perform writes, so these exceptions indicate a bug in the program.
+     */
+    private class ClasspathWorkspace extends SourceWorkspace<Map<Path, Map<PluginType, Set<String>>>> {
+
+        private ClasspathWorkspace(PluginSource source) throws IOException {
+            super(source);
+        }
+
+        @Override
+        protected Map<Path, Map<PluginType, Set<String>>> load(PluginSource source) throws IOException {
+            Map<Path, Map<PluginType, Set<String>>> manifestsBySubLocation = new HashMap<>();
+            for (URL url : source.urls()) {
+                Path jarPath = Paths.get(url.getPath());
+                manifestsBySubLocation.put(jarPath, loadManifest(jarBaseUrl(url)));
+            }
+            return manifestsBySubLocation;
+        }
+
+        public boolean hasManifest(PluginType type, String className) {
+            return manifests.values()
+                    .stream()
+                    .map(m -> m.get(type))
+                    .anyMatch(s -> s.contains(className));
+        }
+
+        public void forEach(BiConsumer<String, PluginType> consumer) {
+            manifests.values().forEach(m -> forEach(m, consumer));
+        }
+
+        @Override
+        public void addManifest(PluginType type, String pluginClass) {
+            throw new UnsupportedOperationException("Cannot change the contents of the classpath");
+        }
+
+        @Override
+        public void removeManifest(PluginType type, String pluginClass) {
+            throw new UnsupportedOperationException("Cannot change the contents of the classpath");
+        }
+
+        @Override
+        protected boolean commit(boolean dryRun) throws IOException, TerseException {
+            // There is never anything to commit for the classpath
+            return false;
+        }
+    }
+
+    /**
+     * A multi-jar workspace is similar to the classpath workspace because it has multiple jars.
+     * However, the multi-jar workspace is writable, and injects a managed jar where it writes added manifests.
+     */
+    private class MultiJarWorkspace extends ClasspathWorkspace {
+
+        private MultiJarWorkspace(PluginSource source) throws IOException {
+            super(source);
+        }
+
+        @Override
+        protected Map<Path, Map<PluginType, Set<String>>> load(PluginSource source) throws IOException {
+            Map<Path, Map<PluginType, Set<String>>> manifests = super.load(source);
+            // In addition to the normal multi-jar paths, inject a managed jar where we can add manifests.
+            Path managedPath = source.location().resolve(MANAGED_PATH);
+            URL url = managedPath.toUri().toURL();
+            manifests.put(managedPath, loadManifest(jarBaseUrl(url)));
+            return manifests;
+        }
+
+        @Override
+        public void addManifest(PluginType type, String pluginClass) {
+            // Add plugins to the managed manifest
+            manifests.get(location().resolve(MANAGED_PATH)).get(type).add(pluginClass);
+        }
+
+        @Override
+        public void removeManifest(PluginType type, String pluginClass) {
+            // If a plugin appears in multiple manifests, remove it from all of them.
+            for (Map<PluginType, Set<String>> manifestState : manifests.values()) {
+                manifestState.get(type).remove(pluginClass);
+            }
+        }
+
+        @Override
+        public boolean commit(boolean dryRun) throws IOException, TerseException {
+            boolean changed = false;
+            for (Map.Entry<Path, Map<PluginType, Set<String>>> manifestSource : manifests.entrySet()) {
+                Path jarPath = manifestSource.getKey();
+                Map<PluginType, Set<String>> before = initial.get(jarPath);
+                Map<PluginType, Set<String>> after = manifestSource.getValue();
+                if (startSync(dryRun, jarPath, before, after)) {
+                    rewriteJar(dryRun, jarPath, after);
+                    changed = true;
+                }
+            }
+            return changed;
+        }
+    }
+
+    /**
+     * The class hierarchy is similar to the single-jar because there can only be one manifest per type.
+     * However, the path to that single manifest is accessed via the pluginLocation.
+     */
+    private class ClassHierarchyWorkspace extends SingleJarWorkspace {
+
+        private ClassHierarchyWorkspace(PluginSource source) throws IOException {
+            super(source);
+        }
+
+        @Override
+        protected Map<PluginType, Set<String>> load(PluginSource source) throws IOException {
+            return loadManifest(source.location().toUri().toURL());
+        }
+
+        protected boolean commit(boolean dryRun) throws IOException, TerseException {
+            if (startSync(dryRun, location(), initial, manifests)) {
+                rewriteClassHierarchyManifest(dryRun, location(), manifests);
+                return true;
+            }
+            return false;
+        }
+    }
+
+    private boolean startSync(boolean dryRun, Path syncLocation, Map<PluginType, Set<String>> before, Map<PluginType, Set<String>> after) {
+        Objects.requireNonNull(syncLocation, "syncLocation must be non-null");
+        Objects.requireNonNull(before, "before must be non-null");
+        Objects.requireNonNull(after, "after must be non-null");
+        if (before.equals(after)) {
+            return false;
+        }
+        Set<String> added = new HashSet<>();
+        after.values().forEach(added::addAll);
+        before.values().forEach(added::removeAll);
+        Set<String> removed = new HashSet<>();
+        before.values().forEach(removed::addAll);
+        after.values().forEach(removed::removeAll);
+        out.printf("%sSync\t\t%s Add %s Remove %s%n", dryRun ? "Dry Run " : "", syncLocation, added, removed);
+        return true;
+    }
+
+    /**
+     * Rewrite a jar on disk to contain a manifests containing the specified entries.
+     * Will create the jar file and manifests if they do not exist.
+     * Will delete the jar file and manifests if they become empty.
+     *
+     * @param dryRun        True if the rewrite should be applied, false if it should be simulated.
+     * @param jarPath       Path to a jar file for a plugin
+     * @param manifestState Map from plugin type to Class names of plugins which should appear in that manifest
+     */
+    private void rewriteJar(boolean dryRun, Path jarPath, Map<PluginType, Set<String>> manifestState) throws IOException, TerseException {
+        Objects.requireNonNull(jarPath, "jarPath must be non-null");
+        Objects.requireNonNull(manifestState, "manifestState must be non-null");
+        Path writableJar = getWritablePath(dryRun, jarPath);
+        if (!manifestState.values().stream().allMatch(Collection::isEmpty) && !Files.exists(writableJar)) {
+            out.printf("\tCreate\t%s%n", jarPath);
+            createJar(writableJar);
+        }
+        try (FileSystem jar = FileSystems.newFileSystem(
+                new URI("jar", writableJar.toUri().toString(), ""),
+                Collections.emptyMap()
+        )) {
+            Path zipRoot = jar.getRootDirectories().iterator().next();
+            rewriteClassHierarchyManifest(dryRun, zipRoot, manifestState);
+        } catch (URISyntaxException e) {
+            throw new IOException(e);
+        }
+        if (Files.exists(writableJar) && jarIsEmpty(writableJar)) {
+            Files.delete(writableJar);
+        }
+    }
+
+    private void createJar(Path path) throws IOException {
+        Objects.requireNonNull(path, "path must be non-null");
+        try (ZipOutputStream stream = new ZipOutputStream(Files.newOutputStream(
+                path,
+                StandardOpenOption.CREATE,
+                StandardOpenOption.WRITE,
+                StandardOpenOption.TRUNCATE_EXISTING
+        ))) {
+            stream.closeEntry();
+        }
+    }
+
+    private boolean jarIsEmpty(Path path) throws IOException {
+        Objects.requireNonNull(path, "path must be non-null");
+        try (ZipInputStream stream = new ZipInputStream(Files.newInputStream(
+                path,
+                StandardOpenOption.READ
+        ))) {
+            return stream.getNextEntry() == null;
+        }
+    }
+
+    /**
+     * Rewrite a multiple manifest files contained inside a class hierarchy.

Review Comment:
   Nit:
   ```suggestion
        * Rewrite multiple manifest files contained inside a class hierarchy.
   ```



##########
tools/src/test/java/org/apache/kafka/tools/ConnectPluginPathTest.java:
##########
@@ -192,6 +194,51 @@ public void testListMultipleWorkerConfigs(PluginLocationType type) {
                 TestPlugins.TestPlugin.SERVICE_LOADER);
     }
 
+    @ParameterizedTest
+    @EnumSource
+    public void testSyncManifests(PluginLocationType type) {
+        CommandResult res = runCommand(
+                "sync-manifests",
+                "--plugin-location",
+                setupLocation(workspace.resolve("location-a"), type, TestPlugins.TestPlugin.NON_MIGRATED_CONVERTER)
+        );
+        assertEquals(0, res.returnCode);
+        assertScanResult(true, TestPlugins.TestPlugin.NON_MIGRATED_CONVERTER, res.reflective);
+        assertScanResult(true, TestPlugins.TestPlugin.NON_MIGRATED_CONVERTER, res.serviceLoading);
+    }
+
+    @ParameterizedTest
+    @EnumSource
+    public void testSyncManifestsDryRun(PluginLocationType type) {
+        CommandResult res = runCommand(
+                "sync-manifests",
+                "--plugin-location",
+                setupLocation(workspace.resolve("location-a"), type, TestPlugins.TestPlugin.NON_MIGRATED_CONVERTER),
+                "--dry-run"
+        );
+        assertEquals(0, res.returnCode);
+        assertScanResult(true, TestPlugins.TestPlugin.NON_MIGRATED_CONVERTER, res.reflective);
+        assertScanResult(false, TestPlugins.TestPlugin.NON_MIGRATED_CONVERTER, res.serviceLoading);
+    }
+
+    @ParameterizedTest
+    @EnumSource
+    public void testSyncManifestsKeepNotFound(PluginLocationType type) {
+        CommandResult res = runCommand(
+                "sync-manifests",
+                "--plugin-location",
+                setupLocation(workspace.resolve("location-a"), type, TestPlugins.TestPlugin.BAD_PACKAGING_STATIC_INITIALIZER_THROWS_REST_EXTENSION),
+                "--plugin-location",
+                setupLocation(workspace.resolve("location-b"), type, TestPlugins.TestPlugin.NON_MIGRATED_CONVERTER),
+                "--keep-not-found"
+        );
+        assertEquals(0, res.returnCode);
+        assertScanResult(true, TestPlugins.TestPlugin.NON_MIGRATED_CONVERTER, res.reflective);
+        assertScanResult(true, TestPlugins.TestPlugin.NON_MIGRATED_CONVERTER, res.serviceLoading);
+        assertScanResult(false, TestPlugins.TestPlugin.BAD_PACKAGING_STATIC_INITIALIZER_THROWS_REST_EXTENSION, res.reflective);
+        assertScanResult(false, TestPlugins.TestPlugin.BAD_PACKAGING_STATIC_INITIALIZER_THROWS_REST_EXTENSION, res.serviceLoading);

Review Comment:
   I'm guessing it'd be pretty painful to also verify that there's still a manifest entry for the `StaticInitializerThrowsRestExtension` class after the command is run?



##########
tools/src/main/java/org/apache/kafka/tools/ManifestWorkspace.java:
##########
@@ -0,0 +1,573 @@
+/*
+ * 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.tools;
+
+import org.apache.kafka.connect.runtime.isolation.PluginSource;
+import org.apache.kafka.connect.runtime.isolation.PluginType;
+
+import java.io.BufferedOutputStream;
+import java.io.BufferedReader;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.InputStreamReader;
+import java.io.OutputStream;
+import java.io.PrintStream;
+import java.net.MalformedURLException;
+import java.net.URI;
+import java.net.URISyntaxException;
+import java.net.URL;
+import java.net.URLConnection;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.FileSystem;
+import java.nio.file.FileSystems;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+import java.nio.file.StandardCopyOption;
+import java.nio.file.StandardOpenOption;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.EnumMap;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.LinkedHashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Set;
+import java.util.function.BiConsumer;
+import java.util.zip.ZipInputStream;
+import java.util.zip.ZipOutputStream;
+
+/**
+ * An in-memory workspace for manipulating {@link java.util.ServiceLoader} manifest files.
+ * <p>Use {@link #forSource(PluginSource)} to get a workspace scoped to a single plugin location, which is able
+ * to accept simulated reads and writes of manifests.
+ * Write the simulated changes to disk via {@link #commit(boolean)}.
+ */
+public class ManifestWorkspace {
+
+    private static final String MANIFEST_PREFIX = "META-INF/services/";
+    private static final Path MANAGED_PATH = Paths.get("connect-plugin-path-shim-0.0.1-SNAPSHOT.jar");
+    private final PrintStream out;
+    private final List<SourceWorkspace<?>> workspaces;
+    private final Map<Path, Path> temporaryOverlayFiles;
+
+    public ManifestWorkspace(PrintStream out) {
+        this.out = out;
+        workspaces = new ArrayList<>();
+        temporaryOverlayFiles = new HashMap<>();
+    }
+
+    public SourceWorkspace<?> forSource(PluginSource source) throws IOException {
+        SourceWorkspace<?> sourceWorkspace;
+        switch (source.type()) {
+            case CLASSPATH:
+                sourceWorkspace = new ClasspathWorkspace(source);
+                break;
+            case MULTI_JAR:
+                sourceWorkspace = new MultiJarWorkspace(source);
+                break;
+            case SINGLE_JAR:
+                sourceWorkspace = new SingleJarWorkspace(source);
+                break;
+            case CLASS_HIERARCHY:
+                sourceWorkspace = new ClassHierarchyWorkspace(source);
+                break;
+            default:
+                throw new IllegalStateException("Unknown source type " + source.type());
+        }
+        workspaces.add(sourceWorkspace);
+        return sourceWorkspace;
+    }
+
+    /**
+     * Commits all queued changes to disk
+     * @return true if any workspace wrote changes to disk, false if all workspaces did not have writes to apply
+     * @throws IOException if an error occurs reading or writing to the filesystem
+     * @throws TerseException if a path is not writable on disk and should be.
+     */
+    public boolean commit(boolean dryRun) throws IOException, TerseException {
+        boolean changed = false;
+        for (SourceWorkspace<?> workspace : workspaces) {
+            changed |= workspace.commit(dryRun);
+        }
+        return changed;
+    }
+
+    /**
+     * A workspace scoped to a single plugin source.
+     * <p>Buffers simulated reads and writes to the plugin path before they can be written to disk.
+     * @param <T> The data structure used by the workspace to store in-memory manifests internally.
+     */
+    public static abstract class SourceWorkspace<T> {
+        private final Path location;
+        private final PluginSource.Type type;
+        protected final T initial;
+        protected final T manifests;
+
+        private SourceWorkspace(PluginSource source) throws IOException {
+            this.location = source.location();
+            this.type = source.type();
+            this.initial = load(source);
+            this.manifests = load(source);
+        }
+
+        public Path location() {
+            return location;
+        }
+
+        public PluginSource.Type type() {
+            return type;
+        }
+
+        protected abstract T load(PluginSource source) throws IOException;
+
+        public abstract boolean hasManifest(PluginType type, String className);
+
+        public abstract void forEach(BiConsumer<String, PluginType> consumer);
+
+        public abstract void addManifest(PluginType type, String pluginClass);
+
+        public abstract void removeManifest(PluginType type, String pluginClass);
+
+        protected abstract boolean commit(boolean dryRun) throws TerseException, IOException;
+
+        protected static Map<PluginType, Set<String>> loadManifest(URL baseUrl) throws MalformedURLException {
+            Map<PluginType, Set<String>> manifests = new EnumMap<>(PluginType.class);
+            for (PluginType type : PluginType.values()) {
+                Set<String> result;
+                try {
+                    URL u = new URL(baseUrl, MANIFEST_PREFIX + type.superClass().getName());
+                    result = parse(u);
+                } catch (RuntimeException e) {
+                    result = new LinkedHashSet<>();
+                }
+                manifests.put(type, result);
+            }
+            return manifests;
+        }
+
+        protected static URL jarBaseUrl(URL fileUrl) throws MalformedURLException {
+            return new URL("jar", "", -1, fileUrl + "!/", null);
+        }
+
+        protected static void forEach(Map<PluginType, Set<String>> manifests, BiConsumer<String, PluginType> consumer) {
+            manifests.forEach((type, classNames) -> classNames.forEach(className -> consumer.accept(className, type)));
+        }
+    }
+
+    /**
+     * A single jar can only contain one manifest per plugin type.
+     */
+    private class SingleJarWorkspace extends SourceWorkspace<Map<PluginType, Set<String>>> {
+
+        private SingleJarWorkspace(PluginSource source) throws IOException {
+            super(source);
+        }
+
+        @Override
+        protected Map<PluginType, Set<String>> load(PluginSource source) throws IOException {
+            return loadManifest(jarBaseUrl(source.urls()[0]));
+        }
+
+        @Override
+        public boolean hasManifest(PluginType type, String className) {
+            return manifests.get(type).contains(className);
+        }
+
+        @Override
+        public void forEach(BiConsumer<String, PluginType> consumer) {
+            forEach(manifests, consumer);
+        }
+
+        @Override
+        public void addManifest(PluginType type, String pluginClass) {
+            manifests.get(type).add(pluginClass);
+        }
+
+        @Override
+        public void removeManifest(PluginType type, String pluginClass) {
+            manifests.get(type).remove(pluginClass);
+        }
+
+        @Override
+        protected boolean commit(boolean dryRun) throws IOException, TerseException {
+            if (startSync(dryRun, location(), initial, manifests)) {
+                rewriteJar(dryRun, location(), manifests);
+                return true;
+            }
+            return false;
+        }
+    }
+
+    /**
+     * A classpath workspace is backed by multiple jars, and that is not writable.
+     * The in-memory format is a map from jar path to the manifests contained in that jar.
+     * The control flow of the caller should not perform writes, so these exceptions indicate a bug in the program.
+     */
+    private class ClasspathWorkspace extends SourceWorkspace<Map<Path, Map<PluginType, Set<String>>>> {
+
+        private ClasspathWorkspace(PluginSource source) throws IOException {
+            super(source);
+        }
+
+        @Override
+        protected Map<Path, Map<PluginType, Set<String>>> load(PluginSource source) throws IOException {
+            Map<Path, Map<PluginType, Set<String>>> manifestsBySubLocation = new HashMap<>();
+            for (URL url : source.urls()) {
+                Path jarPath = Paths.get(url.getPath());
+                manifestsBySubLocation.put(jarPath, loadManifest(jarBaseUrl(url)));
+            }
+            return manifestsBySubLocation;
+        }
+
+        public boolean hasManifest(PluginType type, String className) {
+            return manifests.values()
+                    .stream()
+                    .map(m -> m.get(type))
+                    .anyMatch(s -> s.contains(className));
+        }
+
+        public void forEach(BiConsumer<String, PluginType> consumer) {
+            manifests.values().forEach(m -> forEach(m, consumer));
+        }
+
+        @Override
+        public void addManifest(PluginType type, String pluginClass) {
+            throw new UnsupportedOperationException("Cannot change the contents of the classpath");
+        }
+
+        @Override
+        public void removeManifest(PluginType type, String pluginClass) {
+            throw new UnsupportedOperationException("Cannot change the contents of the classpath");
+        }
+
+        @Override
+        protected boolean commit(boolean dryRun) throws IOException, TerseException {
+            // There is never anything to commit for the classpath
+            return false;
+        }
+    }
+
+    /**
+     * A multi-jar workspace is similar to the classpath workspace because it has multiple jars.
+     * However, the multi-jar workspace is writable, and injects a managed jar where it writes added manifests.
+     */
+    private class MultiJarWorkspace extends ClasspathWorkspace {
+
+        private MultiJarWorkspace(PluginSource source) throws IOException {
+            super(source);
+        }
+
+        @Override
+        protected Map<Path, Map<PluginType, Set<String>>> load(PluginSource source) throws IOException {
+            Map<Path, Map<PluginType, Set<String>>> manifests = super.load(source);
+            // In addition to the normal multi-jar paths, inject a managed jar where we can add manifests.
+            Path managedPath = source.location().resolve(MANAGED_PATH);
+            URL url = managedPath.toUri().toURL();
+            manifests.put(managedPath, loadManifest(jarBaseUrl(url)));
+            return manifests;
+        }
+
+        @Override
+        public void addManifest(PluginType type, String pluginClass) {
+            // Add plugins to the managed manifest
+            manifests.get(location().resolve(MANAGED_PATH)).get(type).add(pluginClass);
+        }
+
+        @Override
+        public void removeManifest(PluginType type, String pluginClass) {
+            // If a plugin appears in multiple manifests, remove it from all of them.
+            for (Map<PluginType, Set<String>> manifestState : manifests.values()) {
+                manifestState.get(type).remove(pluginClass);
+            }
+        }
+
+        @Override
+        public boolean commit(boolean dryRun) throws IOException, TerseException {
+            boolean changed = false;
+            for (Map.Entry<Path, Map<PluginType, Set<String>>> manifestSource : manifests.entrySet()) {
+                Path jarPath = manifestSource.getKey();
+                Map<PluginType, Set<String>> before = initial.get(jarPath);
+                Map<PluginType, Set<String>> after = manifestSource.getValue();
+                if (startSync(dryRun, jarPath, before, after)) {
+                    rewriteJar(dryRun, jarPath, after);
+                    changed = true;
+                }
+            }
+            return changed;
+        }
+    }
+
+    /**
+     * The class hierarchy is similar to the single-jar because there can only be one manifest per type.
+     * However, the path to that single manifest is accessed via the pluginLocation.
+     */
+    private class ClassHierarchyWorkspace extends SingleJarWorkspace {
+
+        private ClassHierarchyWorkspace(PluginSource source) throws IOException {
+            super(source);
+        }
+
+        @Override
+        protected Map<PluginType, Set<String>> load(PluginSource source) throws IOException {
+            return loadManifest(source.location().toUri().toURL());
+        }
+
+        protected boolean commit(boolean dryRun) throws IOException, TerseException {
+            if (startSync(dryRun, location(), initial, manifests)) {
+                rewriteClassHierarchyManifest(dryRun, location(), manifests);
+                return true;
+            }
+            return false;
+        }
+    }
+
+    private boolean startSync(boolean dryRun, Path syncLocation, Map<PluginType, Set<String>> before, Map<PluginType, Set<String>> after) {
+        Objects.requireNonNull(syncLocation, "syncLocation must be non-null");
+        Objects.requireNonNull(before, "before must be non-null");
+        Objects.requireNonNull(after, "after must be non-null");
+        if (before.equals(after)) {
+            return false;
+        }
+        Set<String> added = new HashSet<>();
+        after.values().forEach(added::addAll);
+        before.values().forEach(added::removeAll);
+        Set<String> removed = new HashSet<>();
+        before.values().forEach(removed::addAll);
+        after.values().forEach(removed::removeAll);
+        out.printf("%sSync\t\t%s Add %s Remove %s%n", dryRun ? "Dry Run " : "", syncLocation, added, removed);
+        return true;
+    }
+
+    /**
+     * Rewrite a jar on disk to contain a manifests containing the specified entries.
+     * Will create the jar file and manifests if they do not exist.
+     * Will delete the jar file and manifests if they become empty.
+     *
+     * @param dryRun        True if the rewrite should be applied, false if it should be simulated.
+     * @param jarPath       Path to a jar file for a plugin
+     * @param manifestState Map from plugin type to Class names of plugins which should appear in that manifest
+     */
+    private void rewriteJar(boolean dryRun, Path jarPath, Map<PluginType, Set<String>> manifestState) throws IOException, TerseException {
+        Objects.requireNonNull(jarPath, "jarPath must be non-null");
+        Objects.requireNonNull(manifestState, "manifestState must be non-null");
+        Path writableJar = getWritablePath(dryRun, jarPath);
+        if (!manifestState.values().stream().allMatch(Collection::isEmpty) && !Files.exists(writableJar)) {
+            out.printf("\tCreate\t%s%n", jarPath);
+            createJar(writableJar);
+        }
+        try (FileSystem jar = FileSystems.newFileSystem(
+                new URI("jar", writableJar.toUri().toString(), ""),
+                Collections.emptyMap()
+        )) {
+            Path zipRoot = jar.getRootDirectories().iterator().next();
+            rewriteClassHierarchyManifest(dryRun, zipRoot, manifestState);
+        } catch (URISyntaxException e) {
+            throw new IOException(e);
+        }
+        if (Files.exists(writableJar) && jarIsEmpty(writableJar)) {
+            Files.delete(writableJar);
+        }
+    }
+
+    private void createJar(Path path) throws IOException {
+        Objects.requireNonNull(path, "path must be non-null");
+        try (ZipOutputStream stream = new ZipOutputStream(Files.newOutputStream(
+                path,
+                StandardOpenOption.CREATE,
+                StandardOpenOption.WRITE,
+                StandardOpenOption.TRUNCATE_EXISTING
+        ))) {
+            stream.closeEntry();
+        }
+    }
+
+    private boolean jarIsEmpty(Path path) throws IOException {
+        Objects.requireNonNull(path, "path must be non-null");
+        try (ZipInputStream stream = new ZipInputStream(Files.newInputStream(
+                path,
+                StandardOpenOption.READ
+        ))) {
+            return stream.getNextEntry() == null;
+        }
+    }
+
+    /**
+     * Rewrite a multiple manifest files contained inside a class hierarchy.
+     * Will create the files and parent directories if they not exist.
+     * Will delete the files and parent directories within the plugin location if they become empty.
+     *
+     * @param dryRun         True if the rewrite should be applied, false if it should be simulated.
+     * @param pluginLocation Path to top-level of class hierarchy for a plugin
+     * @param manifestState  Map from plugin type to Class names of plugins which should appear in that manifest
+     */
+    private void rewriteClassHierarchyManifest(boolean dryRun, Path pluginLocation, Map<PluginType, Set<String>> manifestState) throws IOException, TerseException {
+        Objects.requireNonNull(pluginLocation, "pluginLocation must be non-null");
+        Objects.requireNonNull(manifestState, "manifestState must be non-null");
+        if (!Files.exists(pluginLocation)) {
+            throw new TerseException(pluginLocation + " does not exist");
+        }
+        if (!Files.isWritable(pluginLocation)) {
+            throw new TerseException(pluginLocation + " is not writable");
+        }
+        for (Map.Entry<PluginType, Set<String>> manifest : manifestState.entrySet()) {
+            PluginType type = manifest.getKey();
+            Set<String> elements = manifest.getValue();
+            Path manifestPath = pluginLocation.resolve(MANIFEST_PREFIX).resolve(type.superClass().getName());
+            rewriteManifestFile(dryRun, manifestPath, elements);
+        }
+    }
+
+    /**
+     * Rewrite a file, creating it and any parent directories if it does not exist, deleting it if specified
+     *
+     * @param dryRun              True if the rewrite should be applied, false if it should be simulated.
+     * @param filePath            Path to file which should be rewritten.
+     * @param elements            Class names of plugins which should appear in the manifest
+     */
+    private void rewriteManifestFile(boolean dryRun, Path filePath, Set<String> elements) throws IOException, TerseException {
+        Objects.requireNonNull(filePath, "filePath must be non-null");
+        Objects.requireNonNull(elements, "elements must be non-null");
+        Path writableFile = getWritablePath(dryRun, filePath);
+        if (elements.isEmpty()) {
+            if (Files.exists(filePath)) {
+                out.printf("\tDelete\t%s%n", filePath);
+                if (!dryRun) {
+                    Files.delete(writableFile);
+                }
+            }
+            // TODO: delete parent directories if empty up to the containing directory
+        } else {
+            Path parent = filePath.getParent();
+            if (parent != null) {
+                if (Files.exists(parent)) {
+                    if (!Files.isWritable(parent)) {
+                        throw new TerseException(parent + " is not writable");
+                    }
+                } else {
+                    out.printf("\tCreate\t%s%n", parent);
+                    if (!dryRun) {
+                        Files.createDirectories(parent);
+                    }
+                }
+            }
+            if (!Files.exists(filePath)) {
+                out.printf("\tCreate\t%s%n", filePath);
+            }
+            out.printf("\tWrite\t%s with content %s%n", filePath, elements);
+            if (!dryRun) {
+                try (OutputStream stream = new BufferedOutputStream(Files.newOutputStream(
+                        writableFile,
+                        StandardOpenOption.CREATE,
+                        StandardOpenOption.WRITE,
+                        StandardOpenOption.TRUNCATE_EXISTING
+                ))) {
+                    byte[] newline = System.lineSeparator().getBytes(StandardCharsets.UTF_8);
+                    String header = "# Generated by connect-plugin-path.sh";
+                    stream.write(header.getBytes(StandardCharsets.UTF_8));
+                    stream.write(newline);
+                    for (String element : elements) {
+                        stream.write(element.getBytes(StandardCharsets.UTF_8));
+                        stream.write(newline);
+                    }
+                }
+            }
+        }
+    }
+
+    /**
+     * Get a path which is always writable
+     * @param dryRun If true, substitute a temporary file instead of the real file on disk.
+     * @param path Path which must be writable, and may be different from the input path

Review Comment:
   Is this part correct? I'm unsure what the difference between the `path` parameter and the "input path" is. Seems like part of this description may have been referring to the return value instead?



##########
tools/src/main/java/org/apache/kafka/tools/ManifestWorkspace.java:
##########
@@ -0,0 +1,573 @@
+/*
+ * 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.tools;
+
+import org.apache.kafka.connect.runtime.isolation.PluginSource;
+import org.apache.kafka.connect.runtime.isolation.PluginType;
+
+import java.io.BufferedOutputStream;
+import java.io.BufferedReader;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.InputStreamReader;
+import java.io.OutputStream;
+import java.io.PrintStream;
+import java.net.MalformedURLException;
+import java.net.URI;
+import java.net.URISyntaxException;
+import java.net.URL;
+import java.net.URLConnection;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.FileSystem;
+import java.nio.file.FileSystems;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+import java.nio.file.StandardCopyOption;
+import java.nio.file.StandardOpenOption;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.EnumMap;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.LinkedHashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Set;
+import java.util.function.BiConsumer;
+import java.util.zip.ZipInputStream;
+import java.util.zip.ZipOutputStream;
+
+/**
+ * An in-memory workspace for manipulating {@link java.util.ServiceLoader} manifest files.
+ * <p>Use {@link #forSource(PluginSource)} to get a workspace scoped to a single plugin location, which is able
+ * to accept simulated reads and writes of manifests.
+ * Write the simulated changes to disk via {@link #commit(boolean)}.
+ */
+public class ManifestWorkspace {
+
+    private static final String MANIFEST_PREFIX = "META-INF/services/";
+    private static final Path MANAGED_PATH = Paths.get("connect-plugin-path-shim-0.0.1-SNAPSHOT.jar");
+    private final PrintStream out;
+    private final List<SourceWorkspace<?>> workspaces;
+    private final Map<Path, Path> temporaryOverlayFiles;
+
+    public ManifestWorkspace(PrintStream out) {
+        this.out = out;
+        workspaces = new ArrayList<>();
+        temporaryOverlayFiles = new HashMap<>();
+    }
+
+    public SourceWorkspace<?> forSource(PluginSource source) throws IOException {
+        SourceWorkspace<?> sourceWorkspace;
+        switch (source.type()) {
+            case CLASSPATH:
+                sourceWorkspace = new ClasspathWorkspace(source);
+                break;
+            case MULTI_JAR:
+                sourceWorkspace = new MultiJarWorkspace(source);
+                break;
+            case SINGLE_JAR:
+                sourceWorkspace = new SingleJarWorkspace(source);
+                break;
+            case CLASS_HIERARCHY:
+                sourceWorkspace = new ClassHierarchyWorkspace(source);
+                break;
+            default:
+                throw new IllegalStateException("Unknown source type " + source.type());
+        }
+        workspaces.add(sourceWorkspace);
+        return sourceWorkspace;
+    }
+
+    /**
+     * Commits all queued changes to disk
+     * @return true if any workspace wrote changes to disk, false if all workspaces did not have writes to apply
+     * @throws IOException if an error occurs reading or writing to the filesystem
+     * @throws TerseException if a path is not writable on disk and should be.
+     */
+    public boolean commit(boolean dryRun) throws IOException, TerseException {
+        boolean changed = false;
+        for (SourceWorkspace<?> workspace : workspaces) {
+            changed |= workspace.commit(dryRun);
+        }
+        return changed;
+    }
+
+    /**
+     * A workspace scoped to a single plugin source.
+     * <p>Buffers simulated reads and writes to the plugin path before they can be written to disk.
+     * @param <T> The data structure used by the workspace to store in-memory manifests internally.
+     */
+    public static abstract class SourceWorkspace<T> {
+        private final Path location;
+        private final PluginSource.Type type;
+        protected final T initial;
+        protected final T manifests;
+
+        private SourceWorkspace(PluginSource source) throws IOException {
+            this.location = source.location();
+            this.type = source.type();
+            this.initial = load(source);
+            this.manifests = load(source);
+        }
+
+        public Path location() {
+            return location;
+        }
+
+        public PluginSource.Type type() {
+            return type;
+        }
+
+        protected abstract T load(PluginSource source) throws IOException;
+
+        public abstract boolean hasManifest(PluginType type, String className);
+
+        public abstract void forEach(BiConsumer<String, PluginType> consumer);
+
+        public abstract void addManifest(PluginType type, String pluginClass);
+
+        public abstract void removeManifest(PluginType type, String pluginClass);
+
+        protected abstract boolean commit(boolean dryRun) throws TerseException, IOException;
+
+        protected static Map<PluginType, Set<String>> loadManifest(URL baseUrl) throws MalformedURLException {
+            Map<PluginType, Set<String>> manifests = new EnumMap<>(PluginType.class);
+            for (PluginType type : PluginType.values()) {
+                Set<String> result;
+                try {
+                    URL u = new URL(baseUrl, MANIFEST_PREFIX + type.superClass().getName());
+                    result = parse(u);
+                } catch (RuntimeException e) {
+                    result = new LinkedHashSet<>();
+                }
+                manifests.put(type, result);
+            }
+            return manifests;
+        }
+
+        protected static URL jarBaseUrl(URL fileUrl) throws MalformedURLException {
+            return new URL("jar", "", -1, fileUrl + "!/", null);
+        }
+
+        protected static void forEach(Map<PluginType, Set<String>> manifests, BiConsumer<String, PluginType> consumer) {
+            manifests.forEach((type, classNames) -> classNames.forEach(className -> consumer.accept(className, type)));
+        }
+    }
+
+    /**
+     * A single jar can only contain one manifest per plugin type.
+     */
+    private class SingleJarWorkspace extends SourceWorkspace<Map<PluginType, Set<String>>> {
+
+        private SingleJarWorkspace(PluginSource source) throws IOException {
+            super(source);
+        }
+
+        @Override
+        protected Map<PluginType, Set<String>> load(PluginSource source) throws IOException {
+            return loadManifest(jarBaseUrl(source.urls()[0]));
+        }
+
+        @Override
+        public boolean hasManifest(PluginType type, String className) {
+            return manifests.get(type).contains(className);
+        }
+
+        @Override
+        public void forEach(BiConsumer<String, PluginType> consumer) {
+            forEach(manifests, consumer);
+        }
+
+        @Override
+        public void addManifest(PluginType type, String pluginClass) {
+            manifests.get(type).add(pluginClass);
+        }
+
+        @Override
+        public void removeManifest(PluginType type, String pluginClass) {
+            manifests.get(type).remove(pluginClass);
+        }
+
+        @Override
+        protected boolean commit(boolean dryRun) throws IOException, TerseException {
+            if (startSync(dryRun, location(), initial, manifests)) {
+                rewriteJar(dryRun, location(), manifests);
+                return true;
+            }
+            return false;
+        }
+    }
+
+    /**
+     * A classpath workspace is backed by multiple jars, and that is not writable.
+     * The in-memory format is a map from jar path to the manifests contained in that jar.
+     * The control flow of the caller should not perform writes, so these exceptions indicate a bug in the program.
+     */
+    private class ClasspathWorkspace extends SourceWorkspace<Map<Path, Map<PluginType, Set<String>>>> {
+
+        private ClasspathWorkspace(PluginSource source) throws IOException {
+            super(source);
+        }
+
+        @Override
+        protected Map<Path, Map<PluginType, Set<String>>> load(PluginSource source) throws IOException {
+            Map<Path, Map<PluginType, Set<String>>> manifestsBySubLocation = new HashMap<>();
+            for (URL url : source.urls()) {
+                Path jarPath = Paths.get(url.getPath());
+                manifestsBySubLocation.put(jarPath, loadManifest(jarBaseUrl(url)));
+            }
+            return manifestsBySubLocation;
+        }
+
+        public boolean hasManifest(PluginType type, String className) {
+            return manifests.values()
+                    .stream()
+                    .map(m -> m.get(type))
+                    .anyMatch(s -> s.contains(className));
+        }
+
+        public void forEach(BiConsumer<String, PluginType> consumer) {
+            manifests.values().forEach(m -> forEach(m, consumer));
+        }
+
+        @Override
+        public void addManifest(PluginType type, String pluginClass) {
+            throw new UnsupportedOperationException("Cannot change the contents of the classpath");
+        }
+
+        @Override
+        public void removeManifest(PluginType type, String pluginClass) {
+            throw new UnsupportedOperationException("Cannot change the contents of the classpath");
+        }
+
+        @Override
+        protected boolean commit(boolean dryRun) throws IOException, TerseException {
+            // There is never anything to commit for the classpath
+            return false;
+        }
+    }
+
+    /**
+     * A multi-jar workspace is similar to the classpath workspace because it has multiple jars.
+     * However, the multi-jar workspace is writable, and injects a managed jar where it writes added manifests.
+     */
+    private class MultiJarWorkspace extends ClasspathWorkspace {
+
+        private MultiJarWorkspace(PluginSource source) throws IOException {
+            super(source);
+        }
+
+        @Override
+        protected Map<Path, Map<PluginType, Set<String>>> load(PluginSource source) throws IOException {
+            Map<Path, Map<PluginType, Set<String>>> manifests = super.load(source);
+            // In addition to the normal multi-jar paths, inject a managed jar where we can add manifests.
+            Path managedPath = source.location().resolve(MANAGED_PATH);
+            URL url = managedPath.toUri().toURL();
+            manifests.put(managedPath, loadManifest(jarBaseUrl(url)));
+            return manifests;
+        }
+
+        @Override
+        public void addManifest(PluginType type, String pluginClass) {
+            // Add plugins to the managed manifest
+            manifests.get(location().resolve(MANAGED_PATH)).get(type).add(pluginClass);
+        }
+
+        @Override
+        public void removeManifest(PluginType type, String pluginClass) {
+            // If a plugin appears in multiple manifests, remove it from all of them.
+            for (Map<PluginType, Set<String>> manifestState : manifests.values()) {
+                manifestState.get(type).remove(pluginClass);
+            }
+        }
+
+        @Override
+        public boolean commit(boolean dryRun) throws IOException, TerseException {
+            boolean changed = false;
+            for (Map.Entry<Path, Map<PluginType, Set<String>>> manifestSource : manifests.entrySet()) {
+                Path jarPath = manifestSource.getKey();
+                Map<PluginType, Set<String>> before = initial.get(jarPath);
+                Map<PluginType, Set<String>> after = manifestSource.getValue();
+                if (startSync(dryRun, jarPath, before, after)) {
+                    rewriteJar(dryRun, jarPath, after);
+                    changed = true;
+                }
+            }
+            return changed;
+        }
+    }
+
+    /**
+     * The class hierarchy is similar to the single-jar because there can only be one manifest per type.
+     * However, the path to that single manifest is accessed via the pluginLocation.
+     */
+    private class ClassHierarchyWorkspace extends SingleJarWorkspace {
+
+        private ClassHierarchyWorkspace(PluginSource source) throws IOException {
+            super(source);
+        }
+
+        @Override
+        protected Map<PluginType, Set<String>> load(PluginSource source) throws IOException {
+            return loadManifest(source.location().toUri().toURL());
+        }
+
+        protected boolean commit(boolean dryRun) throws IOException, TerseException {
+            if (startSync(dryRun, location(), initial, manifests)) {
+                rewriteClassHierarchyManifest(dryRun, location(), manifests);
+                return true;
+            }
+            return false;
+        }
+    }
+
+    private boolean startSync(boolean dryRun, Path syncLocation, Map<PluginType, Set<String>> before, Map<PluginType, Set<String>> after) {
+        Objects.requireNonNull(syncLocation, "syncLocation must be non-null");
+        Objects.requireNonNull(before, "before must be non-null");
+        Objects.requireNonNull(after, "after must be non-null");
+        if (before.equals(after)) {
+            return false;
+        }
+        Set<String> added = new HashSet<>();
+        after.values().forEach(added::addAll);
+        before.values().forEach(added::removeAll);
+        Set<String> removed = new HashSet<>();
+        before.values().forEach(removed::addAll);
+        after.values().forEach(removed::removeAll);
+        out.printf("%sSync\t\t%s Add %s Remove %s%n", dryRun ? "Dry Run " : "", syncLocation, added, removed);
+        return true;
+    }
+
+    /**
+     * Rewrite a jar on disk to contain a manifests containing the specified entries.

Review Comment:
   Nit:
   ```suggestion
        * Rewrite a jar on disk to add manifests with the specified entries.
   ```



##########
tools/src/main/java/org/apache/kafka/tools/ManifestWorkspace.java:
##########
@@ -0,0 +1,573 @@
+/*
+ * 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.tools;
+
+import org.apache.kafka.connect.runtime.isolation.PluginSource;
+import org.apache.kafka.connect.runtime.isolation.PluginType;
+
+import java.io.BufferedOutputStream;
+import java.io.BufferedReader;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.InputStreamReader;
+import java.io.OutputStream;
+import java.io.PrintStream;
+import java.net.MalformedURLException;
+import java.net.URI;
+import java.net.URISyntaxException;
+import java.net.URL;
+import java.net.URLConnection;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.FileSystem;
+import java.nio.file.FileSystems;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+import java.nio.file.StandardCopyOption;
+import java.nio.file.StandardOpenOption;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.EnumMap;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.LinkedHashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Set;
+import java.util.function.BiConsumer;
+import java.util.zip.ZipInputStream;
+import java.util.zip.ZipOutputStream;
+
+/**
+ * An in-memory workspace for manipulating {@link java.util.ServiceLoader} manifest files.
+ * <p>Use {@link #forSource(PluginSource)} to get a workspace scoped to a single plugin location, which is able
+ * to accept simulated reads and writes of manifests.
+ * Write the simulated changes to disk via {@link #commit(boolean)}.
+ */
+public class ManifestWorkspace {
+
+    private static final String MANIFEST_PREFIX = "META-INF/services/";
+    private static final Path MANAGED_PATH = Paths.get("connect-plugin-path-shim-0.0.1-SNAPSHOT.jar");
+    private final PrintStream out;
+    private final List<SourceWorkspace<?>> workspaces;
+    private final Map<Path, Path> temporaryOverlayFiles;
+
+    public ManifestWorkspace(PrintStream out) {
+        this.out = out;
+        workspaces = new ArrayList<>();
+        temporaryOverlayFiles = new HashMap<>();
+    }
+
+    public SourceWorkspace<?> forSource(PluginSource source) throws IOException {
+        SourceWorkspace<?> sourceWorkspace;
+        switch (source.type()) {
+            case CLASSPATH:
+                sourceWorkspace = new ClasspathWorkspace(source);
+                break;
+            case MULTI_JAR:
+                sourceWorkspace = new MultiJarWorkspace(source);
+                break;
+            case SINGLE_JAR:
+                sourceWorkspace = new SingleJarWorkspace(source);
+                break;
+            case CLASS_HIERARCHY:
+                sourceWorkspace = new ClassHierarchyWorkspace(source);
+                break;
+            default:
+                throw new IllegalStateException("Unknown source type " + source.type());
+        }
+        workspaces.add(sourceWorkspace);
+        return sourceWorkspace;
+    }
+
+    /**
+     * Commits all queued changes to disk
+     * @return true if any workspace wrote changes to disk, false if all workspaces did not have writes to apply
+     * @throws IOException if an error occurs reading or writing to the filesystem
+     * @throws TerseException if a path is not writable on disk and should be.
+     */
+    public boolean commit(boolean dryRun) throws IOException, TerseException {
+        boolean changed = false;
+        for (SourceWorkspace<?> workspace : workspaces) {
+            changed |= workspace.commit(dryRun);
+        }
+        return changed;
+    }
+
+    /**
+     * A workspace scoped to a single plugin source.
+     * <p>Buffers simulated reads and writes to the plugin path before they can be written to disk.
+     * @param <T> The data structure used by the workspace to store in-memory manifests internally.
+     */
+    public static abstract class SourceWorkspace<T> {
+        private final Path location;
+        private final PluginSource.Type type;
+        protected final T initial;
+        protected final T manifests;
+
+        private SourceWorkspace(PluginSource source) throws IOException {
+            this.location = source.location();
+            this.type = source.type();
+            this.initial = load(source);
+            this.manifests = load(source);
+        }
+
+        public Path location() {
+            return location;
+        }
+
+        public PluginSource.Type type() {
+            return type;
+        }
+
+        protected abstract T load(PluginSource source) throws IOException;
+
+        public abstract boolean hasManifest(PluginType type, String className);
+
+        public abstract void forEach(BiConsumer<String, PluginType> consumer);
+
+        public abstract void addManifest(PluginType type, String pluginClass);
+
+        public abstract void removeManifest(PluginType type, String pluginClass);
+
+        protected abstract boolean commit(boolean dryRun) throws TerseException, IOException;
+
+        protected static Map<PluginType, Set<String>> loadManifest(URL baseUrl) throws MalformedURLException {
+            Map<PluginType, Set<String>> manifests = new EnumMap<>(PluginType.class);
+            for (PluginType type : PluginType.values()) {
+                Set<String> result;
+                try {
+                    URL u = new URL(baseUrl, MANIFEST_PREFIX + type.superClass().getName());
+                    result = parse(u);
+                } catch (RuntimeException e) {
+                    result = new LinkedHashSet<>();
+                }
+                manifests.put(type, result);
+            }
+            return manifests;
+        }
+
+        protected static URL jarBaseUrl(URL fileUrl) throws MalformedURLException {
+            return new URL("jar", "", -1, fileUrl + "!/", null);
+        }
+
+        protected static void forEach(Map<PluginType, Set<String>> manifests, BiConsumer<String, PluginType> consumer) {
+            manifests.forEach((type, classNames) -> classNames.forEach(className -> consumer.accept(className, type)));
+        }
+    }
+
+    /**
+     * A single jar can only contain one manifest per plugin type.
+     */
+    private class SingleJarWorkspace extends SourceWorkspace<Map<PluginType, Set<String>>> {
+
+        private SingleJarWorkspace(PluginSource source) throws IOException {
+            super(source);
+        }
+
+        @Override
+        protected Map<PluginType, Set<String>> load(PluginSource source) throws IOException {
+            return loadManifest(jarBaseUrl(source.urls()[0]));
+        }
+
+        @Override
+        public boolean hasManifest(PluginType type, String className) {
+            return manifests.get(type).contains(className);
+        }
+
+        @Override
+        public void forEach(BiConsumer<String, PluginType> consumer) {
+            forEach(manifests, consumer);
+        }
+
+        @Override
+        public void addManifest(PluginType type, String pluginClass) {
+            manifests.get(type).add(pluginClass);
+        }
+
+        @Override
+        public void removeManifest(PluginType type, String pluginClass) {
+            manifests.get(type).remove(pluginClass);
+        }
+
+        @Override
+        protected boolean commit(boolean dryRun) throws IOException, TerseException {
+            if (startSync(dryRun, location(), initial, manifests)) {
+                rewriteJar(dryRun, location(), manifests);
+                return true;
+            }
+            return false;
+        }
+    }
+
+    /**
+     * A classpath workspace is backed by multiple jars, and that is not writable.
+     * The in-memory format is a map from jar path to the manifests contained in that jar.
+     * The control flow of the caller should not perform writes, so these exceptions indicate a bug in the program.
+     */
+    private class ClasspathWorkspace extends SourceWorkspace<Map<Path, Map<PluginType, Set<String>>>> {
+
+        private ClasspathWorkspace(PluginSource source) throws IOException {
+            super(source);
+        }
+
+        @Override
+        protected Map<Path, Map<PluginType, Set<String>>> load(PluginSource source) throws IOException {
+            Map<Path, Map<PluginType, Set<String>>> manifestsBySubLocation = new HashMap<>();
+            for (URL url : source.urls()) {
+                Path jarPath = Paths.get(url.getPath());
+                manifestsBySubLocation.put(jarPath, loadManifest(jarBaseUrl(url)));
+            }
+            return manifestsBySubLocation;
+        }
+
+        public boolean hasManifest(PluginType type, String className) {
+            return manifests.values()
+                    .stream()
+                    .map(m -> m.get(type))
+                    .anyMatch(s -> s.contains(className));
+        }
+
+        public void forEach(BiConsumer<String, PluginType> consumer) {
+            manifests.values().forEach(m -> forEach(m, consumer));
+        }
+
+        @Override
+        public void addManifest(PluginType type, String pluginClass) {
+            throw new UnsupportedOperationException("Cannot change the contents of the classpath");
+        }
+
+        @Override
+        public void removeManifest(PluginType type, String pluginClass) {
+            throw new UnsupportedOperationException("Cannot change the contents of the classpath");
+        }
+
+        @Override
+        protected boolean commit(boolean dryRun) throws IOException, TerseException {
+            // There is never anything to commit for the classpath
+            return false;
+        }
+    }
+
+    /**
+     * A multi-jar workspace is similar to the classpath workspace because it has multiple jars.
+     * However, the multi-jar workspace is writable, and injects a managed jar where it writes added manifests.
+     */
+    private class MultiJarWorkspace extends ClasspathWorkspace {
+
+        private MultiJarWorkspace(PluginSource source) throws IOException {
+            super(source);
+        }
+
+        @Override
+        protected Map<Path, Map<PluginType, Set<String>>> load(PluginSource source) throws IOException {
+            Map<Path, Map<PluginType, Set<String>>> manifests = super.load(source);
+            // In addition to the normal multi-jar paths, inject a managed jar where we can add manifests.
+            Path managedPath = source.location().resolve(MANAGED_PATH);
+            URL url = managedPath.toUri().toURL();
+            manifests.put(managedPath, loadManifest(jarBaseUrl(url)));
+            return manifests;
+        }
+
+        @Override
+        public void addManifest(PluginType type, String pluginClass) {
+            // Add plugins to the managed manifest
+            manifests.get(location().resolve(MANAGED_PATH)).get(type).add(pluginClass);
+        }
+
+        @Override
+        public void removeManifest(PluginType type, String pluginClass) {
+            // If a plugin appears in multiple manifests, remove it from all of them.
+            for (Map<PluginType, Set<String>> manifestState : manifests.values()) {
+                manifestState.get(type).remove(pluginClass);
+            }
+        }
+
+        @Override
+        public boolean commit(boolean dryRun) throws IOException, TerseException {
+            boolean changed = false;
+            for (Map.Entry<Path, Map<PluginType, Set<String>>> manifestSource : manifests.entrySet()) {
+                Path jarPath = manifestSource.getKey();
+                Map<PluginType, Set<String>> before = initial.get(jarPath);
+                Map<PluginType, Set<String>> after = manifestSource.getValue();
+                if (startSync(dryRun, jarPath, before, after)) {
+                    rewriteJar(dryRun, jarPath, after);
+                    changed = true;
+                }
+            }
+            return changed;
+        }
+    }
+
+    /**
+     * The class hierarchy is similar to the single-jar because there can only be one manifest per type.
+     * However, the path to that single manifest is accessed via the pluginLocation.
+     */
+    private class ClassHierarchyWorkspace extends SingleJarWorkspace {
+
+        private ClassHierarchyWorkspace(PluginSource source) throws IOException {
+            super(source);
+        }
+
+        @Override
+        protected Map<PluginType, Set<String>> load(PluginSource source) throws IOException {
+            return loadManifest(source.location().toUri().toURL());
+        }
+
+        protected boolean commit(boolean dryRun) throws IOException, TerseException {
+            if (startSync(dryRun, location(), initial, manifests)) {
+                rewriteClassHierarchyManifest(dryRun, location(), manifests);
+                return true;
+            }
+            return false;
+        }
+    }
+
+    private boolean startSync(boolean dryRun, Path syncLocation, Map<PluginType, Set<String>> before, Map<PluginType, Set<String>> after) {
+        Objects.requireNonNull(syncLocation, "syncLocation must be non-null");
+        Objects.requireNonNull(before, "before must be non-null");
+        Objects.requireNonNull(after, "after must be non-null");
+        if (before.equals(after)) {
+            return false;
+        }
+        Set<String> added = new HashSet<>();
+        after.values().forEach(added::addAll);
+        before.values().forEach(added::removeAll);
+        Set<String> removed = new HashSet<>();
+        before.values().forEach(removed::addAll);
+        after.values().forEach(removed::removeAll);
+        out.printf("%sSync\t\t%s Add %s Remove %s%n", dryRun ? "Dry Run " : "", syncLocation, added, removed);
+        return true;
+    }
+
+    /**
+     * Rewrite a jar on disk to contain a manifests containing the specified entries.
+     * Will create the jar file and manifests if they do not exist.
+     * Will delete the jar file and manifests if they become empty.
+     *
+     * @param dryRun        True if the rewrite should be applied, false if it should be simulated.
+     * @param jarPath       Path to a jar file for a plugin
+     * @param manifestState Map from plugin type to Class names of plugins which should appear in that manifest
+     */
+    private void rewriteJar(boolean dryRun, Path jarPath, Map<PluginType, Set<String>> manifestState) throws IOException, TerseException {
+        Objects.requireNonNull(jarPath, "jarPath must be non-null");
+        Objects.requireNonNull(manifestState, "manifestState must be non-null");
+        Path writableJar = getWritablePath(dryRun, jarPath);
+        if (!manifestState.values().stream().allMatch(Collection::isEmpty) && !Files.exists(writableJar)) {
+            out.printf("\tCreate\t%s%n", jarPath);
+            createJar(writableJar);
+        }
+        try (FileSystem jar = FileSystems.newFileSystem(
+                new URI("jar", writableJar.toUri().toString(), ""),
+                Collections.emptyMap()
+        )) {
+            Path zipRoot = jar.getRootDirectories().iterator().next();
+            rewriteClassHierarchyManifest(dryRun, zipRoot, manifestState);
+        } catch (URISyntaxException e) {
+            throw new IOException(e);
+        }
+        if (Files.exists(writableJar) && jarIsEmpty(writableJar)) {
+            Files.delete(writableJar);
+        }
+    }
+
+    private void createJar(Path path) throws IOException {
+        Objects.requireNonNull(path, "path must be non-null");
+        try (ZipOutputStream stream = new ZipOutputStream(Files.newOutputStream(
+                path,
+                StandardOpenOption.CREATE,
+                StandardOpenOption.WRITE,
+                StandardOpenOption.TRUNCATE_EXISTING
+        ))) {
+            stream.closeEntry();
+        }
+    }
+
+    private boolean jarIsEmpty(Path path) throws IOException {
+        Objects.requireNonNull(path, "path must be non-null");
+        try (ZipInputStream stream = new ZipInputStream(Files.newInputStream(
+                path,
+                StandardOpenOption.READ
+        ))) {
+            return stream.getNextEntry() == null;
+        }
+    }
+
+    /**
+     * Rewrite a multiple manifest files contained inside a class hierarchy.
+     * Will create the files and parent directories if they not exist.
+     * Will delete the files and parent directories within the plugin location if they become empty.
+     *
+     * @param dryRun         True if the rewrite should be applied, false if it should be simulated.
+     * @param pluginLocation Path to top-level of class hierarchy for a plugin
+     * @param manifestState  Map from plugin type to Class names of plugins which should appear in that manifest
+     */
+    private void rewriteClassHierarchyManifest(boolean dryRun, Path pluginLocation, Map<PluginType, Set<String>> manifestState) throws IOException, TerseException {
+        Objects.requireNonNull(pluginLocation, "pluginLocation must be non-null");
+        Objects.requireNonNull(manifestState, "manifestState must be non-null");
+        if (!Files.exists(pluginLocation)) {
+            throw new TerseException(pluginLocation + " does not exist");
+        }
+        if (!Files.isWritable(pluginLocation)) {
+            throw new TerseException(pluginLocation + " is not writable");
+        }
+        for (Map.Entry<PluginType, Set<String>> manifest : manifestState.entrySet()) {
+            PluginType type = manifest.getKey();
+            Set<String> elements = manifest.getValue();
+            Path manifestPath = pluginLocation.resolve(MANIFEST_PREFIX).resolve(type.superClass().getName());
+            rewriteManifestFile(dryRun, manifestPath, elements);
+        }
+    }
+
+    /**
+     * Rewrite a file, creating it and any parent directories if it does not exist, deleting it if specified
+     *
+     * @param dryRun              True if the rewrite should be applied, false if it should be simulated.
+     * @param filePath            Path to file which should be rewritten.
+     * @param elements            Class names of plugins which should appear in the manifest
+     */
+    private void rewriteManifestFile(boolean dryRun, Path filePath, Set<String> elements) throws IOException, TerseException {
+        Objects.requireNonNull(filePath, "filePath must be non-null");
+        Objects.requireNonNull(elements, "elements must be non-null");
+        Path writableFile = getWritablePath(dryRun, filePath);
+        if (elements.isEmpty()) {
+            if (Files.exists(filePath)) {
+                out.printf("\tDelete\t%s%n", filePath);
+                if (!dryRun) {
+                    Files.delete(writableFile);
+                }
+            }
+            // TODO: delete parent directories if empty up to the containing directory
+        } else {
+            Path parent = filePath.getParent();
+            if (parent != null) {
+                if (Files.exists(parent)) {
+                    if (!Files.isWritable(parent)) {
+                        throw new TerseException(parent + " is not writable");
+                    }
+                } else {
+                    out.printf("\tCreate\t%s%n", parent);
+                    if (!dryRun) {
+                        Files.createDirectories(parent);
+                    }
+                }
+            }
+            if (!Files.exists(filePath)) {
+                out.printf("\tCreate\t%s%n", filePath);
+            }
+            out.printf("\tWrite\t%s with content %s%n", filePath, elements);
+            if (!dryRun) {
+                try (OutputStream stream = new BufferedOutputStream(Files.newOutputStream(
+                        writableFile,
+                        StandardOpenOption.CREATE,
+                        StandardOpenOption.WRITE,
+                        StandardOpenOption.TRUNCATE_EXISTING

Review Comment:
   Just for my edification, can you confirm my understanding of this logic?
   
   We unconditionally wipe any existing manifest files at this point, even if they contain entries for plugins that were either present but not loadable, or simply not present.
   
   This is safe because we still search for all manifest entries earlier on while doing reflection and service loader scanning, and optionally hold onto all manifest entries (even for unloadable/non-present plugins) during that time, before including them in the contents of the manifest file we write here.
   
   Is this correct?



##########
tools/src/main/java/org/apache/kafka/tools/ManifestWorkspace.java:
##########
@@ -0,0 +1,573 @@
+/*
+ * 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.tools;
+
+import org.apache.kafka.connect.runtime.isolation.PluginSource;
+import org.apache.kafka.connect.runtime.isolation.PluginType;
+
+import java.io.BufferedOutputStream;
+import java.io.BufferedReader;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.InputStreamReader;
+import java.io.OutputStream;
+import java.io.PrintStream;
+import java.net.MalformedURLException;
+import java.net.URI;
+import java.net.URISyntaxException;
+import java.net.URL;
+import java.net.URLConnection;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.FileSystem;
+import java.nio.file.FileSystems;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+import java.nio.file.StandardCopyOption;
+import java.nio.file.StandardOpenOption;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.EnumMap;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.LinkedHashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Set;
+import java.util.function.BiConsumer;
+import java.util.zip.ZipInputStream;
+import java.util.zip.ZipOutputStream;
+
+/**
+ * An in-memory workspace for manipulating {@link java.util.ServiceLoader} manifest files.
+ * <p>Use {@link #forSource(PluginSource)} to get a workspace scoped to a single plugin location, which is able
+ * to accept simulated reads and writes of manifests.
+ * Write the simulated changes to disk via {@link #commit(boolean)}.
+ */
+public class ManifestWorkspace {
+
+    private static final String MANIFEST_PREFIX = "META-INF/services/";
+    private static final Path MANAGED_PATH = Paths.get("connect-plugin-path-shim-0.0.1-SNAPSHOT.jar");
+    private final PrintStream out;
+    private final List<SourceWorkspace<?>> workspaces;
+    private final Map<Path, Path> temporaryOverlayFiles;
+
+    public ManifestWorkspace(PrintStream out) {
+        this.out = out;
+        workspaces = new ArrayList<>();
+        temporaryOverlayFiles = new HashMap<>();
+    }
+
+    public SourceWorkspace<?> forSource(PluginSource source) throws IOException {
+        SourceWorkspace<?> sourceWorkspace;
+        switch (source.type()) {
+            case CLASSPATH:
+                sourceWorkspace = new ClasspathWorkspace(source);
+                break;
+            case MULTI_JAR:
+                sourceWorkspace = new MultiJarWorkspace(source);
+                break;
+            case SINGLE_JAR:
+                sourceWorkspace = new SingleJarWorkspace(source);
+                break;
+            case CLASS_HIERARCHY:
+                sourceWorkspace = new ClassHierarchyWorkspace(source);
+                break;
+            default:
+                throw new IllegalStateException("Unknown source type " + source.type());
+        }
+        workspaces.add(sourceWorkspace);
+        return sourceWorkspace;
+    }
+
+    /**
+     * Commits all queued changes to disk
+     * @return true if any workspace wrote changes to disk, false if all workspaces did not have writes to apply
+     * @throws IOException if an error occurs reading or writing to the filesystem
+     * @throws TerseException if a path is not writable on disk and should be.
+     */
+    public boolean commit(boolean dryRun) throws IOException, TerseException {
+        boolean changed = false;
+        for (SourceWorkspace<?> workspace : workspaces) {
+            changed |= workspace.commit(dryRun);
+        }
+        return changed;
+    }
+
+    /**
+     * A workspace scoped to a single plugin source.
+     * <p>Buffers simulated reads and writes to the plugin path before they can be written to disk.
+     * @param <T> The data structure used by the workspace to store in-memory manifests internally.
+     */
+    public static abstract class SourceWorkspace<T> {
+        private final Path location;
+        private final PluginSource.Type type;
+        protected final T initial;
+        protected final T manifests;
+
+        private SourceWorkspace(PluginSource source) throws IOException {
+            this.location = source.location();
+            this.type = source.type();
+            this.initial = load(source);
+            this.manifests = load(source);
+        }
+
+        public Path location() {
+            return location;
+        }
+
+        public PluginSource.Type type() {
+            return type;
+        }
+
+        protected abstract T load(PluginSource source) throws IOException;
+
+        public abstract boolean hasManifest(PluginType type, String className);
+
+        public abstract void forEach(BiConsumer<String, PluginType> consumer);
+
+        public abstract void addManifest(PluginType type, String pluginClass);
+
+        public abstract void removeManifest(PluginType type, String pluginClass);
+
+        protected abstract boolean commit(boolean dryRun) throws TerseException, IOException;
+
+        protected static Map<PluginType, Set<String>> loadManifest(URL baseUrl) throws MalformedURLException {
+            Map<PluginType, Set<String>> manifests = new EnumMap<>(PluginType.class);
+            for (PluginType type : PluginType.values()) {
+                Set<String> result;
+                try {
+                    URL u = new URL(baseUrl, MANIFEST_PREFIX + type.superClass().getName());
+                    result = parse(u);
+                } catch (RuntimeException e) {
+                    result = new LinkedHashSet<>();
+                }
+                manifests.put(type, result);
+            }
+            return manifests;
+        }
+
+        protected static URL jarBaseUrl(URL fileUrl) throws MalformedURLException {
+            return new URL("jar", "", -1, fileUrl + "!/", null);
+        }
+
+        protected static void forEach(Map<PluginType, Set<String>> manifests, BiConsumer<String, PluginType> consumer) {
+            manifests.forEach((type, classNames) -> classNames.forEach(className -> consumer.accept(className, type)));
+        }
+    }
+
+    /**
+     * A single jar can only contain one manifest per plugin type.
+     */
+    private class SingleJarWorkspace extends SourceWorkspace<Map<PluginType, Set<String>>> {
+
+        private SingleJarWorkspace(PluginSource source) throws IOException {
+            super(source);
+        }
+
+        @Override
+        protected Map<PluginType, Set<String>> load(PluginSource source) throws IOException {
+            return loadManifest(jarBaseUrl(source.urls()[0]));
+        }
+
+        @Override
+        public boolean hasManifest(PluginType type, String className) {
+            return manifests.get(type).contains(className);
+        }
+
+        @Override
+        public void forEach(BiConsumer<String, PluginType> consumer) {
+            forEach(manifests, consumer);
+        }
+
+        @Override
+        public void addManifest(PluginType type, String pluginClass) {
+            manifests.get(type).add(pluginClass);
+        }
+
+        @Override
+        public void removeManifest(PluginType type, String pluginClass) {
+            manifests.get(type).remove(pluginClass);
+        }
+
+        @Override
+        protected boolean commit(boolean dryRun) throws IOException, TerseException {
+            if (startSync(dryRun, location(), initial, manifests)) {
+                rewriteJar(dryRun, location(), manifests);
+                return true;
+            }
+            return false;
+        }
+    }
+
+    /**
+     * A classpath workspace is backed by multiple jars, and that is not writable.
+     * The in-memory format is a map from jar path to the manifests contained in that jar.
+     * The control flow of the caller should not perform writes, so these exceptions indicate a bug in the program.
+     */
+    private class ClasspathWorkspace extends SourceWorkspace<Map<Path, Map<PluginType, Set<String>>>> {
+
+        private ClasspathWorkspace(PluginSource source) throws IOException {
+            super(source);
+        }
+
+        @Override
+        protected Map<Path, Map<PluginType, Set<String>>> load(PluginSource source) throws IOException {
+            Map<Path, Map<PluginType, Set<String>>> manifestsBySubLocation = new HashMap<>();
+            for (URL url : source.urls()) {
+                Path jarPath = Paths.get(url.getPath());
+                manifestsBySubLocation.put(jarPath, loadManifest(jarBaseUrl(url)));
+            }
+            return manifestsBySubLocation;
+        }
+
+        public boolean hasManifest(PluginType type, String className) {
+            return manifests.values()
+                    .stream()
+                    .map(m -> m.get(type))
+                    .anyMatch(s -> s.contains(className));
+        }
+
+        public void forEach(BiConsumer<String, PluginType> consumer) {
+            manifests.values().forEach(m -> forEach(m, consumer));
+        }
+
+        @Override
+        public void addManifest(PluginType type, String pluginClass) {
+            throw new UnsupportedOperationException("Cannot change the contents of the classpath");
+        }
+
+        @Override
+        public void removeManifest(PluginType type, String pluginClass) {
+            throw new UnsupportedOperationException("Cannot change the contents of the classpath");
+        }
+
+        @Override
+        protected boolean commit(boolean dryRun) throws IOException, TerseException {
+            // There is never anything to commit for the classpath
+            return false;
+        }
+    }
+
+    /**
+     * A multi-jar workspace is similar to the classpath workspace because it has multiple jars.
+     * However, the multi-jar workspace is writable, and injects a managed jar where it writes added manifests.
+     */
+    private class MultiJarWorkspace extends ClasspathWorkspace {
+
+        private MultiJarWorkspace(PluginSource source) throws IOException {
+            super(source);
+        }
+
+        @Override
+        protected Map<Path, Map<PluginType, Set<String>>> load(PluginSource source) throws IOException {
+            Map<Path, Map<PluginType, Set<String>>> manifests = super.load(source);
+            // In addition to the normal multi-jar paths, inject a managed jar where we can add manifests.
+            Path managedPath = source.location().resolve(MANAGED_PATH);
+            URL url = managedPath.toUri().toURL();
+            manifests.put(managedPath, loadManifest(jarBaseUrl(url)));
+            return manifests;
+        }
+
+        @Override
+        public void addManifest(PluginType type, String pluginClass) {
+            // Add plugins to the managed manifest
+            manifests.get(location().resolve(MANAGED_PATH)).get(type).add(pluginClass);
+        }
+
+        @Override
+        public void removeManifest(PluginType type, String pluginClass) {
+            // If a plugin appears in multiple manifests, remove it from all of them.
+            for (Map<PluginType, Set<String>> manifestState : manifests.values()) {
+                manifestState.get(type).remove(pluginClass);
+            }
+        }
+
+        @Override
+        public boolean commit(boolean dryRun) throws IOException, TerseException {
+            boolean changed = false;
+            for (Map.Entry<Path, Map<PluginType, Set<String>>> manifestSource : manifests.entrySet()) {
+                Path jarPath = manifestSource.getKey();
+                Map<PluginType, Set<String>> before = initial.get(jarPath);
+                Map<PluginType, Set<String>> after = manifestSource.getValue();
+                if (startSync(dryRun, jarPath, before, after)) {
+                    rewriteJar(dryRun, jarPath, after);
+                    changed = true;
+                }
+            }
+            return changed;
+        }
+    }
+
+    /**
+     * The class hierarchy is similar to the single-jar because there can only be one manifest per type.
+     * However, the path to that single manifest is accessed via the pluginLocation.
+     */
+    private class ClassHierarchyWorkspace extends SingleJarWorkspace {
+
+        private ClassHierarchyWorkspace(PluginSource source) throws IOException {
+            super(source);
+        }
+
+        @Override
+        protected Map<PluginType, Set<String>> load(PluginSource source) throws IOException {
+            return loadManifest(source.location().toUri().toURL());
+        }
+
+        protected boolean commit(boolean dryRun) throws IOException, TerseException {
+            if (startSync(dryRun, location(), initial, manifests)) {
+                rewriteClassHierarchyManifest(dryRun, location(), manifests);
+                return true;
+            }
+            return false;
+        }
+    }
+
+    private boolean startSync(boolean dryRun, Path syncLocation, Map<PluginType, Set<String>> before, Map<PluginType, Set<String>> after) {
+        Objects.requireNonNull(syncLocation, "syncLocation must be non-null");
+        Objects.requireNonNull(before, "before must be non-null");
+        Objects.requireNonNull(after, "after must be non-null");
+        if (before.equals(after)) {
+            return false;
+        }
+        Set<String> added = new HashSet<>();
+        after.values().forEach(added::addAll);
+        before.values().forEach(added::removeAll);
+        Set<String> removed = new HashSet<>();
+        before.values().forEach(removed::addAll);
+        after.values().forEach(removed::removeAll);
+        out.printf("%sSync\t\t%s Add %s Remove %s%n", dryRun ? "Dry Run " : "", syncLocation, added, removed);
+        return true;
+    }
+
+    /**
+     * Rewrite a jar on disk to contain a manifests containing the specified entries.
+     * Will create the jar file and manifests if they do not exist.
+     * Will delete the jar file and manifests if they become empty.
+     *
+     * @param dryRun        True if the rewrite should be applied, false if it should be simulated.
+     * @param jarPath       Path to a jar file for a plugin
+     * @param manifestState Map from plugin type to Class names of plugins which should appear in that manifest
+     */
+    private void rewriteJar(boolean dryRun, Path jarPath, Map<PluginType, Set<String>> manifestState) throws IOException, TerseException {
+        Objects.requireNonNull(jarPath, "jarPath must be non-null");
+        Objects.requireNonNull(manifestState, "manifestState must be non-null");
+        Path writableJar = getWritablePath(dryRun, jarPath);
+        if (!manifestState.values().stream().allMatch(Collection::isEmpty) && !Files.exists(writableJar)) {
+            out.printf("\tCreate\t%s%n", jarPath);
+            createJar(writableJar);
+        }
+        try (FileSystem jar = FileSystems.newFileSystem(
+                new URI("jar", writableJar.toUri().toString(), ""),
+                Collections.emptyMap()
+        )) {
+            Path zipRoot = jar.getRootDirectories().iterator().next();
+            rewriteClassHierarchyManifest(dryRun, zipRoot, manifestState);
+        } catch (URISyntaxException e) {
+            throw new IOException(e);
+        }
+        if (Files.exists(writableJar) && jarIsEmpty(writableJar)) {
+            Files.delete(writableJar);
+        }
+    }
+
+    private void createJar(Path path) throws IOException {
+        Objects.requireNonNull(path, "path must be non-null");
+        try (ZipOutputStream stream = new ZipOutputStream(Files.newOutputStream(
+                path,
+                StandardOpenOption.CREATE,
+                StandardOpenOption.WRITE,
+                StandardOpenOption.TRUNCATE_EXISTING
+        ))) {
+            stream.closeEntry();
+        }
+    }
+
+    private boolean jarIsEmpty(Path path) throws IOException {
+        Objects.requireNonNull(path, "path must be non-null");
+        try (ZipInputStream stream = new ZipInputStream(Files.newInputStream(
+                path,
+                StandardOpenOption.READ
+        ))) {
+            return stream.getNextEntry() == null;
+        }
+    }
+
+    /**
+     * Rewrite a multiple manifest files contained inside a class hierarchy.
+     * Will create the files and parent directories if they not exist.
+     * Will delete the files and parent directories within the plugin location if they become empty.
+     *
+     * @param dryRun         True if the rewrite should be applied, false if it should be simulated.
+     * @param pluginLocation Path to top-level of class hierarchy for a plugin
+     * @param manifestState  Map from plugin type to Class names of plugins which should appear in that manifest
+     */
+    private void rewriteClassHierarchyManifest(boolean dryRun, Path pluginLocation, Map<PluginType, Set<String>> manifestState) throws IOException, TerseException {
+        Objects.requireNonNull(pluginLocation, "pluginLocation must be non-null");
+        Objects.requireNonNull(manifestState, "manifestState must be non-null");
+        if (!Files.exists(pluginLocation)) {
+            throw new TerseException(pluginLocation + " does not exist");
+        }
+        if (!Files.isWritable(pluginLocation)) {
+            throw new TerseException(pluginLocation + " is not writable");
+        }
+        for (Map.Entry<PluginType, Set<String>> manifest : manifestState.entrySet()) {
+            PluginType type = manifest.getKey();
+            Set<String> elements = manifest.getValue();
+            Path manifestPath = pluginLocation.resolve(MANIFEST_PREFIX).resolve(type.superClass().getName());
+            rewriteManifestFile(dryRun, manifestPath, elements);
+        }
+    }
+
+    /**
+     * Rewrite a file, creating it and any parent directories if it does not exist, deleting it if specified

Review Comment:
   What does "deleting it if specified" mean? I would think that'd refer to a method parameter but there is no, e.g., `boolean delete` here.



-- 
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