Polling efficiency (#378)

Fixes for KCON-26 - Backoff when no data available.
Fixes for KCON-28 - Improve poll method

Creates an AbstractSourceTask in commons to handle response to poll and
backoff calculations as well as start, stop. Implementations need to
implement an Iterator that poll will call to retrieve data.

Private classes Timer and Backoff are created in AbstractSourceTask and
may be moved out at a later date if needed elsewhere.

Changes made to configurations to support configuration extraction in
AbstractSourceTask.

Modifications to S3SourceTask to operate under AbstractSourceTask.

Additional tests added

---------

Co-authored-by: ¨Claude <¨[email protected]¨>
Co-authored-by: Jarkko Jaakola <[email protected]>
Co-authored-by: Murali Basani <[email protected]>

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 {