Skip to content

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

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
mimaison merged 9 commits into from
Mar 3, 2021
Merged

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

Show file tree
Hide file tree
Changes from 6 commits
Commits
Show all changes
9 commits
Select commit Hold shift + click to select a range
dba4a40
Fix for https://issues.apache.org/jira/browse/KAFKA-12170
sknop Jan 10, 2021
fa801de
Fix cast byte array -> string for byte[]
sknop Jan 21, 2021
4213125
Fixes after failed style check
sknop Jan 25, 2021
4e74475
Changes made after code review by Mickael Maison (@mimaison)
sknop Feb 18, 2021
3028511
Address code review feedback: protect case where ByteBuffer has not a…
sknop Feb 22, 2021
411ceb3
Kafka has ready-made Utils class for extracting bytes from a ByteBuffer.
sknop Feb 23, 2021
99d26f4
Changed conversion to a Base64-encoded string instead of Hex
sknop Feb 25, 2021
e772b37
Replaced wildcard with single class imports.
sknop Feb 26, 2021
ff3239c
Updated Overview and Config after code review
sknop Mar 3, 2021
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
16 changes: 16 additions & 0 deletions connect/transforms/src/main/java/org/apache/kafka/connect/transforms/Cast.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand All @@ -39,6 +40,7 @@
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.nio.ByteBuffer;
import java.util.EnumSet;
import java.util.HashMap;
import java.util.List;
Expand Down Expand Up @@ -364,11 +366,25 @@ 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;
Comment thread
sknop marked this conversation as resolved.
return castByteArrayToString(Utils.readBytes(byteBuffer));
} else if (value instanceof byte[]) {
byte[] rawBytes = (byte[]) value;
return castByteArrayToString(rawBytes);
} else {
return value.toString();
}
}

private static String castByteArrayToString(byte[] array) {
StringBuilder sbuf = new StringBuilder();
for (byte b : array) {
sbuf.append(String.format("%02X", b));

@kkonstantine kkonstantine Feb 19, 2021

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Also, I'm a bit curious. Why are we selecting hex here and not base64 for example?
Hex of course is less efficient.
https://issues.apache.org/jira/browse/KAFKA-12170 does not mention anything about the rational, so I thought I'd ask.

@rhauch wdyt about the representation here?

@sknop sknop Feb 21, 2021

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The reason I suggested Hex representation was that in the use case I encountered - database PK field encoded as BINARY retrieved via CDC Source Connector - the representation in the SQL CLI was Hex, so choosing the same format made it easy to compare source and target.

I would suggest that base64 encoding would validate its own dedicated transformer, with an ability to choose padding, for example

@rhauch rhauch Feb 22, 2021

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

As noted in my previous comment, I agree with @kkonstantine that base64 would be preferable. Doing that would align better with the existing Values class used in Connect's header converter mechanism.

If we want to add support for multiple encodings, we would need to have a KIP since it would likely mean changing the SMT configuration.

@sknop sknop Feb 23, 2021

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

If base64 is the better choice we shall go with that. Would we need to adjust the JIRA ticket to reflect this?

I just ran a little test in the JShell:

jshell> byte[] byteArray = new byte[] {(byte) 0xFE, (byte) 0xDC, (byte) 0xBA, (byte) 0x98, 0x76, 0x54, 0x32, 0x10};
byteArray ==> byte[8] { -2, -36, -70, -104, 118, 84, 50, 16 }

jshell> Base64.getEncoder().encode(byteArray)
$2 ==> byte[12] { 47, 116, 121, 54, 109, 72, 90, 85, 77, 104, 65, 61 }

jshell> Base64.getEncoder().encodeToString(byteArray)
$3 ==> "/ty6mHZUMhA="

jshell> Base64.getEncoder().withoutPadding().encodeToString(byteArray)
$4 ==> "/ty6mHZUMhA"

Would you suggest using padding or not?

@rhauch rhauch Feb 23, 2021

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Connect's Values code just uses Base64.getEncoder().encodeToString((byte[]) value) and does not disable padding. Seems like we should follow the same pattern here. Thoughts, @kkonstantine ?

@kkonstantine kkonstantine Feb 24, 2021

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Yes, I agree. encodeToString is the method I had in mind. Used in Values and JsonConverter for a conversion of bytes to string. It's also more efficient.

}
return sbuf.toString();
}

protected abstract Schema operatingSchema(R record);

protected abstract Object operatingValue(R record);
Expand Down
19 changes: 18 additions & 1 deletion connect/transforms/src/test/java/org/apache/kafka/connect/transforms/CastTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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();
Expand All @@ -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);
Expand All @@ -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,
Expand All @@ -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("FEDCBA9876543210", ((Struct) transformed.value()).get("bytes"));
assertEquals("FEDCBA9876543210", ((Struct) transformed.value()).get("byteArray"));

assertNull(((Struct) transformed.value()).get("optional"));

Schema transformedSchema = ((Struct) transformed.value()).schema();
Expand All @@ -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());
}
Expand Down