You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kafka.apache.org by ew...@apache.org on 2018/10/05 03:36:50 UTC
[kafka] branch 0.10.1 updated: KAFKA-7476: Fix Date-based types in
SchemaProjector
This is an automated email from the ASF dual-hosted git repository.
ewencp pushed a commit to branch 0.10.1
in repository https://gitbox.apache.org/repos/asf/kafka.git
The following commit(s) were added to refs/heads/0.10.1 by this push:
new daf38aa KAFKA-7476: Fix Date-based types in SchemaProjector
daf38aa is described below
commit daf38aa825b37b2a86e3c0238ee8e493d54e158f
Author: Robert Yokota <ra...@gmail.com>
AuthorDate: Thu Oct 4 20:34:50 2018 -0700
KAFKA-7476: Fix Date-based types in SchemaProjector
Various converters (AvroConverter and JsonConverter) produce a
SchemaAndValue consisting of a logical schema type and a java.util.Date.
This is a fix for SchemaProjector to properly handle the Date.
Author: Robert Yokota <ra...@gmail.com>
Reviewers: Konstantine Karantasis <ko...@confluent.io>, Ewen Cheslack-Postava <ew...@confluent.io>
Closes #5736 from rayokota/KAFKA-7476
(cherry picked from commit 3edd8e7333ec0bb32ab5ae4ec4814fe30bb8f91d)
Signed-off-by: Ewen Cheslack-Postava <me...@ewencp.org>
---
.../java/org/apache/kafka/connect/data/SchemaProjector.java | 2 +-
.../org/apache/kafka/connect/data/SchemaProjectorTest.java | 11 +++++++++++
2 files changed, 12 insertions(+), 1 deletion(-)
diff --git a/connect/api/src/main/java/org/apache/kafka/connect/data/SchemaProjector.java b/connect/api/src/main/java/org/apache/kafka/connect/data/SchemaProjector.java
index 6277e44..08ee37a 100644
--- a/connect/api/src/main/java/org/apache/kafka/connect/data/SchemaProjector.java
+++ b/connect/api/src/main/java/org/apache/kafka/connect/data/SchemaProjector.java
@@ -159,7 +159,7 @@ public class SchemaProjector {
assert source.type().isPrimitive();
assert target.type().isPrimitive();
Object result;
- if (isPromotable(source.type(), target.type())) {
+ if (isPromotable(source.type(), target.type()) && record instanceof Number) {
Number numberRecord = (Number) record;
switch (target.type()) {
case INT8:
diff --git a/connect/api/src/test/java/org/apache/kafka/connect/data/SchemaProjectorTest.java b/connect/api/src/test/java/org/apache/kafka/connect/data/SchemaProjectorTest.java
index 101be04..ef6d029 100644
--- a/connect/api/src/test/java/org/apache/kafka/connect/data/SchemaProjectorTest.java
+++ b/connect/api/src/test/java/org/apache/kafka/connect/data/SchemaProjectorTest.java
@@ -351,6 +351,17 @@ public class SchemaProjectorTest {
projected = SchemaProjector.project(Timestamp.SCHEMA, 34567L, Timestamp.SCHEMA);
assertEquals(34567L, projected);
+ java.util.Date date = new java.util.Date();
+
+ projected = SchemaProjector.project(Date.SCHEMA, date, Date.SCHEMA);
+ assertEquals(date, projected);
+
+ projected = SchemaProjector.project(Time.SCHEMA, date, Time.SCHEMA);
+ assertEquals(date, projected);
+
+ projected = SchemaProjector.project(Timestamp.SCHEMA, date, Timestamp.SCHEMA);
+ assertEquals(date, projected);
+
Schema namedSchema = SchemaBuilder.int32().name("invalidLogicalTypeName").build();
for (Schema logicalTypeSchema: logicalTypeSchemas) {
try {