fix: Read value-format metadata when the value contains the key fields (#436)

Signed-off-by: Mate Czagany <mate@datasqrl.com>

Author: mateczagany

connectors/kafka-safe-connector/src/main/java/org/apache/flink/streaming/connectors/kafka/table/SafeKafkaDynamicSource.java

@@ -684,12 +684,15 @@ KafkaRecordDeserializationSchema<RowData> createKafkaDeserializationSchema(
         final int adjustedPhysicalArity =
                 DataType.getFieldDataTypes(producedDataType).size() - metadataKeys.size();
 
-        // adjust value format projection to include value format's metadata columns at the end
+        // adjust value format projection to include value format's metadata columns at the end.
+        // Upstream starts the range at keyProjection.length + valueProjection.length, which counts
+        // the key fields twice with 'value.fields-include' = 'ALL' and drops the metadata columns.
+        // TODO: Once FLINK-41255 is released, align this with the upstream fix.
         final int[] adjustedValueProjection =
                 IntStream.concat(
                                 IntStream.of(valueProjection),
                                 IntStream.range(
-                                        keyProjection.length + valueProjection.length,
+                                        DataType.getFieldDataTypes(physicalDataType).size(),
                                         adjustedPhysicalArity))
                         .toArray();
 

connectors/kafka-safe-connector/src/test/java/org/apache/flink/streaming/connectors/kafka/table/SafeKafkaDynamicSourceTest.java

@@ -99,14 +99,32 @@ void shouldDropRecordWithoutKeyWhenReadingKeyFieldsFromValueIsDisabled() throws
         assertThat(deserialize(schema, "42", "42,Ada")).containsExactly(ROW);
     }
 
+    @Test
+    void shouldReadValueFormatMetadataWhenValueContainsKeyFields() throws Exception {
+        var schema = createSchema(ALL_FIELDS, false, false, List.of("value.m"));
+
+        assertThat(deserialize(schema, "42", "42,Ada,m"))
+                .containsExactly(
+                        GenericRowData.of(
+                                42L, StringData.fromString("Ada"), StringData.fromString("m")));
+    }
+
     private static KafkaRecordDeserializationSchema<RowData> createSchema(
             int[] valueProjection, boolean upsertMode) {
         return createSchema(valueProjection, upsertMode, true);
     }
 
-    @SuppressWarnings("unchecked")
     private static KafkaRecordDeserializationSchema<RowData> createSchema(
             int[] valueProjection, boolean upsertMode, boolean keyFieldsFromValue) {
+        return createSchema(valueProjection, upsertMode, keyFieldsFromValue, List.of());
+    }
+
+    @SuppressWarnings("unchecked")
+    private static KafkaRecordDeserializationSchema<RowData> createSchema(
+            int[] valueProjection,
+            boolean upsertMode,
+            boolean keyFieldsFromValue,
+            List<String> formatMetadataKeys) {
         var source =
                 new SafeKafkaDynamicSource(
                         PHYSICAL_TYPE,
@@ -134,17 +152,28 @@ private static KafkaRecordDeserializationSchema<RowData> createSchema(
                         Optional.empty(),
                         new SourceWatermarkConfig());
 
+        var producedFields = new ArrayList<>(DataType.getFields(PHYSICAL_TYPE));
+        formatMetadataKeys.forEach(
+                key -> producedFields.add(DataTypes.FIELD(key, DataTypes.STRING())));
+        source.applyReadableMetadata(
+                formatMetadataKeys, DataTypes.ROW(producedFields.toArray(DataTypes.Field[]::new)));
+
         var keyDeserialization =
                 new CsvDeserializationSchema(
                         fields -> GenericRowData.of(Long.parseLong(fields[0])));
         var valueDeserialization =
                 new CsvDeserializationSchema(
                         fields ->
-                                fields.length == 2
-                                        ? GenericRowData.of(
-                                                Long.parseLong(fields[0]),
-                                                StringData.fromString(fields[1]))
-                                        : GenericRowData.of(StringData.fromString(fields[0])));
+                                switch (fields.length) {
+                                    case 1 -> GenericRowData.of(StringData.fromString(fields[0]));
+                                    case 2 -> GenericRowData.of(
+                                            Long.parseLong(fields[0]),
+                                            StringData.fromString(fields[1]));
+                                    default -> GenericRowData.of(
+                                            Long.parseLong(fields[0]),
+                                            StringData.fromString(fields[1]),
+                                            StringData.fromString(fields[2]));
+                                });
 
         return source.createKafkaDeserializationSchema(
                 keyDeserialization, valueDeserialization, mock(TypeInformation.class));