diff --git a/connect/transforms/src/main/java/org/apache/kafka/connect/transforms/Cast.java b/connect/transforms/src/main/java/org/apache/kafka/connect/transforms/Cast.java index e872b336e8573..0a763cc24f85b 100644 --- a/connect/transforms/src/main/java/org/apache/kafka/connect/transforms/Cast.java +++ b/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> 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)." + "

Use the concrete transformation type designed for the record key (" + Key.class.getName() + ") " + "or value (" + Value.class.getName() + ")."; @@ -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(); } diff --git a/connect/transforms/src/test/java/org/apache/kafka/connect/transforms/CastTest.java b/connect/transforms/src/test/java/org/apache/kafka/connect/transforms/CastTest.java index 9c09554dc6875..7e90a44f504ea 100644 --- a/connect/transforms/src/test/java/org/apache/kafka/connect/transforms/CastTest.java +++ b/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()); }