Polling efficiency

commons/build.gradle.kts

@@ -87,6 +87,7 @@ dependencies {
   testImplementation(jackson.databind)
   testImplementation(testinglibs.mockito.core)
   testImplementation(testinglibs.assertj.core)
+  testImplementation(testinglibs.awaitility)
   testImplementation(testFixtures(project(":commons")))
   testImplementation(testinglibs.woodstox.stax2.api)
   testImplementation(apache.hadoop.mapreduce.client.core)

commons/src/main/java/io/aiven/kafka/connect/common/config/SourceCommonConfig.java

@@ -22,6 +22,8 @@
 
 import io.aiven.kafka.connect.common.config.enums.ErrorsTolerance;
 import io.aiven.kafka.connect.common.source.input.InputFormat;
+import io.aiven.kafka.connect.common.source.input.Transformer;
+import io.aiven.kafka.connect.common.source.input.TransformerFactory;
 
 public class SourceCommonConfig extends CommonConfig {
 
@@ -64,11 +66,15 @@ public String getTargetTopicPartitions() {
     }
 
     public ErrorsTolerance getErrorsTolerance() {
-        return ErrorsTolerance.forName(sourceConfigFragment.getErrorsTolerance());
+        return sourceConfigFragment.getErrorsTolerance();
     }
 
     public int getMaxPollRecords() {
         return sourceConfigFragment.getMaxPollRecords();
     }
 
+    public Transformer getTransformer() {
+        return TransformerFactory.getTransformer(schemaRegistryFragment.getInputFormat());
+    }
+
 }

commons/src/main/java/io/aiven/kafka/connect/common/config/SourceConfigFragment.java

@@ -88,8 +88,8 @@ public int getExpectedMaxMessageBytes() {
         return cfg.getInt(EXPECTED_MAX_MESSAGE_BYTES);
     }
 
-    public String getErrorsTolerance() {
-        return cfg.getString(ERRORS_TOLERANCE);
+    public ErrorsTolerance getErrorsTolerance() {
+        return ErrorsTolerance.forName(cfg.getString(ERRORS_TOLERANCE));
     }
 
     private static class ErrorsToleranceValidator implements ConfigDef.Validator {