fix(api): fix connect.value.schema is not used when sink topic is computed from record

Author: fhussonnois

connect-file-pulse-api/src/main/java/io/streamthoughts/kafka/connect/filepulse/source/TypedFileRecord.java

@@ -28,6 +28,7 @@
 import org.apache.kafka.connect.source.SourceRecord;
 
 import java.util.Map;
+import java.util.Optional;
 import java.util.function.Function;
 
 import static io.streamthoughts.kafka.connect.filepulse.internal.StringUtils.isNotBlank;
@@ -64,15 +65,25 @@ public SourceRecord toSourceRecord(final Map<String, ?> sourcePartition,
                                        final ConnectSchemaMapperOptions options) {
         mapper.setKeepLeadingUnderscoreCharacters(options.isKeepSchemaLeadingUnderscore());
         final TypedStruct value = value();
+
+        final String targetTopic = isNotBlank(internalSourceRecordBuilder.topic()) ?
+                internalSourceRecordBuilder.topic() :
+                defaultTopic;
+
+        final Schema connectSchema = Optional
+                .ofNullable(connectSchemaSupplier.apply(targetTopic))
+                .orElse(connectSchemaSupplier.apply(defaultTopic));
+
         final Schema valueSchema;
-        final Schema connectSchema = connectSchemaSupplier.apply(
-            isNotBlank(internalSourceRecordBuilder.topic()) ? internalSourceRecordBuilder.topic() : defaultTopic
-        );
         if (options.isConnectSchemaMergeEnabled() && value != null) {
+            // Convert dynamic StrutSchema to static Connect Schema
             Schema recordValueSchema = value.schema().map(mapper, false);
             if (connectSchema != null) {
+                // Merge Record Connect Schema with either the previous one or the one
+                // passed through the connector's config.
                 valueSchema = SchemaMerger.merge(connectSchema, recordValueSchema);
             } else {
+                // Use the Record Connect Schema
                 valueSchema = recordValueSchema;
             }
         } else {

connect-file-pulse-api/src/main/java/io/streamthoughts/kafka/connect/filepulse/source/internal/ConnectSchemaMapper.java

@@ -226,7 +226,7 @@ private static Struct toConnectStruct(final Schema connectSchema, final TypedStr
                 if (!isOptional) {
                     throw new DataException(
                             "Failed to convert record to connect data. " +
-                            "Missing required connectField '" + fieldName + "' for record '" + recordName + "'"
+                            "Missing required field '" + fieldName + "' for record '" + recordName + "'"
                     );
                 }
                 continue;
@@ -243,7 +243,7 @@ private static Struct toConnectStruct(final Schema connectSchema, final TypedStr
                     Schema.Type arrayValueType = connectFieldSchema.valueSchema().type();
                     if (!arrayValueType.equals(dataSchemaType)) {
                         throw new DataException(
-                                "Failed to convert record connectField '" +
+                                "Failed to convert record field '" +
                                 recordName + "." + fieldName + "' to connect data. " +
                                 "Types do not match Array[" + arrayValueType + "]<>Array[" + dataSchemaType + "]"
                         );
@@ -264,9 +264,9 @@ else if (dataSchemaType.isPrimitive()) {
                     }
                 } else {
                     throw new DataException(
-                            "Failed to convert record connectField '" +
-                                    recordName + "." + fieldName + "' to connect data. " +
-                                    "Types do not match " + schemaType + "<>" + dataSchemaType
+                            "Failed to convert record field '" +
+                            recordName + "." + fieldName + "' to connect data. " +
+                            "Types do not match " + schemaType + "<>" + dataSchemaType
                     );
                 }
             }

connect-file-pulse-plugin/src/main/java/io/streamthoughts/kafka/connect/filepulse/source/FilePulseSourceTask.java

@@ -100,9 +100,8 @@ public void start(final Map<String, String> props) {
         taskConfig = new SourceTaskConfig(configProperties);
         connectorGroupName = props.get(CONNECT_NAME_CONFIG);
         offsetPolicy = taskConfig.getSourceOffsetPolicy();
-        valueSchemas.put(taskConfig.topic(), taskConfig.getValueConnectSchema());
         defaultTopic = taskConfig.topic();
-
+        valueSchemas.put(defaultTopic, taskConfig.getValueConnectSchema());
         try {
             sharedStore = new StateBackingStoreAccess(
                     connectorGroupName,