Add Comparable interface to K

Signed-off-by: Aindriu Lavelle <[email protected]>
Aiven-Open · Jan 18, 2025 · f4a470b · f4a470b
1 parent 9525803
commit f4a470b
Show file tree

Hide file tree

Showing 4 changed files with 14 additions and 5 deletions.
diff --git a/commons/src/main/java/io/aiven/kafka/connect/common/config/SourceConfigFragment.java b/commons/src/main/java/io/aiven/kafka/connect/common/config/SourceConfigFragment.java
@@ -38,7 +38,7 @@ public final class SourceConfigFragment extends ConfigFragment {
     public static final String TARGET_TOPICS = "topics";
     public static final String ERRORS_TOLERANCE = "errors.tolerance";
 
-    public static final String DISTRIBUTION_TYPE = "object.distribution.strategy";
+    public static final String DISTRIBUTION_TYPE = "distribution.type";
 
     /**
      * Construct the ConfigFragment..

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
@@ -77,7 +77,7 @@ private Pattern configurePattern(final String expectedSourceNameFormat) {
         }
     }
 
-    public <K> Optional<Context<K>> process(final K sourceName) {
+    public <K extends Comparable<K>> Optional<Context<K>> process(final K sourceName) {
         final Optional<Matcher> matcher = fileMatches(sourceName.toString());
         if (matcher.isPresent()) {
             final Context<K> ctx = new Context<>(sourceName);

diff --git a/commons/src/main/java/io/aiven/kafka/connect/common/source/task/DistributionType.java b/commons/src/main/java/io/aiven/kafka/connect/common/source/task/DistributionType.java
@@ -69,8 +69,8 @@ public static DistributionType forName(final String name) {
                 return distributionType;
             }
         }
-        throw new ConfigException(String.format("Unknown object.distribution.strategy type: %s, allowed values %s ",
-                name, Arrays.toString(DistributionType.values())));
+        throw new ConfigException(String.format("Unknown distribution.type : %s, allowed values %s ", name,
+                Arrays.toString(DistributionType.values())));
     }
 
     /**

diff --git a/...src/test/java/io/aiven/kafka/connect/common/source/task/HashDistributionStrategyTest.java b/...src/test/java/io/aiven/kafka/connect/common/source/task/HashDistributionStrategyTest.java
@@ -115,12 +115,16 @@ private Context<String> getContext(final String expectedSourceName, final String
         return ctx.get();
     }
 
-    static class HashCodeKey {
+    static class HashCodeKey implements Comparable<HashCodeKey> {
         private final int hashCodeValue;
         public HashCodeKey(final int hashCodeValue) {
             this.hashCodeValue = hashCodeValue;
         }
 
+        private int getHashCodeValue() {
+            return hashCodeValue;
+        }
+
         @Override
         public boolean equals(final Object other) {
             if (this == other) {
@@ -137,5 +141,10 @@ public boolean equals(final Object other) {
         public int hashCode() {
             return hashCodeValue;
         }
+
+        @Override
+        public int compareTo(final HashCodeKey hashCodeKey) {
+            return Integer.compare(this.hashCodeValue, hashCodeKey.getHashCodeValue());
+        }
     }
 }