fix: Rename methods for readability and use standard infer

Author: RyanSkraba

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

@@ -32,6 +32,7 @@
 import org.apache.kafka.connect.data.Schema;
 import org.apache.kafka.connect.data.SchemaBuilder;
 import org.apache.kafka.connect.data.Struct;
+import org.apache.kafka.connect.data.Values;
 import org.apache.kafka.connect.errors.DataException;
 import org.apache.kafka.connect.transforms.Transformation;
 
@@ -98,8 +99,14 @@ public void configure(final Map<String, ?> settings) {
     }
 
 
-    private void validateUnextractableKeyFields(final R record) {
-        // If the key fields have names, but the key value doesn't support names
+    /**
+     * Validation check to ensure that if any named fields exist, they can be extracted from the structured data in the
+     * key.
+     *
+     * @param record The incoming record to be transformed.
+     * @throws DataException if the key does not support extracting named fields.
+     */
+    private void validateNoUnextractableKeyFields(final R record) {
         if (keyFieldsHasNamedFields && record.keySchema() != null && record.keySchema().type() != Schema.Type.STRUCT) {
             throw new DataException(
                     String.format("Named key fields %s cannot be copied from the key schema: %s", valueFields.values(),
@@ -114,14 +121,24 @@ private void validateUnextractableKeyFields(final R record) {
         }
     }
 
-    private void validateKeyFieldSchemaRequired(final R record) {
+    /**
+     * Validation check to fail quickly when the entire key is copied into a structured value with a schema, but has a
+     * schemaless class.
+     *
+     * @param record The incoming record to be transformed.
+     * @throws DataException if the value class requires the key to have a schema but the key is a schemaless class.
+     */
+    private void validateKeySchemaRequirementsMet(final R record) {
         if (keyFieldsHasWildcard && record.keySchema() == null && record.key() instanceof Map) {
             if (record.valueSchema() != null && record.valueSchema().type() == Schema.Type.STRUCT) {
                 throw new DataException("The value requires a schema, but the key class is a schemaless Map");
             }
         }
     }
 
+    /**
+     * Validation check to ensure that the value can receive columns from the key, i.e. as either a Struct or Map.
+     */
     private void validateStructuredValue(final R record) {
         if (record.valueSchema() == null && !(record.value() instanceof Map)
                 || record.valueSchema() != null && record.valueSchema().type() != Schema.Type.STRUCT) {
@@ -131,15 +148,20 @@ private void validateStructuredValue(final R record) {
 
     @Override
     public R apply(final R record) {
-        validateUnextractableKeyFields(record);
-        validateKeyFieldSchemaRequired(record);
+        validateNoUnextractableKeyFields(record);
+        validateKeySchemaRequirementsMet(record);
         validateStructuredValue(record);
 
         if (record.value() instanceof Struct) {
             if (record.keySchema() != null) {
                 return applyToStruct(record, record.keySchema());
             } else {
-                return applyToStruct(record, inferSchemaFromPrimitive(record.key()));
+                final Schema inferredSchema = Values.inferSchema(record.key());
+                if (inferredSchema == null) {
+                    throw new DataException(
+                            "Cannot infer schema for unsupported key class: " + record.key().getClass().getName());
+                }
+                return applyToStruct(record, inferredSchema);
             }
         } else {
             return applyToMap(record);
@@ -200,34 +222,6 @@ private R applyToStruct(final R record, final Schema keySchema) {
                 newValue.schema(), newValue, record.timestamp());
     }
 
-    /**
-     * Infers the schema from a primitive key.
-     *
-     * @param key The key value.
-     * @return The inferred schema.
-     */
-    private Schema inferSchemaFromPrimitive(final Object key) {
-        if (key instanceof String) {
-            return Schema.OPTIONAL_STRING_SCHEMA;
-        } else if (key instanceof Boolean) {
-            return Schema.OPTIONAL_BOOLEAN_SCHEMA;
-        } else if (key instanceof Byte) {
-            return Schema.OPTIONAL_INT8_SCHEMA;
-        } else if (key instanceof Short) {
-            return Schema.OPTIONAL_INT16_SCHEMA;
-        } else if (key instanceof Integer) {
-            return Schema.OPTIONAL_INT32_SCHEMA;
-        } else if (key instanceof Long) {
-            return Schema.OPTIONAL_INT64_SCHEMA;
-        } else if (key instanceof Float) {
-            return Schema.OPTIONAL_FLOAT32_SCHEMA;
-        } else if (key instanceof Double) {
-            return Schema.OPTIONAL_FLOAT64_SCHEMA;
-        } else {
-            throw new DataException("Cannot infer schema for unsupported key class: " + key.getClass().getName());
-        }
-    }
-
     /**
      * Merges the key and value schemas into a new schema, according to the configuration of keyFields to copy into the
      * new value schema, and how they are renamed via valueFields.

src/test/java/io/aiven/kafka/connect/transforms/KeyToValueTest.java

@@ -49,9 +49,9 @@ public class KeyToValueTest {
     private static Struct createStruct(final Object... kvs) {
         SchemaBuilder sb = SchemaBuilder.struct();
         for (int i = 0; i < kvs.length; i += 2) {
-            Schema fieldSchema = Schema.OPTIONAL_STRING_SCHEMA;
+            Schema fieldSchema = Schema.STRING_SCHEMA;
             if (kvs[i + 1] instanceof Integer) {
-                fieldSchema = Schema.OPTIONAL_INT32_SCHEMA;
+                fieldSchema = Schema.INT32_SCHEMA;
             } else if (kvs[i + 1] instanceof Struct) {
                 fieldSchema = ((Struct) kvs[i + 1]).schema();
             }
@@ -107,7 +107,7 @@ public void schemaStructToSchemaStruct() {
     @Test
     public void schemaIntToSchemaStructWholeKey() {
         final int key = 123;
-        final SinkRecord input = new SinkRecord("", 0, Schema.OPTIONAL_INT32_SCHEMA, key, XYZ_STRUCT.schema(),
+        final SinkRecord input = new SinkRecord("", 0, Schema.INT32_SCHEMA, key, XYZ_STRUCT.schema(),
                 XYZ_STRUCT, 0);
 
         final Struct expectedValue = createStruct("x", 100, "y", 200, "z", 300, "_key", 123);
@@ -139,7 +139,7 @@ public void schemaStructToSchemaStructWithRenaming() {
     @Test
     public void schemaIntToSchemaStructWholeKeyWithRenaming() {
         final int key = 123;
-        final SinkRecord input = new SinkRecord("", 0, Schema.OPTIONAL_INT32_SCHEMA, key, XYZ_STRUCT.schema(),
+        final SinkRecord input = new SinkRecord("", 0, Schema.INT32_SCHEMA, key, XYZ_STRUCT.schema(),
                 XYZ_STRUCT, 0);
 
         final Struct expectedValue = createStruct("x", 123, "y", 123, "z", 300, "m", 123, "n", 123, "_key", 123);