feat(extract-topic): add support for schemaless values

Author: jeqo

src/main/java/io/aiven/kafka/connect/transforms/ExtractTopic.java

@@ -32,7 +32,7 @@
 
 public abstract class ExtractTopic<R extends ConnectRecord<R>> implements Transformation<R> {
 
-    private static final List<Schema.Type> SUPPORTED_TYPES_TO_CONVERT_FROM = Arrays.asList(
+    private static final List<Schema.Type> SUPPORTED_SCHEMA_TYPES_TO_CONVERT_FROM = Arrays.asList(
         Schema.Type.INT8,
         Schema.Type.INT16,
         Schema.Type.INT32,
@@ -43,6 +43,17 @@ public abstract class ExtractTopic<R extends ConnectRecord<R>> implements Transf
         Schema.Type.STRING
     );
 
+    private static final List<Class<?>> SUPPORTED_VALUE_CLASS_TO_CONVERT_FROM = Arrays.asList(
+        Byte.class,
+        Short.class,
+        Integer.class,
+        Long.class,
+        Double.class,
+        Float.class,
+        Boolean.class,
+        String.class
+    );
+
     private ExtractTopicConfig config;
 
     @Override
@@ -58,17 +69,25 @@ public void configure(final Map<String, ?> settings) {
     @Override
     public R apply(final R record) {
         final SchemaAndValue schemaAndValue = getSchemaAndValue(record);
-        if (schemaAndValue.schema() == null) {
-            throw new DataException(dataPlace() + " schema can't be null: " + record);
-        }
 
         final Optional<String> newTopic;
-        if (config.fieldName().isPresent()) {
-            newTopic = getNewTopicForNamedField(
-                record.toString(), schemaAndValue.schema(), schemaAndValue.value(), config.fieldName().get());
-        } else {
-            newTopic = getNewTopicWithoutFieldName(
-                record.toString(), schemaAndValue.schema(), schemaAndValue.value());
+
+        if (schemaAndValue.schema() == null) { // schemaless values (Map)
+            if (config.fieldName().isPresent()) {
+                newTopic = topicNameFromNamedFieldSchemaless(
+                    record.toString(), schemaAndValue.value(), config.fieldName().get());
+            } else {
+                newTopic = topicNameWithoutFieldNameSchemaless(
+                    record.toString(), schemaAndValue.value());
+            }
+        } else { // schema-based values (Struct)
+            if (config.fieldName().isPresent()) {
+                newTopic = topicNameFromNamedFieldWithSchema(
+                    record.toString(), schemaAndValue.schema(), schemaAndValue.value(), config.fieldName().get());
+            } else {
+                newTopic = topicNameWithoutFieldNameWithSchema(
+                    record.toString(), schemaAndValue.schema(), schemaAndValue.value());
+            }
         }
 
         if (newTopic.isPresent()) {
@@ -91,10 +110,66 @@ public R apply(final R record) {
 
     protected abstract SchemaAndValue getSchemaAndValue(final R record);
 
-    private Optional<String> getNewTopicForNamedField(final String recordStr,
-                                                      final Schema schema,
-                                                      final Object value,
-                                                      final String fieldName) {
+    private Optional<String> topicNameFromNamedFieldSchemaless(final String recordStr,
+                                                               final Object value,
+                                                               final String fieldName) {
+        if (value == null) {
+            throw new DataException(dataPlace() + " can't be null if field name is specified: " + recordStr);
+        }
+
+        if (!(value instanceof Map)) {
+            throw new DataException(dataPlace() + " type must be Map if field name is specified: " + recordStr);
+        }
+
+        @SuppressWarnings("unchecked") final Map<String, Object> valueMap = (Map<String, Object>) value;
+
+        final Optional<String> result = Optional.ofNullable(valueMap.get(fieldName))
+            .map(field -> {
+                if (!SUPPORTED_VALUE_CLASS_TO_CONVERT_FROM.contains(field.getClass())) {
+                    throw new DataException(fieldName + " type in " + dataPlace()
+                        + " " + value
+                        + " must be " + SUPPORTED_VALUE_CLASS_TO_CONVERT_FROM
+                        + ": " + recordStr);
+                }
+                return field;
+            })
+            .map(Object::toString);
+
+        if (result.isPresent() && !result.get().equals("")) {
+            return result;
+        } else {
+            if (config.skipMissingOrNull()) {
+                return Optional.empty();
+            } else {
+                throw new DataException(fieldName + " in " + dataPlace() + " can't be null or empty: " + recordStr);
+            }
+        }
+    }
+
+    private Optional<String> topicNameWithoutFieldNameSchemaless(final String recordStr,
+                                                                 final Object value) {
+        if (value == null || "".equals(value)) {
+            if (config.skipMissingOrNull()) {
+                return Optional.empty();
+            } else {
+                throw new DataException(dataPlace() + " can't be null or empty: " + recordStr);
+            }
+        }
+
+        if (!SUPPORTED_VALUE_CLASS_TO_CONVERT_FROM.contains(value.getClass())) {
+            throw new DataException("type in " + dataPlace()
+                + " " + value
+                + " must be " + SUPPORTED_VALUE_CLASS_TO_CONVERT_FROM
+                + ": " + recordStr);
+        }
+
+        return Optional.of(value.toString());
+    }
+
+    private Optional<String> topicNameFromNamedFieldWithSchema(final String recordStr,
+                                                               final Schema schema,
+                                                               final Object value,
+                                                               final String fieldName) {
         if (Schema.Type.STRUCT != schema.type()) {
             throw new DataException(dataPlace() + " schema type must be STRUCT if field name is specified: "
                 + recordStr);
@@ -113,9 +188,9 @@ private Optional<String> getNewTopicForNamedField(final String recordStr,
             }
         }
 
-        if (!SUPPORTED_TYPES_TO_CONVERT_FROM.contains(field.schema().type())) {
+        if (!SUPPORTED_SCHEMA_TYPES_TO_CONVERT_FROM.contains(field.schema().type())) {
             throw new DataException(fieldName + " schema type in " + dataPlace()
-                + " must be " + SUPPORTED_TYPES_TO_CONVERT_FROM
+                + " must be " + SUPPORTED_SCHEMA_TYPES_TO_CONVERT_FROM
                 + ": " + recordStr);
         }
 
@@ -134,12 +209,12 @@ private Optional<String> getNewTopicForNamedField(final String recordStr,
         }
     }
 
-    private Optional<String> getNewTopicWithoutFieldName(final String recordStr,
-                                                         final Schema schema,
-                                                         final Object value) {
-        if (!SUPPORTED_TYPES_TO_CONVERT_FROM.contains(schema.type())) {
+    private Optional<String> topicNameWithoutFieldNameWithSchema(final String recordStr,
+                                                                 final Schema schema,
+                                                                 final Object value) {
+        if (!SUPPORTED_SCHEMA_TYPES_TO_CONVERT_FROM.contains(schema.type())) {
             throw new DataException(dataPlace() + " schema type must be "
-                + SUPPORTED_TYPES_TO_CONVERT_FROM
+                + SUPPORTED_SCHEMA_TYPES_TO_CONVERT_FROM
                 + " if field name is not specified: "
                 + recordStr);
         }