fix(filters): fix fields by index logic

Author: TheWorkshopCom

connect-file-pulse-api/src/main/java/io/streamthoughts/kafka/connect/filepulse/data/StructSchema.java

@@ -109,10 +109,7 @@ public List<TypedField> fields() {
 
     public List<TypedField> fieldsByIndex() {
         ArrayList<TypedField> ordered = new ArrayList<>(fields.values());
-        // order elements in array to match field column index
-        for (TypedField field: fields.values()) {
-            ordered.add(field.index(),field);
-        }
+        ordered.sort(Comparator.comparing(TypedField::index));
         return ordered;
     }
 

connect-file-pulse-filters/src/test/java/io/streamthoughts/kafka/connect/filepulse/config/DelimitedRowFilterConfigTest.java

@@ -67,20 +67,29 @@ public void shouldCreateConfigGivenOverrideValues() {
     public void shouldCreateConfigGivenSchema() {
 
         Map<String, String> props = new HashMap<>() {{
-            put(READER_FIELD_COLUMNS_CONFIG, "field1:BOOLEAN;field2:INT32;field3:STRING");
+            put(READER_FIELD_COLUMNS_CONFIG, "x1:BOOLEAN;c2:INT32;y3:STRING");
         }};
 
         DelimitedRowFilterConfig config = new DelimitedRowFilterConfig(filter.configDef(), props);
 
         StructSchema schema = config.schema();
         Assert.assertNotNull(schema);
 
+        // Fields ordered by name (default)
         List<TypedField> fields = schema.fields();
         Assert.assertEquals(3, fields.size());
 
-        Assert.assertEquals("field1", fields.get(0).name());
-        Assert.assertEquals("field2", fields.get(1).name());
-        Assert.assertEquals("field3", fields.get(2).name());
+        Assert.assertEquals("c2", fields.get(0).name());
+        Assert.assertEquals("x1", fields.get(1).name());
+        Assert.assertEquals("y3", fields.get(2).name());
+
+        // Fields ordered by index
+        List<TypedField> fieldsByIndex = schema.fieldsByIndex();
+        Assert.assertEquals(3, fieldsByIndex.size());
+
+        Assert.assertEquals("x1", fieldsByIndex.get(0).name());
+        Assert.assertEquals("c2", fieldsByIndex.get(1).name());
+        Assert.assertEquals("y3", fieldsByIndex.get(2).name());
     }
 
 }