fix(filters): fix wrong index for DelimitedRowIndex

streamthoughts · Apr 11, 2023 · df7c71f · df7c71f
1 parent e843cff
commit df7c71f
Show file tree

Hide file tree

Showing 3 changed files with 27 additions and 1 deletion.
diff --git a/...-pulse-api/src/main/java/io/streamthoughts/kafka/connect/filepulse/data/StructSchema.java b/...-pulse-api/src/main/java/io/streamthoughts/kafka/connect/filepulse/data/StructSchema.java
@@ -110,6 +110,15 @@ public List<TypedField> fields() {
         return ordered;
     }
 
+    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);
+        }
+        return ordered;
+    }
+
     void set(final String fieldName, final Schema fieldSchema) {
         if (fieldName == null || fieldName.isEmpty()) {
             throw new DataException("fieldName cannot be null.");

diff --git a/...ain/java/io/streamthoughts/kafka/connect/filepulse/filter/AbstractDelimitedRowFilter.java b/...ain/java/io/streamthoughts/kafka/connect/filepulse/filter/AbstractDelimitedRowFilter.java
@@ -75,7 +75,7 @@ public void configure(final Map<String, ?> configs) {
 
         this.schema = this.configs.schema();
         if (schema != null) {
-            final List<TypedField> fields = schema.fields();
+            final List<TypedField> fields = schema.fieldsByIndex();
             IntStream.range(0, fields.size()).forEach(i -> columnsTypesByIndex.put(i, fields.get(i)));
         }
     }

diff --git a/...rc/test/java/io/streamthoughts/kafka/connect/filepulse/filter/DelimitedRowFilterTest.java b/...rc/test/java/io/streamthoughts/kafka/connect/filepulse/filter/DelimitedRowFilterTest.java
@@ -158,6 +158,23 @@ public void should_use_configured_schema() {
         Assert.assertTrue(record.getBoolean("c3"));
     }
 
+    @Test
+    public void should_use_configured_schema_when_field_names_out_of_order() {
+        configs.put(READER_FIELD_COLUMNS_CONFIG, "x1:STRING;c2:INTEGER;y3:BOOLEAN");
+        filter.configure(configs, alias -> null);
+        RecordsIterable<TypedStruct> output = filter.apply(null, DEFAULT_STRUCT, false);
+        Assert.assertNotNull(output);
+        Assert.assertEquals(1, output.size());
+
+        final TypedStruct record = output.iterator().next();
+        Assert.assertEquals(Type.STRING, record.get("x1").type());
+        Assert.assertEquals(Type.INTEGER, record.get("c2").type());
+        Assert.assertEquals(Type.BOOLEAN, record.get("y3").type());
+        Assert.assertEquals("value1", record.getString("x1"));
+        Assert.assertEquals(2, record.getInt("c2").intValue());
+        Assert.assertTrue(record.getBoolean("y3"));
+    }
+
     @Test
     public void should_only_convert_non_empty_values_given_schema() {
         // Given