You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kafka.apache.org by jg...@apache.org on 2018/08/20 23:50:05 UTC
[kafka] branch 1.0 updated: KAFKA-5891;
Proper handling of LogicalTypes in Cast (#4633)
This is an automated email from the ASF dual-hosted git repository.
jgus pushed a commit to branch 1.0
in repository https://gitbox.apache.org/repos/asf/kafka.git
The following commit(s) were added to refs/heads/1.0 by this push:
new e774086 KAFKA-5891; Proper handling of LogicalTypes in Cast (#4633)
e774086 is described below
commit e77408655b9fce52078d3dee0fb000c5b4a78e02
Author: Maciej Bryński <ma...@brynski.pl>
AuthorDate: Tue Aug 21 01:43:11 2018 +0200
KAFKA-5891; Proper handling of LogicalTypes in Cast (#4633)
Currently logical types are dropped during Cast Transformation.
This patch fixes this behaviour.
Reviewers: Randall Hauch <rh...@gmail.com>, Jason Gustafson <ja...@confluent.io>
---
.../java/org/apache/kafka/connect/transforms/Cast.java | 18 +++++++++++-------
.../org/apache/kafka/connect/transforms/CastTest.java | 5 +++++
2 files changed, 16 insertions(+), 7 deletions(-)
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 17be48c..5bce3f3 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
@@ -165,13 +165,17 @@ public abstract class Cast<R extends ConnectRecord<R>> implements Transformation
} else {
builder = SchemaUtil.copySchemaBasics(valueSchema, SchemaBuilder.struct());
for (Field field : valueSchema.fields()) {
- SchemaBuilder fieldBuilder =
- convertFieldType(casts.containsKey(field.name()) ? casts.get(field.name()) : field.schema().type());
- if (field.schema().isOptional())
- fieldBuilder.optional();
- if (field.schema().defaultValue() != null)
- fieldBuilder.defaultValue(castValueToType(field.schema().defaultValue(), fieldBuilder.type()));
- builder.field(field.name(), fieldBuilder.build());
+ if (casts.containsKey(field.name())) {
+ SchemaBuilder fieldBuilder = convertFieldType(casts.get(field.name()));
+ if (field.schema().isOptional())
+ fieldBuilder.optional();
+ if (field.schema().defaultValue() != null)
+ fieldBuilder.defaultValue(castValueToType(field.schema().defaultValue(), fieldBuilder.type()));
+ builder.field(field.name(), fieldBuilder.build());
+ } else {
+ builder.field(field.name(), field.schema());
+ }
+
}
}
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 b190189..decd043 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
@@ -21,12 +21,14 @@ import org.apache.kafka.common.config.ConfigException;
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.Timestamp;
import org.apache.kafka.connect.errors.DataException;
import org.apache.kafka.connect.source.SourceRecord;
import org.junit.After;
import org.junit.Test;
import java.util.Collections;
+import java.util.Date;
import java.util.HashMap;
import java.util.Map;
@@ -304,6 +306,7 @@ public class CastTest {
builder.field("boolean", Schema.BOOLEAN_SCHEMA);
builder.field("string", Schema.STRING_SCHEMA);
builder.field("optional", Schema.OPTIONAL_FLOAT32_SCHEMA);
+ builder.field("timestamp", Timestamp.SCHEMA);
Schema supportedTypesSchema = builder.build();
Struct recordValue = new Struct(supportedTypesSchema);
@@ -315,6 +318,7 @@ public class CastTest {
recordValue.put("float64", -64.);
recordValue.put("boolean", true);
recordValue.put("string", "42");
+ recordValue.put("timestamp", new Date(0));
// optional field intentionally omitted
SourceRecord transformed = xformValue.apply(new SourceRecord(null, null, "topic", 0,
@@ -331,6 +335,7 @@ public class CastTest {
assertEquals(true, ((Struct) transformed.value()).schema().field("float64").schema().defaultValue());
assertEquals((byte) 1, ((Struct) transformed.value()).get("boolean"));
assertEquals(42, ((Struct) transformed.value()).get("string"));
+ assertEquals(new Date(0), ((Struct) transformed.value()).get("timestamp"));
assertNull(((Struct) transformed.value()).get("optional"));
}