diff --git a/avro-converter/src/main/java/io/confluent/copycat/avro/AvroData.java b/avro-converter/src/main/java/io/confluent/copycat/avro/AvroData.java index aeee70921e34..d9586d7cdd83 100644 --- a/avro-converter/src/main/java/io/confluent/copycat/avro/AvroData.java +++ b/avro-converter/src/main/java/io/confluent/copycat/avro/AvroData.java @@ -18,17 +18,20 @@ import io.confluent.kafka.serializers.NonRecordContainer; -import org.apache.avro.AvroRuntimeException; import org.apache.avro.generic.GenericEnumSymbol; import org.apache.avro.generic.GenericFixed; import org.apache.avro.generic.GenericRecord; import org.apache.avro.generic.GenericRecordBuilder; import org.apache.avro.generic.IndexedRecord; +import org.apache.kafka.copycat.data.Date; +import org.apache.kafka.copycat.data.Decimal; import org.apache.kafka.copycat.data.Field; import org.apache.kafka.copycat.data.Schema; import org.apache.kafka.copycat.data.SchemaAndValue; import org.apache.kafka.copycat.data.SchemaBuilder; import org.apache.kafka.copycat.data.Struct; +import org.apache.kafka.copycat.data.Time; +import org.apache.kafka.copycat.data.Timestamp; import org.apache.kafka.copycat.errors.DataException; import org.codehaus.jackson.JsonNode; import org.codehaus.jackson.node.ArrayNode; @@ -36,6 +39,7 @@ import org.codehaus.jackson.node.ObjectNode; import java.io.IOException; +import java.math.BigDecimal; import java.nio.ByteBuffer; import java.util.ArrayList; import java.util.Arrays; @@ -65,6 +69,7 @@ public class AvroData { public static final String COPYCAT_DOC_PROP = "copycat.doc"; public static final String COPYCAT_VERSION_PROP = "copycat.version"; public static final String COPYCAT_DEFAULT_VALUE_PROP = "copycat.default"; + public static final String COPYCAT_PARAMETERS_PROP = "copycat.parameters"; public static final String COPYCAT_TYPE_PROP = "copycat.type"; @@ -153,6 +158,94 @@ public class AvroData { .getElementType(); } + // Convert values in Copycat form into their logical types. These logical converters are discovered by logical type + // names specified in the field + private static final HashMap TO_COPYCAT_LOGICAL_CONVERTERS = new HashMap<>(); + static { + TO_COPYCAT_LOGICAL_CONVERTERS.put(Decimal.LOGICAL_NAME, new LogicalTypeConverter() { + @Override + public Object convert(Schema schema, Object value) { + if (value instanceof byte[]) { + return Decimal.toLogical(schema, (byte[]) value); + } else if (value instanceof ByteBuffer) { + return Decimal.toLogical(schema, ((ByteBuffer) value).array()); + } + throw new DataException("Invalid type for Decimal, underlying representation should be bytes but was " + value.getClass()); + } + }); + + TO_COPYCAT_LOGICAL_CONVERTERS.put(Date.LOGICAL_NAME, new LogicalTypeConverter() { + @Override + public Object convert(Schema schema, Object value) { + if (!(value instanceof Integer)) + throw new DataException("Invalid type for Date, underlying representation should be int32 but was " + value.getClass()); + return Date.toLogical(schema, (int) value); + } + }); + + TO_COPYCAT_LOGICAL_CONVERTERS.put(Time.LOGICAL_NAME, new LogicalTypeConverter() { + @Override + public Object convert(Schema schema, Object value) { + if (!(value instanceof Integer)) + throw new DataException("Invalid type for Time, underlying representation should be int32 but was " + value.getClass()); + return Time.toLogical(schema, (int) value); + } + }); + + TO_COPYCAT_LOGICAL_CONVERTERS.put(Timestamp.LOGICAL_NAME, new LogicalTypeConverter() { + @Override + public Object convert(Schema schema, Object value) { + if (!(value instanceof Long)) + throw new DataException("Invalid type for Timestamp, underlying representation should be int64 but was " + value.getClass()); + return Timestamp.toLogical(schema, (long) value); + } + }); + } + + private static final HashMap TO_AVRO_LOGICAL_CONVERTERS + = new HashMap<>(); + static { + TO_AVRO_LOGICAL_CONVERTERS.put(Decimal.LOGICAL_NAME, new LogicalTypeConverter() { + @Override + public Object convert(Schema schema, Object value) { + if (!(value instanceof BigDecimal)) + throw new DataException( + "Invalid type for Decimal, expected BigDecimal but was " + value.getClass()); + return Decimal.fromLogical(schema, (BigDecimal) value); + } + }); + + TO_AVRO_LOGICAL_CONVERTERS.put(Date.LOGICAL_NAME, new LogicalTypeConverter() { + @Override + public Object convert(Schema schema, Object value) { + if (!(value instanceof java.util.Date)) + throw new DataException( + "Invalid type for Date, expected Date but was " + value.getClass()); + return Date.fromLogical(schema, (java.util.Date) value); + } + }); + + TO_AVRO_LOGICAL_CONVERTERS.put(Time.LOGICAL_NAME, new LogicalTypeConverter() { + @Override + public Object convert(Schema schema, Object value) { + if (!(value instanceof java.util.Date)) + throw new DataException( + "Invalid type for Time, expected Date but was " + value.getClass()); + return Time.fromLogical(schema, (java.util.Date) value); + } + }); + + TO_AVRO_LOGICAL_CONVERTERS.put(Timestamp.LOGICAL_NAME, new LogicalTypeConverter() { + @Override + public Object convert(Schema schema, Object value) { + if (!(value instanceof java.util.Date)) + throw new DataException( + "Invalid type for Timestamp, expected Date but was " + value.getClass()); + return Timestamp.fromLogical(schema, (java.util.Date) value); + } + }); + } + /** * Convert this object, in Copycat data format, into an Avro object. */ @@ -166,24 +259,33 @@ public static Object fromCopycatData(Schema schema, Object value) { * been converted and makes the use of NonRecordContainer optional * @param schema the Copycat schema * @param avroSchema the corresponding - * @param value the Copycat data to convert + * @param logicalValue the Copycat data to convert, which may be a value for a logical type * @param requireContainer if true, wrap primitives, maps, and arrays in a NonRecordContainer * before returning them * @return the converted data */ private static Object fromCopycatData(Schema schema, org.apache.avro.Schema avroSchema, - Object value, boolean requireContainer) { - Schema.Type schemaType = schema != null ? schema.type() : schemaTypeForSchemalessJavaType(value); + Object logicalValue, boolean requireContainer) { + Schema.Type schemaType = schema != null ? schema.type() : schemaTypeForSchemalessJavaType(logicalValue); if (schemaType == null) { // Schemaless null data return maybeAddContainer(avroSchema, maybeWrapSchemaless(null, null, null), requireContainer); } boolean schemaOptional = schema != null ? schema.isOptional() : true; - if (!schemaOptional && value == null) { + if (!schemaOptional && logicalValue == null) { throw new DataException("Found null value for non-optional schema"); } + // If this is a logical type, convert it from the convenient Java type to the underlying + // serializeable format + Object value = logicalValue; + if (schema != null && schema.name() != null) { + LogicalTypeConverter logicalConverter = TO_AVRO_LOGICAL_CONVERTERS.get(schema.name()); + if (logicalConverter != null) + value = logicalConverter.convert(schema, logicalValue); + } + try { switch (schemaType) { case INT8: { @@ -456,6 +558,9 @@ public static org.apache.avro.Schema fromCopycatSchema(Schema schema) { baseSchema.addProp(COPYCAT_VERSION_PROP, JsonNodeFactory.instance.numberNode(schema.version())); } + if (schema.parameters() != null) { + baseSchema.addProp(COPYCAT_PARAMETERS_PROP, parametersFromCopycat(schema.parameters())); + } if (schema.defaultValue() != null) { baseSchema.addProp(COPYCAT_DEFAULT_VALUE_PROP, defaultValueFromCopycat(schema, schema.defaultValue())); @@ -582,6 +687,14 @@ private static JsonNode defaultValueFromCopycat(Schema schema, Object defaultVal } } + private static JsonNode parametersFromCopycat(Map params) { + ObjectNode result = JsonNodeFactory.instance.objectNode(); + for (Map.Entry entry : params.entrySet()) { + result.put(entry.getKey(), entry.getValue()); + } + return result; + } + /** * Convert the given object, in Avro format, into an Copycat data object. */ @@ -657,57 +770,67 @@ private static Object toCopycatData(Schema schema, Object value) { return null; } + Object converted = null; switch (schema.type()) { // Pass through types case INT32: { Integer intValue = (Integer) value; // Validate type - return value; + converted = value; + break; } case INT64: { Long longValue = (Long) value; // Validate type - return value; + converted = value; + break; } case FLOAT32: { Float floatValue = (Float) value; // Validate type - return value; + converted = value; + break; } case FLOAT64: { Double doubleValue = (Double) value; // Validate type - return value; + converted = value; + break; } case BOOLEAN: { Boolean boolValue = (Boolean) value; // Validate type - return value; + converted = value; + break; } case INT8: // Encoded as an Integer - return ((Integer) value).byteValue(); + converted = ((Integer) value).byteValue(); + break; case INT16: // Encoded as an Integer - return ((Integer) value).shortValue(); + converted = ((Integer) value).shortValue(); + break; case STRING: if (value instanceof String) { - return value; + converted = value; } else if (value instanceof CharSequence || value instanceof GenericEnumSymbol || value instanceof Enum) { - return value.toString(); + converted = value.toString(); } else { throw new DataException("Invalid class for string type, expecting String or " + "CharSequence but found " + value.getClass()); } + break; case BYTES: if (value instanceof byte[]) { - return ByteBuffer.wrap((byte[]) value); + converted = ByteBuffer.wrap((byte[]) value); } else if (value instanceof ByteBuffer) { - return value; + converted = value; } else { throw new DataException("Invalid class for bytes type, expecting byte[] or ByteBuffer " + "but found " + value.getClass()); } + break; case ARRAY: { Schema valueSchema = schema.valueSchema(); @@ -716,7 +839,8 @@ private static Object toCopycatData(Schema schema, Object value) { for (Object elem : original) { result.add(toCopycatData(valueSchema, elem)); } - return result; + converted = result; + break; } case MAP: { @@ -731,7 +855,7 @@ private static Object toCopycatData(Schema schema, Object value) { result.put(entry.getKey().toString(), toCopycatData(valueSchema, entry.getValue())); } - return result; + converted = result; } else { // Arbitrary keys List original = (List) value; @@ -743,8 +867,9 @@ private static Object toCopycatData(Schema schema, Object value) { Object convertedValue = toCopycatData(valueSchema, entry.get(avroValueFieldIndex)); result.put(convertedKey, convertedValue); } - return result; + converted = result; } + break; } case STRUCT: { @@ -760,12 +885,15 @@ private static Object toCopycatData(Schema schema, Object value) { if (isInstanceOfAvroSchemaTypeForSimpleSchema(fieldSchema, value) || (valueRecordSchema != null && valueRecordSchema.equals(fieldSchema))) { - return new Struct(schema).put(unionMemberFieldName(fieldSchema), - toCopycatData(fieldSchema, value)); + converted = new Struct(schema).put(unionMemberFieldName(fieldSchema), + toCopycatData(fieldSchema, value)); + break; } } - throw new DataException( - "Did not find matching union field for data: " + value.toString()); + if (converted == null) { + throw new DataException( + "Did not find matching union field for data: " + value.toString()); + } } else { IndexedRecord original = (IndexedRecord) value; Struct result = new Struct(schema); @@ -775,13 +903,21 @@ private static Object toCopycatData(Schema schema, Object value) { = toCopycatData(field.schema(), original.get(avroFieldIndex)); result.put(field, convertedFieldValue); } - return result; + converted = result; } + break; } default: throw new DataException("Unknown Copycat schema type: " + schema.type()); } + + if (schema != null && schema.name() != null) { + LogicalTypeConverter logicalConverter = TO_COPYCAT_LOGICAL_CONVERTERS.get(schema.name()); + if (logicalConverter != null) + converted = logicalConverter.convert(schema, converted); + } + return converted; } catch (ClassCastException e) { throw new DataException("Invalid type for " + schema.type() + ": " + value.getClass()); } @@ -950,6 +1086,24 @@ public static Schema toCopycatSchema(org.apache.avro.Schema schema, boolean forc builder.version(version.asInt()); } + JsonNode parameters = schema.getJsonProp(COPYCAT_PARAMETERS_PROP); + if (parameters != null) { + if (!parameters.isObject()) { + throw new DataException("Expected JSON object for schema parameters but found: " + + parameters); + } + Iterator> paramIt = parameters.getFields(); + while (paramIt.hasNext()) { + Map.Entry field = paramIt.next(); + JsonNode jsonValue = field.getValue(); + if (!jsonValue.isTextual()) { + throw new DataException("Expected schema parameter values to be strings but found: " + + jsonValue); + } + builder.parameter(field.getKey(), jsonValue.getTextValue()); + } + } + if (fieldDefaultVal == null) { fieldDefaultVal = schema.getJsonProp(COPYCAT_DEFAULT_VALUE_PROP); } @@ -1130,4 +1284,8 @@ private static String[] splitName(String fullName) { } return result; } + + private interface LogicalTypeConverter { + Object convert(Schema schema, Object value); + } } diff --git a/avro-converter/src/test/java/io/confluent/copycat/avro/AvroDataTest.java b/avro-converter/src/test/java/io/confluent/copycat/avro/AvroDataTest.java index 90bd62d3e45f..d0a53b81afb3 100644 --- a/avro-converter/src/test/java/io/confluent/copycat/avro/AvroDataTest.java +++ b/avro-converter/src/test/java/io/confluent/copycat/avro/AvroDataTest.java @@ -20,23 +20,52 @@ import org.apache.avro.generic.GenericRecord; import org.apache.avro.generic.GenericRecordBuilder; import org.apache.avro.util.Utf8; +import org.apache.kafka.copycat.data.Date; +import org.apache.kafka.copycat.data.Decimal; import org.apache.kafka.copycat.data.Schema; import org.apache.kafka.copycat.data.SchemaAndValue; import org.apache.kafka.copycat.data.SchemaBuilder; import org.apache.kafka.copycat.data.Struct; +import org.apache.kafka.copycat.data.Time; +import org.apache.kafka.copycat.data.Timestamp; import org.apache.kafka.copycat.errors.DataException; import org.codehaus.jackson.node.JsonNodeFactory; +import org.codehaus.jackson.node.ObjectNode; import org.junit.Test; +import java.math.BigDecimal; +import java.math.BigInteger; import java.nio.ByteBuffer; import java.util.Arrays; +import java.util.Calendar; import java.util.Collections; +import java.util.GregorianCalendar; +import java.util.TimeZone; import io.confluent.kafka.serializers.NonRecordContainer; import static org.junit.Assert.*; public class AvroDataTest { + private static final int TEST_SCALE = 2; + private static final BigDecimal TEST_DECIMAL = new BigDecimal(new BigInteger("156"), TEST_SCALE); + private static final byte[] TEST_DECIMAL_BYTES = new byte[]{0, -100}; + + private static final GregorianCalendar EPOCH; + private static final GregorianCalendar EPOCH_PLUS_TEN_THOUSAND_DAYS; + private static final GregorianCalendar EPOCH_PLUS_TEN_THOUSAND_MILLIS; + static { + EPOCH = new GregorianCalendar(1970, Calendar.JANUARY, 1, 0, 0, 0); + EPOCH.setTimeZone(TimeZone.getTimeZone("UTC")); + + EPOCH_PLUS_TEN_THOUSAND_DAYS = new GregorianCalendar(1970, Calendar.JANUARY, 1, 0, 0, 0); + EPOCH_PLUS_TEN_THOUSAND_DAYS.setTimeZone(TimeZone.getTimeZone("UTC")); + EPOCH_PLUS_TEN_THOUSAND_DAYS.add(Calendar.DATE, 10000); + + EPOCH_PLUS_TEN_THOUSAND_MILLIS = new GregorianCalendar(1970, Calendar.JANUARY, 1, 0, 0, 0); + EPOCH_PLUS_TEN_THOUSAND_MILLIS.setTimeZone(TimeZone.getTimeZone("UTC")); + EPOCH_PLUS_TEN_THOUSAND_MILLIS.add(Calendar.MILLISECOND, 10000); + } // Copycat -> Avro @@ -190,6 +219,7 @@ public void testFromCopycatComplex() { public void testFromCopycatOptionalPrimitiveWithMetadata() { Schema schema = SchemaBuilder.string(). doc("doc").defaultValue("foo").name("io.confluent.stringtype").version(2).optional() + .parameter("foo", "bar").parameter("baz", "baz") .build(); // Missing some metadata, used to validate missing properties on the Avro schema will cause @@ -206,6 +236,10 @@ public void testFromCopycatOptionalPrimitiveWithMetadata() { JsonNodeFactory.instance.numberNode(2)); avroStringSchema.addProp("copycat.doc", "doc"); avroStringSchema.addProp("copycat.default", "foo"); + ObjectNode params = JsonNodeFactory.instance.objectNode(); + params.put("foo", "bar"); + params.put("baz", "baz"); + avroStringSchema.addProp("copycat.parameters", params); org.apache.avro.Schema avroSchema = org.apache.avro.SchemaBuilder.builder().unionOf() .type(avroStringSchema).and() @@ -244,6 +278,45 @@ public void testFromCopycatRecordWithMetadata() { assertEquals(avroRecord, convertedRecord); } + @Test + public void testFromCopycatLogicalDecimal() { + org.apache.avro.Schema avroSchema = org.apache.avro.SchemaBuilder.builder().bytesType(); + avroSchema.addProp("copycat.name", "org.apache.kafka.copycat.data.Decimal"); + avroSchema.addProp("copycat.version", JsonNodeFactory.instance.numberNode(1)); + ObjectNode avroParams = JsonNodeFactory.instance.objectNode(); + avroParams.put("scale", "2"); + avroSchema.addProp("copycat.parameters", avroParams); + checkNonRecordConversion(avroSchema, ByteBuffer.wrap(TEST_DECIMAL_BYTES), + Decimal.schema(2), TEST_DECIMAL); + } + + @Test + public void testFromCopycatLogicalDate() { + org.apache.avro.Schema avroSchema = org.apache.avro.SchemaBuilder.builder().intType(); + avroSchema.addProp("copycat.name", "org.apache.kafka.copycat.data.Date"); + avroSchema.addProp("copycat.version", JsonNodeFactory.instance.numberNode(1)); + checkNonRecordConversion(avroSchema, 10000, Date.SCHEMA, + EPOCH_PLUS_TEN_THOUSAND_DAYS.getTime()); + } + + @Test + public void testFromCopycatLogicalTime() { + org.apache.avro.Schema avroSchema = org.apache.avro.SchemaBuilder.builder().intType(); + avroSchema.addProp("copycat.name", "org.apache.kafka.copycat.data.Time"); + avroSchema.addProp("copycat.version", JsonNodeFactory.instance.numberNode(1)); + checkNonRecordConversion(avroSchema, 10000, Time.SCHEMA, + EPOCH_PLUS_TEN_THOUSAND_MILLIS.getTime()); + } + + @Test + public void testFromCopycatLogicalTimestamp() { + org.apache.avro.Schema avroSchema = org.apache.avro.SchemaBuilder.builder().longType(); + avroSchema.addProp("copycat.name", "org.apache.kafka.copycat.data.Timestamp"); + avroSchema.addProp("copycat.version", JsonNodeFactory.instance.numberNode(1)); + java.util.Date date = new java.util.Date(); + checkNonRecordConversion(avroSchema, date.getTime(), Timestamp.SCHEMA, date); + } + @Test(expected = DataException.class) public void testFromCopycatMismatchSchemaPrimitive() { AvroData.fromCopycatData(Schema.OPTIONAL_BOOLEAN_SCHEMA, 12); @@ -460,6 +533,47 @@ public void testToCopycatRecord() { AvroData.toCopycatData(avroSchema, avroRecord)); } + // Avro -> Copycat: Copycat logical types + + @Test + public void testToCopycatDecimal() { + org.apache.avro.Schema avroSchema = org.apache.avro.SchemaBuilder.builder().bytesType(); + avroSchema.addProp("copycat.name", "org.apache.kafka.copycat.data.Decimal"); + avroSchema.addProp("copycat.version", JsonNodeFactory.instance.numberNode(1)); + ObjectNode avroParams = JsonNodeFactory.instance.objectNode(); + avroParams.put("scale", "2"); + avroSchema.addProp("copycat.parameters", avroParams); + assertEquals(new SchemaAndValue(Decimal.schema(2), TEST_DECIMAL), + AvroData.toCopycatData(avroSchema, TEST_DECIMAL_BYTES)); + } + + @Test + public void testToCopycatDate() { + org.apache.avro.Schema avroSchema = org.apache.avro.SchemaBuilder.builder().intType(); + avroSchema.addProp("copycat.name", "org.apache.kafka.copycat.data.Date"); + avroSchema.addProp("copycat.version", JsonNodeFactory.instance.numberNode(1)); + assertEquals(new SchemaAndValue(Date.SCHEMA, EPOCH_PLUS_TEN_THOUSAND_DAYS.getTime()), + AvroData.toCopycatData(avroSchema, 10000)); + } + + @Test + public void testToCopycatTime() { + org.apache.avro.Schema avroSchema = org.apache.avro.SchemaBuilder.builder().intType(); + avroSchema.addProp("copycat.name", "org.apache.kafka.copycat.data.Time"); + avroSchema.addProp("copycat.version", JsonNodeFactory.instance.numberNode(1)); + assertEquals(new SchemaAndValue(Time.SCHEMA, EPOCH_PLUS_TEN_THOUSAND_MILLIS.getTime()), + AvroData.toCopycatData(avroSchema, 10000)); + } + + @Test + public void testToCopycatTimestamp() { + org.apache.avro.Schema avroSchema = org.apache.avro.SchemaBuilder.builder().longType(); + avroSchema.addProp("copycat.name", "org.apache.kafka.copycat.data.Timestamp"); + avroSchema.addProp("copycat.version", JsonNodeFactory.instance.numberNode(1)); + java.util.Date date = new java.util.Date(); + assertEquals(new SchemaAndValue(Timestamp.SCHEMA, date), + AvroData.toCopycatData(avroSchema, date.getTime())); + } // Avro -> Copycat: Copycat types with no corresponding Avro type @@ -600,6 +714,7 @@ public void testToCopycatEnum() { public void testToCopycatOptionalPrimitiveWithCopycatMetadata() { Schema schema = SchemaBuilder.string(). doc("doc").defaultValue("foo").name("io.confluent.stringtype").version(2).optional() + .parameter("foo", "bar").parameter("baz", "baz") .build(); org.apache.avro.Schema avroStringSchema = org.apache.avro.SchemaBuilder.builder().stringType(); @@ -608,6 +723,10 @@ public void testToCopycatOptionalPrimitiveWithCopycatMetadata() { JsonNodeFactory.instance.numberNode(2)); avroStringSchema.addProp("copycat.doc", "doc"); avroStringSchema.addProp("copycat.default", "foo"); + ObjectNode params = JsonNodeFactory.instance.objectNode(); + params.put("foo", "bar"); + params.put("baz", "baz"); + avroStringSchema.addProp("copycat.parameters", params); org.apache.avro.Schema avroSchema = org.apache.avro.SchemaBuilder.builder().unionOf() .type(avroStringSchema).and()