Refactored

Author: robvadai

src/integrationTest/scala/com/segence/kafka/connect/kafka/EmbeddedKafkaSpecSupport.scala

@@ -6,14 +6,16 @@ import io.confluent.kafka.serializers.{AbstractKafkaSchemaSerDeConfig, KafkaAvro
 import io.github.embeddedkafka.schemaregistry.connect.EmbeddedKafkaConnect._
 import io.github.embeddedkafka.schemaregistry.EmbeddedKafkaConfig
 import org.apache.kafka.common.serialization.{Deserializer, Serializer}
+import org.scalatest.wordspec.AnyWordSpecLike
+import org.scalatest.matchers.should.Matchers._
 
 import java.nio.file.Files
 import scala.jdk.CollectionConverters._
 import scala.util.{Failure, Success}
 import scala.concurrent.duration._
 import scala.language.postfixOps
 
-class KafkaConnectorSpec extends EmbeddedKafkaSpecSupport {
+class KafkaConnectorSpec extends AnyWordSpecLike {
 
   implicit val kafkaConfig: EmbeddedKafkaConfig = EmbeddedKafkaConfig(
     schemaRegistryPort = 6002

src/integrationTest/scala/com/segence/kafka/connect/kafka/KafkaConnectorSpec.scala

@@ -35,6 +35,13 @@ public class KafkaSinkTask extends SinkTask {
     private Converter keyConverter;
     private Converter valueConverter;
 
+    private Converter instantiateConverter(Map<String, String> configuration, ConnectorConfigurationEntry configKeyName) throws ClassNotFoundException, NoSuchMethodException, InvocationTargetException, InstantiationException, IllegalAccessException {
+        final var clazz = getClass().getClassLoader().loadClass(
+            configuration.get(configKeyName.getConfigKeyName())
+        );
+        return (Converter) clazz.getDeclaredConstructor().newInstance();
+    }
+
     /**
      * Returns the Kafka topic set
      *
@@ -91,24 +98,18 @@ public void start(Map<String, String> configuration) {
         }
 
         try {
-            final var clazz = getClass().getClassLoader().loadClass(
-                configuration.get(ConnectorConfigurationEntry.KEY_CONVERTER_CLASS.getConfigKeyName())
-            );
-            keyConverter = (Converter) clazz.getDeclaredConstructor().newInstance();
+            keyConverter = instantiateConverter(configuration, ConnectorConfigurationEntry.KEY_CONVERTER_CLASS);
             keyConverter.configure(ConnectorConfiguration.getKeyConverterProperties(configuration), true);
-            LOGGER.debug("Instantiated Key Converter class: {}", clazz);
+            LOGGER.debug("Instantiated Key Converter class: {}", keyConverter.getClass().getCanonicalName());
         } catch (ClassNotFoundException | NoSuchMethodException | IllegalAccessException
                  | InstantiationException | InvocationTargetException e) {
             throw new IllegalArgumentException(e);
         }
 
         try {
-            final var clazz = getClass().getClassLoader().loadClass(
-                configuration.get(ConnectorConfigurationEntry.VALUE_CONVERTER_CLASS.getConfigKeyName())
-            );
-            valueConverter = (Converter) clazz.getDeclaredConstructor().newInstance();
+            valueConverter = instantiateConverter(configuration, ConnectorConfigurationEntry.VALUE_CONVERTER_CLASS);
             valueConverter.configure(ConnectorConfiguration.getValueConverterProperties(configuration), false);
-            LOGGER.debug("Instantiated Value Converter class: {}", clazz);
+            LOGGER.debug("Instantiated Value Converter class: {}", valueConverter.getClass().getCanonicalName());
         } catch (ClassNotFoundException | NoSuchMethodException | IllegalAccessException
                  | InstantiationException | InvocationTargetException e) {
             throw new IllegalArgumentException(e);