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>

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.

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;
     }
 }

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);