content: /*
   * Licensed to the Apache Software Foundation (ASF) under one or more
   * contributor license agreements. See the NOTICE file distributed with
   * this work for additional information regarding copyright ownership.
   * The ASF licenses this file to You under the Apache License, Version 2.0
   * (the "License"); you may not use this file except in compliance with
   * the License. You may obtain a copy of the License at
   *
   *    http://www.apache.org/licenses/LICENSE-2.0
   *
   * Unless required by applicable law or agreed to in writing, software
   * distributed under the License is distributed on an "AS IS" BASIS,
   * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
   * See the License for the specific language governing permissions and
   * limitations under the License.
   */
  package org.apache.kafka.connect.runtime.isolation;
  
  import org.apache.kafka.common.config.ConfigDef;
  import org.apache.kafka.connect.components.Versioned;
  import org.apache.kafka.connect.connector.ConnectRecord;
  import org.apache.kafka.connect.data.Schema;
  import org.apache.kafka.connect.data.SchemaAndValue;
  import org.apache.kafka.connect.sink.SinkConnector;
  import org.apache.kafka.connect.source.SourceConnector;
  import org.apache.kafka.connect.storage.Converter;
  import org.apache.kafka.connect.storage.HeaderConverter;
  import org.apache.kafka.connect.tools.MockSinkConnector;
  import org.apache.kafka.connect.tools.MockSourceConnector;
  import org.apache.kafka.connect.transforms.Transformation;
  
  import org.junit.jupiter.api.BeforeEach;
  import org.junit.jupiter.api.Test;
  import org.junit.jupiter.api.io.TempDir;
  
  import java.io.IOException;
  import java.nio.file.Files;
  import java.nio.file.Path;
  import java.nio.file.Paths;
  import java.util.ArrayList;
  import java.util.Arrays;
  import java.util.Collections;
  import java.util.HashMap;
  import java.util.List;
  import java.util.Map;
  import java.util.SortedSet;
  import java.util.TreeSet;
  
  import static org.junit.jupiter.api.Assertions.assertEquals;
  import static org.junit.jupiter.api.Assertions.assertFalse;
  import static org.junit.jupiter.api.Assertions.assertTrue;
  
  public class PluginUtilsTest {
      @TempDir
      Path rootDir;
  
      private Path pluginPath;
  
      @BeforeEach
      public void setUp() throws Exception {
          pluginPath = rootDir.resolve("plugins");
          Files.createDirectories(pluginPath);
          pluginPath = pluginPath.toRealPath();
      }
  
      @Test
      public void testJavaLibraryClasses() {
          assertFalse(PluginUtils.shouldLoadInIsolation("java."));
          assertFalse(PluginUtils.shouldLoadInIsolation("java.lang.Object"));
          assertFalse(PluginUtils.shouldLoadInIsolation("java.lang.String"));
          assertFalse(PluginUtils.shouldLoadInIsolation("java.util.HashMap$Entry"));
          assertFalse(PluginUtils.shouldLoadInIsolation("java.io.Serializable"));
          assertFalse(PluginUtils.shouldLoadInIsolation("javax.rmi."));
          assertFalse(PluginUtils.shouldLoadInIsolation(
                  "javax.management.loading.ClassLoaderRepository")
          );
          assertFalse(PluginUtils.shouldLoadInIsolation("org.omg.CORBA."));
          assertFalse(PluginUtils.shouldLoadInIsolation("org.omg.CORBA.Object"));
          assertFalse(PluginUtils.shouldLoadInIsolation("org.w3c.dom."));
          assertFalse(PluginUtils.shouldLoadInIsolation("org.w3c.dom.traversal.TreeWalker"));
          assertFalse(PluginUtils.shouldLoadInIsolation("org.xml.sax."));
          assertFalse(PluginUtils.shouldLoadInIsolation("org.xml.sax.EntityResolver"));
      }
  
      @Test
      public void testThirdPartyClasses() {
          assertFalse(PluginUtils.shouldLoadInIsolation("org.slf4j."));
          assertFalse(PluginUtils.shouldLoadInIsolation("org.slf4j.LoggerFactory"));
      }
  
      @Test
      public void testKafkaDependencyClasses() {
          assertFalse(PluginUtils.shouldLoadInIsolation("org.apache.kafka.common."));
          assertFalse(PluginUtils.shouldLoadInIsolation(
                  "org.apache.kafka.common.config.AbstractConfig")
          );
          assertFalse(PluginUtils.shouldLoadInIsolation(
                  "org.apache.kafka.common.config.ConfigDef$Type")
          );
          assertFalse(PluginUtils.shouldLoadInIsolation(
                  "org.apache.kafka.common.serialization.Deserializer")
          );
          assertFalse(PluginUtils.shouldLoadInIsolation(
                  "org.apache.kafka.clients.producer.ProducerConfig")
          );
          assertFalse(PluginUtils.shouldLoadInIsolation(
                  "org.apache.kafka.clients.consumer.ConsumerConfig")
          );
          assertFalse(PluginUtils.shouldLoadInIsolation(
                  "org.apache.kafka.clients.admin.KafkaAdminClient")
          );
      }
  
      @Test
      public void testConnectApiClasses() {
          List<String> apiClasses = Arrays.asList(
              // Enumerate all packages and classes
              "org.apache.kafka.connect.",
              "org.apache.kafka.connect.components.",
              "org.apache.kafka.connect.components.Versioned",
              //"org.apache.kafka.connect.connector.policy.", isolated by default
              "org.apache.kafka.connect.connector.policy.ConnectorClientConfigOverridePolicy",
              "org.apache.kafka.connect.connector.policy.ConnectorClientConfigRequest",
              "org.apache.kafka.connect.connector.policy.ConnectorClientConfigRequest$ClientType",
              "org.apache.kafka.connect.connector.",
              "org.apache.kafka.connect.connector.Connector",
              "org.apache.kafka.connect.connector.ConnectorContext",
              "org.apache.kafka.connect.connector.ConnectRecord",
              "org.apache.kafka.connect.connector.Task",
              "org.apache.kafka.connect.data.",
              "org.apache.kafka.connect.data.ConnectSchema",
              "org.apache.kafka.connect.data.Date",
              "org.apache.kafka.connect.data.Decimal",
              "org.apache.kafka.connect.data.Field",
              "org.apache.kafka.connect.data.Schema",
              "org.apache.kafka.connect.data.SchemaAndValue",
              "org.apache.kafka.connect.data.SchemaBuilder",
              "org.apache.kafka.connect.data.SchemaProjector",
              "org.apache.kafka.connect.data.Struct",
              "org.apache.kafka.connect.data.Time",
              "org.apache.kafka.connect.data.Timestamp",
              "org.apache.kafka.connect.data.Values",
              "org.apache.kafka.connect.errors.",
              "org.apache.kafka.connect.errors.AlreadyExistsException",
              "org.apache.kafka.connect.errors.ConnectException",
              "org.apache.kafka.connect.errors.DataException",
              "org.apache.kafka.connect.errors.IllegalWorkerStateException",
              "org.apache.kafka.connect.errors.NotFoundException",
              "org.apache.kafka.connect.errors.RetriableException",
              "org.apache.kafka.connect.errors.SchemaBuilderException",
              "org.apache.kafka.connect.errors.SchemaProjectorException",
              "org.apache.kafka.connect.header.",
              "org.apache.kafka.connect.header.ConnectHeader",
              "org.apache.kafka.connect.header.ConnectHeaders",
              "org.apache.kafka.connect.header.Header",
              "org.apache.kafka.connect.header.Headers",
              "org.apache.kafka.connect.health.",
              "org.apache.kafka.connect.health.AbstractState",
              "org.apache.kafka.connect.health.ConnectClusterDetails",
              "org.apache.kafka.connect.health.ConnectClusterState",
              "org.apache.kafka.connect.health.ConnectorHealth",
              "org.apache.kafka.connect.health.ConnectorState",
              "org.apache.kafka.connect.health.ConnectorType",
              "org.apache.kafka.connect.health.TaskState",
              "org.apache.kafka.connect.rest.",
              "org.apache.kafka.connect.rest.ConnectRestExtension",
              "org.apache.kafka.connect.rest.ConnectRestExtensionContext",
              "org.apache.kafka.connect.sink.",
              "org.apache.kafka.connect.sink.SinkConnector",
              "org.apache.kafka.connect.sink.SinkRecord",
              "org.apache.kafka.connect.sink.SinkTask",
              "org.apache.kafka.connect.sink.SinkTaskContext",
              "org.apache.kafka.connect.sink.ErrantRecordReporter",
              "org.apache.kafka.connect.source.",
              "org.apache.kafka.connect.source.SourceConnector",
              "org.apache.kafka.connect.source.SourceRecord",
              "org.apache.kafka.connect.source.SourceTask",
              "org.apache.kafka.connect.source.SourceTaskContext",
              "org.apache.kafka.connect.storage.",
              "org.apache.kafka.connect.storage.Converter",
              "org.apache.kafka.connect.storage.ConverterConfig",
              "org.apache.kafka.connect.storage.ConverterType",
              "org.apache.kafka.connect.storage.HeaderConverter",
              "org.apache.kafka.connect.storage.OffsetStorageReader",
              //"org.apache.kafka.connect.storage.SimpleHeaderConverter", explicitly isolated
              //"org.apache.kafka.connect.storage.StringConverter", explicitly isolated
              "org.apache.kafka.connect.storage.StringConverterConfig",
              //"org.apache.kafka.connect.transforms.", isolated by default
              "org.apache.kafka.connect.transforms.Transformation",
              "org.apache.kafka.connect.transforms.predicates.Predicate",
              "org.apache.kafka.connect.util.",
              "org.apache.kafka.connect.util.ConnectorUtils"
          );
          // Classes in the API should never be loaded in isolation.
          for (String clazz : apiClasses) {
              assertFalse(PluginUtils.shouldLoadInIsolation(clazz),
                  clazz + " from 'api' is loaded in isolation but should not be");
          }
      }
  
      @Test
      public void testConnectRuntimeClasses() {
          // Only list packages, because there are too many classes.
          List<String> runtimeClasses = Arrays.asList(
              "org.apache.kafka.connect.cli.",
              //"org.apache.kafka.connect.connector.policy.", isolated by default
              //"org.apache.kafka.connect.converters.", isolated by default
              "org.apache.kafka.connect.runtime.",
              "org.apache.kafka.connect.runtime.distributed",
              "org.apache.kafka.connect.runtime.errors",
              "org.apache.kafka.connect.runtime.health",
              "org.apache.kafka.connect.runtime.isolation",
              "org.apache.kafka.connect.runtime.rest.",
              "org.apache.kafka.connect.runtime.rest.entities.",
              "org.apache.kafka.connect.runtime.rest.errors.",
              "org.apache.kafka.connect.runtime.rest.resources.",
              "org.apache.kafka.connect.runtime.rest.util.",
              "org.apache.kafka.connect.runtime.standalone.",
              "org.apache.kafka.connect.runtime.rest.",
              "org.apache.kafka.connect.storage.",
              "org.apache.kafka.connect.tools.",
              "org.apache.kafka.connect.util."
          );
          for (String clazz : runtimeClasses) {
              assertFalse(PluginUtils.shouldLoadInIsolation(clazz),
                  clazz + " from 'runtime' is loaded in isolation but should not be");
          }
      }
  
      @Test
      public void testAllowedRuntimeClasses() {
          List<String> jsonConverterClasses = Arrays.asList(
              "org.apache.kafka.connect.connector.policy.",
              "org.apache.kafka.connect.connector.policy.AbstractConnectorClientConfigOverridePolicy",
              "org.apache.kafka.connect.connector.policy.AllConnectorClientConfigOverridePolicy",
              "org.apache.kafka.connect.connector.policy.NoneConnectorClientConfigOverridePolicy",
              "org.apache.kafka.connect.connector.policy.PrincipalConnectorClientConfigOverridePolicy",
              "org.apache.kafka.connect.converters.",
              "org.apache.kafka.connect.converters.ByteArrayConverter",
              "org.apache.kafka.connect.converters.DoubleConverter",
              "org.apache.kafka.connect.converters.FloatConverter",
              "org.apache.kafka.connect.converters.IntegerConverter",
              "org.apache.kafka.connect.converters.LongConverter",
              "org.apache.kafka.connect.converters.NumberConverter",
              "org.apache.kafka.connect.converters.NumberConverterConfig",
              "org.apache.kafka.connect.converters.ShortConverter",
              //"org.apache.kafka.connect.storage.", not isolated by default
              "org.apache.kafka.connect.storage.StringConverter",
              "org.apache.kafka.connect.storage.SimpleHeaderConverter"
          );
          for (String clazz : jsonConverterClasses) {
              assertTrue(PluginUtils.shouldLoadInIsolation(clazz),
                  clazz + " from 'runtime' is not loaded in isolation but should be");
          }
      }
  
      @Test
      public void testTransformsClasses() {
          List<String> transformsClasses = Arrays.asList(
              "org.apache.kafka.connect.transforms.",
              "org.apache.kafka.connect.transforms.util.",
              "org.apache.kafka.connect.transforms.util.NonEmptyListValidator",
              "org.apache.kafka.connect.transforms.util.RegexValidator",
              "org.apache.kafka.connect.transforms.util.Requirements",
              "org.apache.kafka.connect.transforms.util.SchemaUtil",
              "org.apache.kafka.connect.transforms.util.SimpleConfig",
              "org.apache.kafka.connect.transforms.Cast",
              "org.apache.kafka.connect.transforms.Cast$Key",
              "org.apache.kafka.connect.transforms.Cast$Value",
              "org.apache.kafka.connect.transforms.ExtractField",
              "org.apache.kafka.connect.transforms.ExtractField$Key",
              "org.apache.kafka.connect.transforms.ExtractField$Value",
              "org.apache.kafka.connect.transforms.Flatten",
              "org.apache.kafka.connect.transforms.Flatten$Key",
              "org.apache.kafka.connect.transforms.Flatten$Value",
              "org.apache.kafka.connect.transforms.HoistField",
              "org.apache.kafka.connect.transforms.HoistField$Key",
              "org.apache.kafka.connect.transforms.HoistField$Key",
              "org.apache.kafka.connect.transforms.InsertField",
              "org.apache.kafka.connect.transforms.InsertField$Key",
              "org.apache.kafka.connect.transforms.InsertField$Value",
              "org.apache.kafka.connect.transforms.MaskField",
              "org.apache.kafka.connect.transforms.MaskField$Key",
              "org.apache.kafka.connect.transforms.MaskField$Value",
              "org.apache.kafka.connect.transforms.RegexRouter",
              "org.apache.kafka.connect.transforms.ReplaceField",
              "org.apache.kafka.connect.transforms.ReplaceField$Key",
              "org.apache.kafka.connect.transforms.ReplaceField$Value",
              "org.apache.kafka.connect.transforms.SetSchemaMetadata",
              "org.apache.kafka.connect.transforms.SetSchemaMetadata$Key",
              "org.apache.kafka.connect.transforms.SetSchemaMetadata$Value",
              "org.apache.kafka.connect.transforms.TimestampConverter",
              "org.apache.kafka.connect.transforms.TimestampConverter$Key",
              "org.apache.kafka.connect.transforms.TimestampConverter$Value",
              "org.apache.kafka.connect.transforms.TimestampRouter",
              "org.apache.kafka.connect.transforms.TimestampRouter$Key",
              "org.apache.kafka.connect.transforms.TimestampRouter$Value",
              "org.apache.kafka.connect.transforms.ValueToKey",
              "org.apache.kafka.connect.transforms.predicates.",
              "org.apache.kafka.connect.transforms.predicates.HasHeaderKey",
              "org.apache.kafka.connect.transforms.predicates.RecordIsTombstone",
              "org.apache.kafka.connect.transforms.predicates.TopicNameMatches"
          );
          for (String clazz : transformsClasses) {
              assertTrue(PluginUtils.shouldLoadInIsolation(clazz),
                  clazz + " from 'transforms' is not loaded in isolation but should be");
          }
      }
  
      @Test
      public void testAllowedJsonConverterClasses() {
          List<String> jsonConverterClasses = Arrays.asList(
              "org.apache.kafka.connect.json.",
              "org.apache.kafka.connect.json.DecimalFormat",
              "org.apache.kafka.connect.json.JsonConverter",
              "org.apache.kafka.connect.json.JsonConverterConfig",
              "org.apache.kafka.connect.json.JsonDeserializer",
              "org.apache.kafka.connect.json.JsonSchema",
              "org.apache.kafka.connect.json.JsonSerializer"
          );
          for (String clazz : jsonConverterClasses) {
              assertTrue(PluginUtils.shouldLoadInIsolation(clazz),
                  clazz + " from 'json' is not loaded in isolation but should be");
          }
      }
  
      @Test
      public void testAllowedFileConnectors() {
          List<String> jsonConverterClasses = Arrays.asList(
              "org.apache.kafka.connect.file.",
              "org.apache.kafka.connect.file.FileStreamSinkConnector",
              "org.apache.kafka.connect.file.FileStreamSinkTask",
              "org.apache.kafka.connect.file.FileStreamSourceConnector",
              "org.apache.kafka.connect.file.FileStreamSourceTask"
          );
          for (String clazz : jsonConverterClasses) {
              assertTrue(PluginUtils.shouldLoadInIsolation(clazz),
                  clazz + " from 'file' is not loaded in isolation but should be");
          }
      }
  
      @Test
      public void testAllowedBasicAuthExtensionClasses() {
          List<String> basicAuthExtensionClasses = Arrays.asList(
              "org.apache.kafka.connect.rest.basic.auth.extension.BasicAuthSecurityRestExtension"
              //"org.apache.kafka.connect.rest.basic.auth.extension.JaasBasicAuthFilter", TODO fix?
              //"org.apache.kafka.connect.rest.basic.auth.extension.PropertyFileLoginModule" TODO fix?
          );
          for (String clazz : basicAuthExtensionClasses) {
              assertTrue(PluginUtils.shouldLoadInIsolation(clazz),
                  clazz + " from 'basic-auth-extension' is not loaded in isolation but should be");
          }
      }
  
      @Test
      public void testMirrorClasses() {
          assertTrue(PluginUtils.shouldLoadInIsolation(
                  "org.apache.kafka.connect.mirror.MirrorSourceTask")
          );
          assertTrue(PluginUtils.shouldLoadInIsolation(
                  "org.apache.kafka.connect.mirror.MirrorSourceConnector")
          );
      }
  
      @Test
      public void testClientConfigProvider() {
          assertFalse(PluginUtils.shouldLoadInIsolation(
                  "org.apache.kafka.common.config.provider.ConfigProvider")
          );
          assertTrue(PluginUtils.shouldLoadInIsolation(
                  "org.apache.kafka.common.config.provider.FileConfigProvider")
          );
          assertTrue(PluginUtils.shouldLoadInIsolation(
                  "org.apache.kafka.common.config.provider.FutureConfigProvider")
          );
      }
  
      @Test
      public void testEmptyPluginUrls() throws Exception {
          assertEquals(Collections.emptyList(), PluginUtils.pluginUrls(pluginPath));
      }
  
      @Test
      public void testEmptyStructurePluginUrls() throws Exception {
          createBasicDirectoryLayout();
          assertEquals(Collections.emptyList(), PluginUtils.pluginUrls(pluginPath));
      }
  
      @Test
      public void testPluginUrlsWithJars() throws Exception {
          createBasicDirectoryLayout();
  
          List<Path> expectedUrls = createBasicExpectedUrls();
  
          assertUrls(expectedUrls, PluginUtils.pluginUrls(pluginPath));
      }
  
      @Test
      public void testOrderOfPluginUrlsWithJars() throws Exception {
          createBasicDirectoryLayout();
          // Here this method is just used to create the files. The result is not used.
          createBasicExpectedUrls();
  
          List<Path> actual = PluginUtils.pluginUrls(pluginPath);
          // 'simple-transform.jar' is created first. In many cases, without sorting within the
          // PluginUtils, this jar will be placed before 'another-transform.jar'. However this is
          // not guaranteed because a DirectoryStream does not maintain a certain order in its
          // results. Besides this test case, sorted order in every call to assertUrls below.
          int i = Arrays.toString(actual.toArray()).indexOf("another-transform.jar");
          int j = Arrays.toString(actual.toArray()).indexOf("simple-transform.jar");
          assertTrue(i < j);
      }
  
      @Test
      public void testPluginUrlsWithZips() throws Exception {
          createBasicDirectoryLayout();
  
          List<Path> expectedUrls = new ArrayList<>();
          expectedUrls.add(Files.createFile(pluginPath.resolve("connectorA/my-sink.zip")));
          expectedUrls.add(Files.createFile(pluginPath.resolve("connectorB/a-source.zip")));
          expectedUrls.add(Files.createFile(pluginPath.resolve("transformC/simple-transform.zip")));
          expectedUrls.add(Files.createFile(
                  pluginPath.resolve("transformC/deps/another-transform.zip"))
          );
  
          assertUrls(expectedUrls, PluginUtils.pluginUrls(pluginPath));
      }
  
      @Test
      public void testPluginUrlsWithClasses() throws Exception {
          Files.createDirectories(pluginPath.resolve("org/apache/kafka/converters"));
          Files.createDirectories(pluginPath.resolve("com/mycompany/transforms"));
          Files.createDirectories(pluginPath.resolve("edu/research/connectors"));
          Files.createFile(pluginPath.resolve("org/apache/kafka/converters/README.txt"));
          Files.createFile(pluginPath.resolve("org/apache/kafka/converters/AlienFormat.class"));
          Files.createDirectories(pluginPath.resolve("com/mycompany/transforms/Blackhole.class"));
          Files.createDirectories(pluginPath.resolve("edu/research/connectors/HalSink.class"));
  
          List<Path> expectedUrls = new ArrayList<>();
          expectedUrls.add(pluginPath);
  
          assertUrls(expectedUrls, PluginUtils.pluginUrls(pluginPath));
      }
  
      @Test
      public void testPluginUrlsWithAbsoluteSymlink() throws Exception {
          createBasicDirectoryLayout();
  
          Path anotherPath = rootDir.resolve("moreplugins");
          Files.createDirectories(anotherPath);
          anotherPath = anotherPath.toRealPath();
          Files.createDirectories(anotherPath.resolve("connectorB-deps"));
          Files.createSymbolicLink(
                  pluginPath.resolve("connectorB/deps/symlink"),
                  anotherPath.resolve("connectorB-deps")
          );
  
          List<Path> expectedUrls = createBasicExpectedUrls();
          expectedUrls.add(Files.createFile(anotherPath.resolve("connectorB-deps/converter.jar")));
  
          assertUrls(expectedUrls, PluginUtils.pluginUrls(pluginPath));
      }
  
      @Test
      public void testPluginUrlsWithRelativeSymlinkBackwards() throws Exception {
          createBasicDirectoryLayout();
  
          Path anotherPath = rootDir.resolve("moreplugins");
          Files.createDirectories(anotherPath);
          anotherPath = anotherPath.toRealPath();
          Files.createDirectories(anotherPath.resolve("connectorB-deps"));
          Files.createSymbolicLink(
                  pluginPath.resolve("connectorB/deps/symlink"),
                  Paths.get("../../../moreplugins/connectorB-deps")
          );
  
          List<Path> expectedUrls = createBasicExpectedUrls();
          expectedUrls.add(Files.createFile(anotherPath.resolve("connectorB-deps/converter.jar")));
  
          assertUrls(expectedUrls, PluginUtils.pluginUrls(pluginPath));
      }
  
      @Test
      public void testPluginUrlsWithRelativeSymlinkForwards() throws Exception {
          // Since this test case defines a relative symlink within an already included path, the main
          // assertion of this test is absence of exceptions and correct resolution of paths.
          createBasicDirectoryLayout();
          Files.createDirectories(pluginPath.resolve("connectorB/deps/more"));
          Files.createSymbolicLink(
                  pluginPath.resolve("connectorB/deps/symlink"),
                  Paths.get("more")
          );
  
          List<Path> expectedUrls = createBasicExpectedUrls();
          expectedUrls.add(
                  Files.createFile(pluginPath.resolve("connectorB/deps/more/converter.jar"))
          );
  
          assertUrls(expectedUrls, PluginUtils.pluginUrls(pluginPath));
      }
  
      @Test
      public void testNonCollidingAliases() {
          SortedSet<PluginDesc<SinkConnector>> sinkConnectors = new TreeSet<>();
          sinkConnectors.add(new PluginDesc<>(MockSinkConnector.class, null, PluginType.SINK, MockSinkConnector.class.getClassLoader()));
          SortedSet<PluginDesc<SourceConnector>> sourceConnectors = new TreeSet<>();
          sourceConnectors.add(new PluginDesc<>(MockSourceConnector.class, null, PluginType.SOURCE, MockSourceConnector.class.getClassLoader()));
          SortedSet<PluginDesc<Converter>> converters = new TreeSet<>();
          converters.add(new PluginDesc<>(CollidingConverter.class, null, PluginType.CONVERTER, CollidingConverter.class.getClassLoader()));
          PluginScanResult result = new PluginScanResult(
                  sinkConnectors,
                  sourceConnectors,
                  converters,
                  Collections.emptySortedSet(),
                  Collections.emptySortedSet(),
                  Collections.emptySortedSet(),
                  Collections.emptySortedSet(),
                  Collections.emptySortedSet(),
                  Collections.emptySortedSet()
          );
          Map<String, String> actualAliases = PluginUtils.computeAliases(result);
          Map<String, String> expectedAliases = new HashMap<>();
          expectedAliases.put("MockSinkConnector", MockSinkConnector.class.getName());
          expectedAliases.put("MockSink", MockSinkConnector.class.getName());
          expectedAliases.put("MockSourceConnector", MockSourceConnector.class.getName());
          expectedAliases.put("MockSource", MockSourceConnector.class.getName());
          expectedAliases.put("CollidingConverter", CollidingConverter.class.getName());
          expectedAliases.put("Colliding", CollidingConverter.class.getName());
          assertEquals(expectedAliases, actualAliases);
      }
  
      @Test
      public void testMultiVersionAlias() {
          SortedSet<PluginDesc<SinkConnector>> sinkConnectors = new TreeSet<>();
          // distinct versions don't cause an alias collision (the class name is the same)
          sinkConnectors.add(new PluginDesc<>(MockSinkConnector.class, null, PluginType.SINK, MockSinkConnector.class.getClassLoader()));
          sinkConnectors.add(new PluginDesc<>(MockSinkConnector.class, "1.0", PluginType.SINK, MockSinkConnector.class.getClassLoader()));
          assertEquals(2, sinkConnectors.size());
          PluginScanResult result = new PluginScanResult(
                  sinkConnectors,
                  Collections.emptySortedSet(),
                  Collections.emptySortedSet(),
                  Collections.emptySortedSet(),
                  Collections.emptySortedSet(),
                  Collections.emptySortedSet(),
                  Collections.emptySortedSet(),
                  Collections.emptySortedSet(),
                  Collections.emptySortedSet()
          );
          Map<String, String> actualAliases = PluginUtils.computeAliases(result);
          Map<String, String> expectedAliases = new HashMap<>();
          expectedAliases.put("MockSinkConnector", MockSinkConnector.class.getName());
          expectedAliases.put("MockSink", MockSinkConnector.class.getName());
          assertEquals(expectedAliases, actualAliases);
      }
  
      @Test
      public void testCollidingPrunedAlias() {
          SortedSet<PluginDesc<Converter>> converters = new TreeSet<>();
          converters.add(new PluginDesc<>(CollidingConverter.class, null, PluginType.CONVERTER, CollidingConverter.class.getClassLoader()));
          SortedSet<PluginDesc<HeaderConverter>> headerConverters = new TreeSet<>();
          headerConverters.add(new PluginDesc<>(CollidingHeaderConverter.class, null, PluginType.HEADER_CONVERTER, CollidingHeaderConverter.class.getClassLoader()));
          PluginScanResult result = new PluginScanResult(
                  Collections.emptySortedSet(),
                  Collections.emptySortedSet(),
                  converters,
                  headerConverters,
                  Collections.emptySortedSet(),
                  Collections.emptySortedSet(),
                  Collections.emptySortedSet(),
                  Collections.emptySortedSet(),
                  Collections.emptySortedSet()
          );
          Map<String, String> actualAliases = PluginUtils.computeAliases(result);
          Map<String, String> expectedAliases = new HashMap<>();
          expectedAliases.put("CollidingConverter", CollidingConverter.class.getName());
          expectedAliases.put("CollidingHeaderConverter", CollidingHeaderConverter.class.getName());
          assertEquals(expectedAliases, actualAliases);
      }
  
      @SuppressWarnings("unchecked")
      @Test
      public void testCollidingSimpleAlias() {
          SortedSet<PluginDesc<Converter>> converters = new TreeSet<>();
          converters.add(new PluginDesc<>(CollidingConverter.class, null, PluginType.CONVERTER, CollidingConverter.class.getClassLoader()));
          SortedSet<PluginDesc<Transformation<?>>> transformations = new TreeSet<>();
          transformations.add(new PluginDesc<>((Class<? extends Transformation<?>>) (Class<?>) Colliding.class, null, PluginType.TRANSFORMATION, Colliding.class.getClassLoader()));
          PluginScanResult result = new PluginScanResult(
                  Collections.emptySortedSet(),
                  Collections.emptySortedSet(),
                  converters,
                  Collections.emptySortedSet(),
                  transformations,
                  Collections.emptySortedSet(),
                  Collections.emptySortedSet(),
                  Collections.emptySortedSet(),
                  Collections.emptySortedSet()
          );
          Map<String, String> actualAliases = PluginUtils.computeAliases(result);
          Map<String, String> expectedAliases = new HashMap<>();
          expectedAliases.put("CollidingConverter", CollidingConverter.class.getName());
          assertEquals(expectedAliases, actualAliases);
      }
  
      public static class CollidingConverter implements Converter, Versioned {
          @Override
          public void configure(Map<String, ?> configs, boolean isKey) {
          }
  
          @Override
          public byte[] fromConnectData(String topic, Schema schema, Object value) {
              return new byte[0];
          }
  
          @Override
          public SchemaAndValue toConnectData(String topic, byte[] value) {
              return null;
          }
  
          @Override
          public String version() {
              return "1.0";
          }
      }
  
      public static class CollidingHeaderConverter implements HeaderConverter, Versioned {
  
          @Override
          public SchemaAndValue toConnectHeader(String topic, String headerKey, byte[] value) {
              return null;
          }
  
          @Override
          public byte[] fromConnectHeader(String topic, String headerKey, Schema schema, Object value) {
              return new byte[0];
          }
  
          @Override
          public ConfigDef config() {
              return null;
          }
  
          @Override
          public void close() throws IOException {
          }
  
          @Override
          public void configure(Map<String, ?> configs) {
          }
  
          @Override
          public String version() {
              return "1.0";
          }
      }
  
      public static class Colliding<R extends ConnectRecord<R>> implements Transformation<R>, Versioned {
  
          @Override
          public String version() {
              return "1.0";
          }
  
          @Override
          public void configure(Map<String, ?> configs) {
          }
  
          @Override
          public R apply(R record) {
              return null;
          }
  
          @Override
          public ConfigDef config() {
              return null;
          }
  
          @Override
          public void close() {
          }
      }
  
      private void createBasicDirectoryLayout() throws IOException {
          Files.createDirectories(pluginPath.resolve("connectorA"));
          Files.createDirectories(pluginPath.resolve("connectorB/deps"));
          Files.createDirectories(pluginPath.resolve("transformC/deps"));
          Files.createDirectories(pluginPath.resolve("transformC/more-deps"));
          Files.createFile(pluginPath.resolve("transformC/more-deps/README.txt"));
      }
  
      private List<Path> createBasicExpectedUrls() throws IOException {
          List<Path> expectedUrls = new ArrayList<>();
          expectedUrls.add(Files.createFile(pluginPath.resolve("connectorA/my-sink.jar")));
          expectedUrls.add(Files.createFile(pluginPath.resolve("connectorB/a-source.jar")));
          expectedUrls.add(Files.createFile(pluginPath.resolve("transformC/simple-transform.jar")));
          expectedUrls.add(Files.createFile(
                  pluginPath.resolve("transformC/deps/another-transform.jar"))
          );
          return expectedUrls;
      }
  
      private void assertUrls(List<Path> expected, List<Path> actual) {
          Collections.sort(expected);
          // not sorting 'actual' because it should be returned sorted from within the PluginUtils.
          assertEquals(expected, actual);
      }
  }
  