root
/
kafka
mirror of https://github.com/apache/kafka.git
1
0
Fork 0
Code Issues Actions 4 Packages Projects Releases Wiki Activity

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 <rayokota@gmail.com>

Reviewers: Konstantine Karantasis <konstantine@confluent.io>, Ewen Cheslack-Postava <ewen@confluent.io>

Closes #5736 from rayokota/KAFKA-7476

(cherry picked from commit 3edd8e7333)
Signed-off-by: Ewen Cheslack-Postava <me@ewencp.org>
This commit is contained in:
Robert Yokota 2018-10-04 20:34:50 -07:00 committed by Ewen Cheslack-Postava
parent 59991f69cd
commit a9ca1079bd
2 changed files with 12 additions and 1 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
connect/api/src/main/java/org/apache/kafka/connect/data/SchemaProjector.java
View File

@ -160,7 +160,7 @@ public class SchemaProjector {
assert source.type().isPrimitive(); assert source.type().isPrimitive();
assert target.type().isPrimitive(); assert target.type().isPrimitive();
Object result; Object result;
if (isPromotable(source.type(), target.type())) { if (isPromotable(source.type(), target.type()) && record instanceof Number) {
Number numberRecord = (Number) record; Number numberRecord = (Number) record;
switch (target.type()) { switch (target.type()) {
case INT8: case INT8:

11
connect/api/src/test/java/org/apache/kafka/connect/data/SchemaProjectorTest.java
View File

@ -352,6 +352,17 @@ public class SchemaProjectorTest {
projected = SchemaProjector.project(Timestamp.SCHEMA, 34567L, Timestamp.SCHEMA); projected = SchemaProjector.project(Timestamp.SCHEMA, 34567L, Timestamp.SCHEMA);
assertEquals(34567L, projected); 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(); Schema namedSchema = SchemaBuilder.int32().name("invalidLogicalTypeName").build();
for (Schema logicalTypeSchema: logicalTypeSchemas) { for (Schema logicalTypeSchema: logicalTypeSchemas) {
try { try {