From 8f30d88d933f200d0fe4ed3dc4d86d6b1995edc2 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Aindri=C3=BA=20Lavelle?=
 <121855584+aindriu-aiven@users.noreply.github.com>
Date: Tue, 4 Feb 2025 09:47:17 +0000
Subject: [PATCH] Reduce warnings from FilePatternUtils (#394)

Start Offset as used in the Sink Record is a long, this int needs to be
updated to use a long for conversion.

Signed-off-by: Aindriu Lavelle <aindriu.lavelle@aiven.io>
---
 .../connect/common/source/input/utils/FilePatternUtils.java | 4 ++--
 .../io/aiven/kafka/connect/common/source/task/Context.java  | 6 +++---
 .../common/source/input/utils/FilePatternUtilsTest.java     | 6 ++++--
 3 files changed, 9 insertions(+), 7 deletions(-)

diff --git a/commons/src/main/java/io/aiven/kafka/connect/common/source/input/utils/FilePatternUtils.java b/commons/src/main/java/io/aiven/kafka/connect/common/source/input/utils/FilePatternUtils.java
index 3f78431e..15923a10 100644
--- a/commons/src/main/java/io/aiven/kafka/connect/common/source/input/utils/FilePatternUtils.java
+++ b/commons/src/main/java/io/aiven/kafka/connect/common/source/input/utils/FilePatternUtils.java
@@ -145,9 +145,9 @@ private Optional<Integer> getPartitionId(final Matcher matcher, final String sou
 
     }
 
-    private Optional<Integer> getOffset(final Matcher matcher, final String sourceName) {
+    private Optional<Long> getOffset(final Matcher matcher, final String sourceName) {
         try {
-            return Optional.of(Integer.parseInt(matcher.group(PATTERN_START_OFFSET_KEY)));
+            return Optional.of(Long.parseLong(matcher.group(PATTERN_START_OFFSET_KEY)));
         } catch (IllegalArgumentException e) {
             // It is possible that when checking for the group it does not match and returns an
             // illegalStateException, Number format exception is also covered by this in this case.
diff --git a/commons/src/main/java/io/aiven/kafka/connect/common/source/task/Context.java b/commons/src/main/java/io/aiven/kafka/connect/common/source/task/Context.java
index ebbba42d..0d4c85ec 100644
--- a/commons/src/main/java/io/aiven/kafka/connect/common/source/task/Context.java
+++ b/commons/src/main/java/io/aiven/kafka/connect/common/source/task/Context.java
@@ -29,7 +29,7 @@ public class Context<K extends Comparable<K>> {
 
     private String topic;
     private Integer partition;
-    private Integer offset;
+    private Long offset;
     private K storageKey;
 
     public Context(final K storageKey) {
@@ -74,11 +74,11 @@ public final void setStorageKey(final K storageKey) {
         this.storageKey = storageKey;
     }
 
-    public final Optional<Integer> getOffset() {
+    public final Optional<Long> getOffset() {
         return Optional.ofNullable(offset);
     }
 
-    public final void setOffset(final Integer offset) {
+    public final void setOffset(final Long offset) {
         this.offset = offset;
     }
 }
diff --git a/commons/src/test/java/io/aiven/kafka/connect/common/source/input/utils/FilePatternUtilsTest.java b/commons/src/test/java/io/aiven/kafka/connect/common/source/input/utils/FilePatternUtilsTest.java
index 6df39ee8..a893fe88 100644
--- a/commons/src/test/java/io/aiven/kafka/connect/common/source/input/utils/FilePatternUtilsTest.java
+++ b/commons/src/test/java/io/aiven/kafka/connect/common/source/input/utils/FilePatternUtilsTest.java
@@ -48,9 +48,11 @@ void checkTopicDistribution(final String expectedSourceFormat, final String sour
             "{{partition}}-{{start_offset}}-{{topic}}.txt, logs2-1-logs2.txt, logs2,2,0001",
             "{{topic}}-{{start_offset}}-{{partition}}.txt, logs2-99999-0001.txt, logs2,1,99999",
             "{{partition}}-{{start_offset}}-{{topic}}.txt, logs0002-01-logs2.txt, logs2,2,0001",
-            "{{partition}}-{{start_offset}}-{{topic}}.txt, logs002-1-logs2.txt, logs2,2,0001", })
+            "{{partition}}-{{start_offset}}-{{topic}}.txt, logs002-1-logs2.txt, logs2,2,0001",
+            "{{partition}}-{{start_offset}}-{{topic}}.txt, logs002-9223372036854775807-logs2.txt, logs2,2,9223372036854775807",
+            "{{partition}}-{{start_offset}}-{{topic}}.txt, logs002-8273685692-logs2.txt, logs2,2,8273685692" })
     void checkTopicDistribution(final String expectedSourceFormat, final String sourceName, final String expectedTopic,
-            final int expectedPartition, final int expectedOffset) {
+            final int expectedPartition, final long expectedOffset) {
 
         final FilePatternUtils utils = new FilePatternUtils(expectedSourceFormat);
         final Optional<Context<String>> ctx = utils.process(sourceName);