Address PR Comments

Signed-off-by: Aindriu Lavelle <aindriu.lavelle@aiven.io>
Aiven-Open · Jan 27, 2025 · 197ae5f · 197ae5f
1 parent cb84ae7
commit 197ae5f
Show file tree

Hide file tree

Showing 3 changed files with 4 additions and 17 deletions.
diff --git a/commons/src/main/java/io/aiven/kafka/connect/common/source/OffsetManager.java b/commons/src/main/java/io/aiven/kafka/connect/common/source/OffsetManager.java
@@ -53,21 +53,8 @@ public class OffsetManager<E extends OffsetManager.OffsetManagerEntry<E>> {
      *            the context for this instance to use.
      */
     public OffsetManager(final SourceTaskContext context) {
-        this(context, new ConcurrentHashMap<>());
-    }
-
-    /**
-     * Package private for testing.
-     *
-     * @param context
-     *            the context for this instance to use.
-     * @param offsets
-     *            the offsets
-     */
-    protected OffsetManager(final SourceTaskContext context,
-            final ConcurrentMap<Map<String, Object>, Map<String, Object>> offsets) {
         this.context = context;
-        this.offsets = offsets;
+        this.offsets = new ConcurrentHashMap<>();
     }
 
     /**

diff --git a/...connector/src/integration-test/java/io/aiven/kafka/connect/s3/source/IntegrationBase.java b/...connector/src/integration-test/java/io/aiven/kafka/connect/s3/source/IntegrationBase.java
@@ -248,7 +248,7 @@ static List<String> consumeByteMessages(final String topic, final int expectedMe
             String bootstrapServers) {
         final Properties consumerProperties = getConsumerProperties(bootstrapServers, ByteArrayDeserializer.class,
                 ByteArrayDeserializer.class);
-        final List<byte[]> objects = consumeMessages(topic, expectedMessageCount, Duration.ofSeconds(90),
+        final List<byte[]> objects = consumeMessages(topic, expectedMessageCount, Duration.ofSeconds(120),
                 consumerProperties);
         return objects.stream().map(String::new).collect(Collectors.toList());
     }

diff --git a/s3-source-connector/src/main/java/io/aiven/kafka/connect/s3/source/utils/S3SourceRecord.java b/s3-source-connector/src/main/java/io/aiven/kafka/connect/s3/source/utils/S3SourceRecord.java
@@ -71,11 +71,11 @@ public void setValueData(final SchemaAndValue valueData) {
     }
 
     public String getTopic() {
-        return context.getTopic().isPresent() ? context.getTopic().get() : null;
+        return context.getTopic().orElse(null);
     }
 
     public Integer getPartition() {
-        return context.getPartition().isPresent() ? context.getPartition().get() : null;
+        return context.getPartition().orElse(null);
     }
 
     public String getObjectKey() {