Update

bakdata · Jan 14, 2025 · 262229a · 262229a
1 parent 5d77dd4
commit 262229a
Showing 1 changed file with 8 additions and 1 deletion.
diff --git a/large-message-connect/src/main/java/com/bakdata/kafka/LargeMessageConverterConfig.java b/large-message-connect/src/main/java/com/bakdata/kafka/LargeMessageConverterConfig.java
@@ -25,9 +25,11 @@
 package com.bakdata.kafka;
 
 import java.util.Map;
+import org.apache.kafka.common.KafkaException;
 import org.apache.kafka.common.config.ConfigDef;
 import org.apache.kafka.common.config.ConfigDef.Importance;
 import org.apache.kafka.common.config.ConfigDef.Type;
+import org.apache.kafka.common.utils.Utils;
 import org.apache.kafka.connect.converters.ByteArrayConverter;
 import org.apache.kafka.connect.storage.Converter;
 
@@ -76,7 +78,12 @@ private static ConfigDef configDef() {
     }
 
     Converter getConverter() {
-        return this.getConfiguredInstance(CONVERTER_CLASS_CONFIG, Converter.class);
+        final Class<?> converterClass = this.getClass(CONVERTER_CLASS_CONFIG);
+        final Object converter = Utils.newInstance(converterClass);
+        if (!(converter instanceof Converter)) {
+            throw new KafkaException(converterClass.getName() + " is not an instance of " + Converter.class.getName());
+        }
+        return (Converter) converter;
     }
 
 }