KAFKA-12170: Fix for Connect Cast SMT to correctly transform a Byte array into a string

connect/transforms/src/main/java/org/apache/kafka/connect/transforms/Cast.java

@@ -22,6 +22,7 @@
 import org.apache.kafka.common.cache.SynchronizedCache;
 import org.apache.kafka.common.config.ConfigDef;
 import org.apache.kafka.common.config.ConfigException;
+import org.apache.kafka.common.utils.Utils;
 import org.apache.kafka.connect.connector.ConnectRecord;
 import org.apache.kafka.connect.data.ConnectSchema;
 import org.apache.kafka.connect.data.Date;
@@ -39,6 +40,8 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import java.nio.ByteBuffer;
+import java.util.Base64;
 import java.util.EnumSet;
 import java.util.HashMap;
 import java.util.List;
@@ -56,7 +59,8 @@ public abstract class Cast<R extends ConnectRecord<R>> implements Transformation
     // allow casting nested fields.
     public static final String OVERVIEW_DOC =
             "Cast fields or the entire key or value to a specific type, e.g. to force an integer field to a smaller "
-                    + "width. Only simple primitive types are supported -- integers, floats, boolean, and string. "
+                    + "width. Cast from integers, floats, boolean and string to any other type, "
+                    + "and cast binary to string (base64 encoded)."
                     + "<p/>Use the concrete transformation type designed for the record key (<code>" + Key.class.getName() + "</code>) "
                     + "or value (<code>" + Value.class.getName() + "</code>).";
 
@@ -82,7 +86,7 @@ public String toString() {
             ConfigDef.Importance.HIGH,
             "List of fields and the type to cast them to of the form field1:type,field2:type to cast fields of "
                     + "Maps or Structs. A single type to cast the entire value. Valid types are int8, int16, int32, "
-                    + "int64, float32, float64, boolean, and string.");
+                    + "int64, float32, float64, boolean, and string. Note that binary fields can only be cast to string.");
 
     private static final String PURPOSE = "cast types";
 
@@ -364,6 +368,12 @@ private static String castToString(Object value) {
         if (value instanceof java.util.Date) {
             java.util.Date dateValue = (java.util.Date) value;
             return Values.dateFormatFor(dateValue).format(dateValue);
+        } else if (value instanceof ByteBuffer) {
+            ByteBuffer byteBuffer = (ByteBuffer) value;
+            return Base64.getEncoder().encodeToString(Utils.readBytes(byteBuffer));
+        } else if (value instanceof byte[]) {
+            byte[] rawBytes = (byte[]) value;
+            return Base64.getEncoder().encodeToString(rawBytes);
         } else {
             return value.toString();
         }

connect/transforms/src/test/java/org/apache/kafka/connect/transforms/CastTest.java

@@ -17,6 +17,7 @@
 
 package org.apache.kafka.connect.transforms;
 
+import java.nio.ByteBuffer;
 import java.util.Arrays;
 import java.util.List;
 import java.util.concurrent.TimeUnit;
@@ -425,7 +426,11 @@ public void castLogicalToString() {
     @Test
     public void castFieldsWithSchema() {
         Date day = new Date(MILLIS_PER_DAY);
-        xformValue.configure(Collections.singletonMap(Cast.SPEC_CONFIG, "int8:int16,int16:int32,int32:int64,int64:boolean,float32:float64,float64:boolean,boolean:int8,string:int32,bigdecimal:string,date:string,optional:int32"));
+        byte[] byteArray = new byte[] {(byte) 0xFE, (byte) 0xDC, (byte) 0xBA, (byte) 0x98, 0x76, 0x54, 0x32, 0x10};
+        ByteBuffer byteBuffer = ByteBuffer.wrap(Arrays.copyOf(byteArray, byteArray.length));
+
+        xformValue.configure(Collections.singletonMap(Cast.SPEC_CONFIG,
+                "int8:int16,int16:int32,int32:int64,int64:boolean,float32:float64,float64:boolean,boolean:int8,string:int32,bigdecimal:string,date:string,optional:int32,bytes:string,byteArray:string"));
 
         // Include an optional fields and fields with defaults to validate their values are passed through properly
         SchemaBuilder builder = SchemaBuilder.struct();
@@ -442,6 +447,9 @@ public void castFieldsWithSchema() {
         builder.field("date", org.apache.kafka.connect.data.Date.SCHEMA);
         builder.field("optional", Schema.OPTIONAL_FLOAT32_SCHEMA);
         builder.field("timestamp", Timestamp.SCHEMA);
+        builder.field("bytes", Schema.BYTES_SCHEMA);
+        builder.field("byteArray", Schema.BYTES_SCHEMA);
+
         Schema supportedTypesSchema = builder.build();
 
         Struct recordValue = new Struct(supportedTypesSchema);
@@ -456,6 +464,9 @@ public void castFieldsWithSchema() {
         recordValue.put("date", day);
         recordValue.put("string", "42");
         recordValue.put("timestamp", new Date(0));
+        recordValue.put("bytes", byteBuffer);
+        recordValue.put("byteArray", byteArray);
+
         // optional field intentionally omitted
 
         SourceRecord transformed = xformValue.apply(new SourceRecord(null, null, "topic", 0,
@@ -475,6 +486,9 @@ public void castFieldsWithSchema() {
         assertEquals("42", ((Struct) transformed.value()).get("bigdecimal"));
         assertEquals(Values.dateFormatFor(day).format(day), ((Struct) transformed.value()).get("date"));
         assertEquals(new Date(0), ((Struct) transformed.value()).get("timestamp"));
+        assertEquals("/ty6mHZUMhA=", ((Struct) transformed.value()).get("bytes"));
+        assertEquals("/ty6mHZUMhA=", ((Struct) transformed.value()).get("byteArray"));
+
         assertNull(((Struct) transformed.value()).get("optional"));
 
         Schema transformedSchema = ((Struct) transformed.value()).schema();
@@ -489,6 +503,9 @@ public void castFieldsWithSchema() {
         assertEquals(Schema.STRING_SCHEMA.type(), transformedSchema.field("bigdecimal").schema().type());
         assertEquals(Schema.STRING_SCHEMA.type(), transformedSchema.field("date").schema().type());
         assertEquals(Schema.OPTIONAL_INT32_SCHEMA.type(), transformedSchema.field("optional").schema().type());
+        assertEquals(Schema.STRING_SCHEMA.type(), transformedSchema.field("bytes").schema().type());
+        assertEquals(Schema.STRING_SCHEMA.type(), transformedSchema.field("byteArray").schema().type());
+
         // The following fields are not changed
         assertEquals(Timestamp.SCHEMA.type(), transformedSchema.field("timestamp").schema().type());
     }