kafka-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From gwens...@apache.org
Subject [10/26] kafka git commit: KAFKA-2774: Rename Copycat to Kafka Connect
Date Mon, 09 Nov 2015 06:11:29 GMT
http://git-wip-us.apache.org/repos/asf/kafka/blob/f2031d40/copycat/json/src/main/java/org/apache/kafka/copycat/json/JsonConverter.java
----------------------------------------------------------------------
diff --git a/copycat/json/src/main/java/org/apache/kafka/copycat/json/JsonConverter.java b/copycat/json/src/main/java/org/apache/kafka/copycat/json/JsonConverter.java
deleted file mode 100644
index ca8f029..0000000
--- a/copycat/json/src/main/java/org/apache/kafka/copycat/json/JsonConverter.java
+++ /dev/null
@@ -1,735 +0,0 @@
-/**
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements.  See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License.  You may obtain a copy of the License at
- * <p/>
- * http://www.apache.org/licenses/LICENSE-2.0
- * <p/>
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- **/
-
-package org.apache.kafka.copycat.json;
-
-import com.fasterxml.jackson.databind.JsonNode;
-import com.fasterxml.jackson.databind.node.ArrayNode;
-import com.fasterxml.jackson.databind.node.JsonNodeFactory;
-import com.fasterxml.jackson.databind.node.ObjectNode;
-import org.apache.kafka.common.cache.Cache;
-import org.apache.kafka.common.cache.LRUCache;
-import org.apache.kafka.common.cache.SynchronizedCache;
-import org.apache.kafka.common.errors.SerializationException;
-import org.apache.kafka.copycat.data.*;
-import org.apache.kafka.copycat.errors.DataException;
-import org.apache.kafka.copycat.storage.Converter;
-
-import java.io.IOException;
-import java.math.BigDecimal;
-import java.nio.ByteBuffer;
-import java.util.ArrayList;
-import java.util.Collection;
-import java.util.HashMap;
-import java.util.Iterator;
-import java.util.Map;
-
-/**
- * Implementation of Converter that uses JSON to store schemas and objects.
- */
-public class JsonConverter implements Converter {
-    private static final String SCHEMAS_ENABLE_CONFIG = "schemas.enable";
-    private static final boolean SCHEMAS_ENABLE_DEFAULT = true;
-    private static final String SCHEMAS_CACHE_CONFIG = "schemas.cache.size";
-    private static final int SCHEMAS_CACHE_SIZE_DEFAULT = 1000;
-
-    private static final HashMap<Schema.Type, JsonToCopycatTypeConverter> TO_COPYCAT_CONVERTERS = new HashMap<>();
-
-    private static Object checkOptionalAndDefault(Schema schema) {
-        if (schema.defaultValue() != null)
-            return schema.defaultValue();
-        if (schema.isOptional())
-            return null;
-        throw new DataException("Invalid null value for required field");
-    }
-
-    static {
-        TO_COPYCAT_CONVERTERS.put(Schema.Type.BOOLEAN, new JsonToCopycatTypeConverter() {
-            @Override
-            public Object convert(Schema schema, JsonNode value) {
-                if (value.isNull()) return checkOptionalAndDefault(schema);
-                return value.booleanValue();
-            }
-        });
-        TO_COPYCAT_CONVERTERS.put(Schema.Type.INT8, new JsonToCopycatTypeConverter() {
-            @Override
-            public Object convert(Schema schema, JsonNode value) {
-                if (value.isNull()) return checkOptionalAndDefault(schema);
-                return (byte) value.intValue();
-            }
-        });
-        TO_COPYCAT_CONVERTERS.put(Schema.Type.INT16, new JsonToCopycatTypeConverter() {
-            @Override
-            public Object convert(Schema schema, JsonNode value) {
-                if (value.isNull()) return checkOptionalAndDefault(schema);
-                return (short) value.intValue();
-            }
-        });
-        TO_COPYCAT_CONVERTERS.put(Schema.Type.INT32, new JsonToCopycatTypeConverter() {
-            @Override
-            public Object convert(Schema schema, JsonNode value) {
-                if (value.isNull()) return checkOptionalAndDefault(schema);
-                return value.intValue();
-            }
-        });
-        TO_COPYCAT_CONVERTERS.put(Schema.Type.INT64, new JsonToCopycatTypeConverter() {
-            @Override
-            public Object convert(Schema schema, JsonNode value) {
-                if (value.isNull()) return checkOptionalAndDefault(schema);
-                return value.longValue();
-            }
-        });
-        TO_COPYCAT_CONVERTERS.put(Schema.Type.FLOAT32, new JsonToCopycatTypeConverter() {
-            @Override
-            public Object convert(Schema schema, JsonNode value) {
-                if (value.isNull()) return checkOptionalAndDefault(schema);
-                return value.floatValue();
-            }
-        });
-        TO_COPYCAT_CONVERTERS.put(Schema.Type.FLOAT64, new JsonToCopycatTypeConverter() {
-            @Override
-            public Object convert(Schema schema, JsonNode value) {
-                if (value.isNull()) return checkOptionalAndDefault(schema);
-                return value.doubleValue();
-            }
-        });
-        TO_COPYCAT_CONVERTERS.put(Schema.Type.BYTES, new JsonToCopycatTypeConverter() {
-            @Override
-            public Object convert(Schema schema, JsonNode value) {
-                try {
-                    if (value.isNull()) return checkOptionalAndDefault(schema);
-                    return value.binaryValue();
-                } catch (IOException e) {
-                    throw new DataException("Invalid bytes field", e);
-                }
-            }
-        });
-        TO_COPYCAT_CONVERTERS.put(Schema.Type.STRING, new JsonToCopycatTypeConverter() {
-            @Override
-            public Object convert(Schema schema, JsonNode value) {
-                if (value.isNull()) return checkOptionalAndDefault(schema);
-                return value.textValue();
-            }
-        });
-        TO_COPYCAT_CONVERTERS.put(Schema.Type.ARRAY, new JsonToCopycatTypeConverter() {
-            @Override
-            public Object convert(Schema schema, JsonNode value) {
-                if (value.isNull()) return checkOptionalAndDefault(schema);
-
-                Schema elemSchema = schema == null ? null : schema.valueSchema();
-                ArrayList<Object> result = new ArrayList<>();
-                for (JsonNode elem : value) {
-                    result.add(convertToCopycat(elemSchema, elem));
-                }
-                return result;
-            }
-        });
-        TO_COPYCAT_CONVERTERS.put(Schema.Type.MAP, new JsonToCopycatTypeConverter() {
-            @Override
-            public Object convert(Schema schema, JsonNode value) {
-                if (value.isNull()) return checkOptionalAndDefault(schema);
-
-                Schema keySchema = schema == null ? null : schema.keySchema();
-                Schema valueSchema = schema == null ? null : schema.valueSchema();
-
-                // If the map uses strings for keys, it should be encoded in the natural JSON format. If it uses other
-                // primitive types or a complex type as a key, it will be encoded as a list of pairs. If we don't have a
-                // schema, we default to encoding in a Map.
-                Map<Object, Object> result = new HashMap<>();
-                if (schema == null || keySchema.type() == Schema.Type.STRING) {
-                    if (!value.isObject())
-                        throw new DataException("Map's with string fields should be encoded as JSON objects, but found " + value.getNodeType());
-                    Iterator<Map.Entry<String, JsonNode>> fieldIt = value.fields();
-                    while (fieldIt.hasNext()) {
-                        Map.Entry<String, JsonNode> entry = fieldIt.next();
-                        result.put(entry.getKey(), convertToCopycat(valueSchema, entry.getValue()));
-                    }
-                } else {
-                    if (!value.isArray())
-                        throw new DataException("Map's with non-string fields should be encoded as JSON array of tuples, but found " + value.getNodeType());
-                    for (JsonNode entry : value) {
-                        if (!entry.isArray())
-                            throw new DataException("Found invalid map entry instead of array tuple: " + entry.getNodeType());
-                        if (entry.size() != 2)
-                            throw new DataException("Found invalid map entry, expected length 2 but found :" + entry.size());
-                        result.put(convertToCopycat(keySchema, entry.get(0)),
-                                convertToCopycat(valueSchema, entry.get(1)));
-                    }
-                }
-                return result;
-            }
-        });
-        TO_COPYCAT_CONVERTERS.put(Schema.Type.STRUCT, new JsonToCopycatTypeConverter() {
-            @Override
-            public Object convert(Schema schema, JsonNode value) {
-                if (value.isNull()) return checkOptionalAndDefault(schema);
-
-                if (!value.isObject())
-                    throw new DataException("Structs should be encoded as JSON objects, but found " + value.getNodeType());
-
-                // We only have ISchema here but need Schema, so we need to materialize the actual schema. Using ISchema
-                // avoids having to materialize the schema for non-Struct types but it cannot be avoided for Structs since
-                // they require a schema to be provided at construction. However, the schema is only a SchemaBuilder during
-                // translation of schemas to JSON; during the more common translation of data to JSON, the call to schema.schema()
-                // just returns the schema Object and has no overhead.
-                Struct result = new Struct(schema.schema());
-                for (Field field : schema.fields())
-                    result.put(field, convertToCopycat(field.schema(), value.get(field.name())));
-
-                return result;
-            }
-        });
-    }
-
-    // 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<String, LogicalTypeConverter> 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[]))
-                    throw new DataException("Invalid type for Decimal, underlying representation should be bytes but was " + value.getClass());
-                return Decimal.toLogical(schema, (byte[]) value);
-            }
-        });
-
-        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<String, LogicalTypeConverter> TO_JSON_LOGICAL_CONVERTERS = new HashMap<>();
-    static {
-        TO_JSON_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_JSON_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_JSON_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_JSON_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);
-            }
-        });
-    }
-
-
-    private boolean enableSchemas = SCHEMAS_ENABLE_DEFAULT;
-    private int cacheSize = SCHEMAS_CACHE_SIZE_DEFAULT;
-    private Cache<Schema, ObjectNode> fromCopycatSchemaCache;
-    private Cache<JsonNode, Schema> toCopycatSchemaCache;
-
-    private final JsonSerializer serializer = new JsonSerializer();
-    private final JsonDeserializer deserializer = new JsonDeserializer();
-
-    @Override
-    public void configure(Map<String, ?> configs, boolean isKey) {
-        Object enableConfigsVal = configs.get(SCHEMAS_ENABLE_CONFIG);
-        if (enableConfigsVal != null)
-            enableSchemas = enableConfigsVal.toString().equals("true");
-
-        serializer.configure(configs, isKey);
-        deserializer.configure(configs, isKey);
-
-        Object cacheSizeVal = configs.get(SCHEMAS_CACHE_SIZE_DEFAULT);
-        if (cacheSizeVal != null)
-            cacheSize = (int) cacheSizeVal;
-        fromCopycatSchemaCache = new SynchronizedCache<>(new LRUCache<Schema, ObjectNode>(cacheSize));
-        toCopycatSchemaCache = new SynchronizedCache<>(new LRUCache<JsonNode, Schema>(cacheSize));
-    }
-
-    @Override
-    public byte[] fromCopycatData(String topic, Schema schema, Object value) {
-        JsonNode jsonValue = enableSchemas ? convertToJsonWithEnvelope(schema, value) : convertToJsonWithoutEnvelope(schema, value);
-        try {
-            return serializer.serialize(topic, jsonValue);
-        } catch (SerializationException e) {
-            throw new DataException("Converting Copycat data to byte[] failed due to serialization error: ", e);
-        }
-    }
-
-    @Override
-    public SchemaAndValue toCopycatData(String topic, byte[] value) {
-        JsonNode jsonValue;
-        try {
-            jsonValue = deserializer.deserialize(topic, value);
-        } catch (SerializationException e) {
-            throw new DataException("Converting byte[] to Copycat data failed due to serialization error: ", e);
-        }
-
-        if (enableSchemas && (jsonValue == null || !jsonValue.isObject() || jsonValue.size() != 2 || !jsonValue.has("schema") || !jsonValue.has("payload")))
-            throw new DataException("JsonDeserializer with schemas.enable requires \"schema\" and \"payload\" fields and may not contain additional fields");
-
-        // The deserialized data should either be an envelope object containing the schema and the payload or the schema
-        // was stripped during serialization and we need to fill in an all-encompassing schema.
-        if (!enableSchemas) {
-            ObjectNode envelope = JsonNodeFactory.instance.objectNode();
-            envelope.set("schema", null);
-            envelope.set("payload", jsonValue);
-            jsonValue = envelope;
-        }
-
-        return jsonToCopycat(jsonValue);
-    }
-
-    private SchemaAndValue jsonToCopycat(JsonNode jsonValue) {
-        if (jsonValue == null)
-            return SchemaAndValue.NULL;
-
-        if (!jsonValue.isObject() || jsonValue.size() != 2 || !jsonValue.has(JsonSchema.ENVELOPE_SCHEMA_FIELD_NAME) || !jsonValue.has(JsonSchema.ENVELOPE_PAYLOAD_FIELD_NAME))
-            throw new DataException("JSON value converted to Copycat must be in envelope containing schema");
-
-        Schema schema = asCopycatSchema(jsonValue.get(JsonSchema.ENVELOPE_SCHEMA_FIELD_NAME));
-        return new SchemaAndValue(schema, convertToCopycat(schema, jsonValue.get(JsonSchema.ENVELOPE_PAYLOAD_FIELD_NAME)));
-    }
-
-    private ObjectNode asJsonSchema(Schema schema) {
-        if (schema == null)
-            return null;
-
-        ObjectNode cached = fromCopycatSchemaCache.get(schema);
-        if (cached != null)
-            return cached;
-
-        final ObjectNode jsonSchema;
-        switch (schema.type()) {
-            case BOOLEAN:
-                jsonSchema = JsonSchema.BOOLEAN_SCHEMA.deepCopy();
-                break;
-            case BYTES:
-                jsonSchema = JsonSchema.BYTES_SCHEMA.deepCopy();
-                break;
-            case FLOAT64:
-                jsonSchema = JsonSchema.DOUBLE_SCHEMA.deepCopy();
-                break;
-            case FLOAT32:
-                jsonSchema = JsonSchema.FLOAT_SCHEMA.deepCopy();
-                break;
-            case INT8:
-                jsonSchema = JsonSchema.INT8_SCHEMA.deepCopy();
-                break;
-            case INT16:
-                jsonSchema = JsonSchema.INT16_SCHEMA.deepCopy();
-                break;
-            case INT32:
-                jsonSchema = JsonSchema.INT32_SCHEMA.deepCopy();
-                break;
-            case INT64:
-                jsonSchema = JsonSchema.INT64_SCHEMA.deepCopy();
-                break;
-            case STRING:
-                jsonSchema = JsonSchema.STRING_SCHEMA.deepCopy();
-                break;
-            case ARRAY:
-                jsonSchema = JsonNodeFactory.instance.objectNode().put(JsonSchema.SCHEMA_TYPE_FIELD_NAME, JsonSchema.ARRAY_TYPE_NAME);
-                jsonSchema.set(JsonSchema.ARRAY_ITEMS_FIELD_NAME, asJsonSchema(schema.valueSchema()));
-                break;
-            case MAP:
-                jsonSchema = JsonNodeFactory.instance.objectNode().put(JsonSchema.SCHEMA_TYPE_FIELD_NAME, JsonSchema.MAP_TYPE_NAME);
-                jsonSchema.set(JsonSchema.MAP_KEY_FIELD_NAME, asJsonSchema(schema.keySchema()));
-                jsonSchema.set(JsonSchema.MAP_VALUE_FIELD_NAME, asJsonSchema(schema.valueSchema()));
-                break;
-            case STRUCT:
-                jsonSchema = JsonNodeFactory.instance.objectNode().put(JsonSchema.SCHEMA_TYPE_FIELD_NAME, JsonSchema.STRUCT_TYPE_NAME);
-                ArrayNode fields = JsonNodeFactory.instance.arrayNode();
-                for (Field field : schema.fields()) {
-                    ObjectNode fieldJsonSchema = asJsonSchema(field.schema());
-                    fieldJsonSchema.put(JsonSchema.STRUCT_FIELD_NAME_FIELD_NAME, field.name());
-                    fields.add(fieldJsonSchema);
-                }
-                jsonSchema.set(JsonSchema.STRUCT_FIELDS_FIELD_NAME, fields);
-                break;
-            default:
-                throw new DataException("Couldn't translate unsupported schema type " + schema + ".");
-        }
-
-        jsonSchema.put(JsonSchema.SCHEMA_OPTIONAL_FIELD_NAME, schema.isOptional());
-        if (schema.name() != null)
-            jsonSchema.put(JsonSchema.SCHEMA_NAME_FIELD_NAME, schema.name());
-        if (schema.version() != null)
-            jsonSchema.put(JsonSchema.SCHEMA_VERSION_FIELD_NAME, schema.version());
-        if (schema.doc() != null)
-            jsonSchema.put(JsonSchema.SCHEMA_DOC_FIELD_NAME, schema.doc());
-        if (schema.parameters() != null) {
-            ObjectNode jsonSchemaParams = JsonNodeFactory.instance.objectNode();
-            for (Map.Entry<String, String> prop : schema.parameters().entrySet())
-                jsonSchemaParams.put(prop.getKey(), prop.getValue());
-            jsonSchema.put(JsonSchema.SCHEMA_PARAMETERS_FIELD_NAME, jsonSchemaParams);
-        }
-        if (schema.defaultValue() != null)
-            jsonSchema.set(JsonSchema.SCHEMA_DEFAULT_FIELD_NAME, convertToJson(schema, schema.defaultValue()));
-
-        fromCopycatSchemaCache.put(schema, jsonSchema);
-        return jsonSchema;
-    }
-
-
-    private Schema asCopycatSchema(JsonNode jsonSchema) {
-        if (jsonSchema.isNull())
-            return null;
-
-        Schema cached = toCopycatSchemaCache.get(jsonSchema);
-        if (cached != null)
-            return cached;
-
-        JsonNode schemaTypeNode = jsonSchema.get(JsonSchema.SCHEMA_TYPE_FIELD_NAME);
-        if (schemaTypeNode == null || !schemaTypeNode.isTextual())
-            throw new DataException("Schema must contain 'type' field");
-
-        final SchemaBuilder builder;
-        switch (schemaTypeNode.textValue()) {
-            case JsonSchema.BOOLEAN_TYPE_NAME:
-                builder = SchemaBuilder.bool();
-                break;
-            case JsonSchema.INT8_TYPE_NAME:
-                builder = SchemaBuilder.int8();
-                break;
-            case JsonSchema.INT16_TYPE_NAME:
-                builder = SchemaBuilder.int16();
-                break;
-            case JsonSchema.INT32_TYPE_NAME:
-                builder = SchemaBuilder.int32();
-                break;
-            case JsonSchema.INT64_TYPE_NAME:
-                builder = SchemaBuilder.int64();
-                break;
-            case JsonSchema.FLOAT_TYPE_NAME:
-                builder = SchemaBuilder.float32();
-                break;
-            case JsonSchema.DOUBLE_TYPE_NAME:
-                builder = SchemaBuilder.float64();
-                break;
-            case JsonSchema.BYTES_TYPE_NAME:
-                builder = SchemaBuilder.bytes();
-                break;
-            case JsonSchema.STRING_TYPE_NAME:
-                builder = SchemaBuilder.string();
-                break;
-            case JsonSchema.ARRAY_TYPE_NAME:
-                JsonNode elemSchema = jsonSchema.get(JsonSchema.ARRAY_ITEMS_FIELD_NAME);
-                if (elemSchema == null)
-                    throw new DataException("Array schema did not specify the element type");
-                builder = SchemaBuilder.array(asCopycatSchema(elemSchema));
-                break;
-            case JsonSchema.MAP_TYPE_NAME:
-                JsonNode keySchema = jsonSchema.get(JsonSchema.MAP_KEY_FIELD_NAME);
-                if (keySchema == null)
-                    throw new DataException("Map schema did not specify the key type");
-                JsonNode valueSchema = jsonSchema.get(JsonSchema.MAP_VALUE_FIELD_NAME);
-                if (valueSchema == null)
-                    throw new DataException("Map schema did not specify the value type");
-                builder = SchemaBuilder.map(asCopycatSchema(keySchema), asCopycatSchema(valueSchema));
-                break;
-            case JsonSchema.STRUCT_TYPE_NAME:
-                builder = SchemaBuilder.struct();
-                JsonNode fields = jsonSchema.get(JsonSchema.STRUCT_FIELDS_FIELD_NAME);
-                if (fields == null || !fields.isArray())
-                    throw new DataException("Struct schema's \"fields\" argument is not an array.");
-                for (JsonNode field : fields) {
-                    JsonNode jsonFieldName = field.get(JsonSchema.STRUCT_FIELD_NAME_FIELD_NAME);
-                    if (jsonFieldName == null || !jsonFieldName.isTextual())
-                        throw new DataException("Struct schema's field name not specified properly");
-                    builder.field(jsonFieldName.asText(), asCopycatSchema(field));
-                }
-                break;
-            default:
-                throw new DataException("Unknown schema type: " + schemaTypeNode.textValue());
-        }
-
-
-        JsonNode schemaOptionalNode = jsonSchema.get(JsonSchema.SCHEMA_OPTIONAL_FIELD_NAME);
-        if (schemaOptionalNode != null && schemaOptionalNode.isBoolean() && schemaOptionalNode.booleanValue())
-            builder.optional();
-        else
-            builder.required();
-
-        JsonNode schemaNameNode = jsonSchema.get(JsonSchema.SCHEMA_NAME_FIELD_NAME);
-        if (schemaNameNode != null && schemaNameNode.isTextual())
-            builder.name(schemaNameNode.textValue());
-
-        JsonNode schemaVersionNode = jsonSchema.get(JsonSchema.SCHEMA_VERSION_FIELD_NAME);
-        if (schemaVersionNode != null && schemaVersionNode.isIntegralNumber()) {
-            builder.version(schemaVersionNode.intValue());
-        }
-
-        JsonNode schemaDocNode = jsonSchema.get(JsonSchema.SCHEMA_DOC_FIELD_NAME);
-        if (schemaDocNode != null && schemaDocNode.isTextual())
-            builder.doc(schemaDocNode.textValue());
-
-        JsonNode schemaParamsNode = jsonSchema.get(JsonSchema.SCHEMA_PARAMETERS_FIELD_NAME);
-        if (schemaParamsNode != null && schemaParamsNode.isObject()) {
-            Iterator<Map.Entry<String, JsonNode>> paramsIt = schemaParamsNode.fields();
-            while (paramsIt.hasNext()) {
-                Map.Entry<String, JsonNode> entry = paramsIt.next();
-                JsonNode paramValue = entry.getValue();
-                if (!paramValue.isTextual())
-                    throw new DataException("Schema parameters must have string values.");
-                builder.parameter(entry.getKey(), paramValue.textValue());
-            }
-        }
-
-        JsonNode schemaDefaultNode = jsonSchema.get(JsonSchema.SCHEMA_DEFAULT_FIELD_NAME);
-        if (schemaDefaultNode != null)
-            builder.defaultValue(convertToCopycat(builder, schemaDefaultNode));
-
-        Schema result = builder.build();
-        toCopycatSchemaCache.put(jsonSchema, result);
-        return result;
-    }
-
-
-    /**
-     * Convert this object, in org.apache.kafka.copycat.data format, into a JSON object with an envelope object
-     * containing schema and payload fields.
-     * @param schema the schema for the data
-     * @param value the value
-     * @return JsonNode-encoded version
-     */
-    private JsonNode convertToJsonWithEnvelope(Schema schema, Object value) {
-        return new JsonSchema.Envelope(asJsonSchema(schema), convertToJson(schema, value)).toJsonNode();
-    }
-
-    private JsonNode convertToJsonWithoutEnvelope(Schema schema, Object value) {
-        return convertToJson(schema, value);
-    }
-
-    /**
-     * Convert this object, in the org.apache.kafka.copycat.data format, into a JSON object, returning both the schema
-     * and the converted object.
-     */
-    private static JsonNode convertToJson(Schema schema, Object logicalValue) {
-        if (logicalValue == null) {
-            if (schema == null) // Any schema is valid and we don't have a default, so treat this as an optional schema
-                return null;
-            if (schema.defaultValue() != null)
-                return convertToJson(schema, schema.defaultValue());
-            if (schema.isOptional())
-                return JsonNodeFactory.instance.nullNode();
-            throw new DataException("Conversion error: null value for field that is required and has no default value");
-        }
-
-        Object value = logicalValue;
-        if (schema != null && schema.name() != null) {
-            LogicalTypeConverter logicalConverter = TO_JSON_LOGICAL_CONVERTERS.get(schema.name());
-            if (logicalConverter != null)
-                value = logicalConverter.convert(schema, logicalValue);
-        }
-
-        try {
-            final Schema.Type schemaType;
-            if (schema == null) {
-                schemaType = CopycatSchema.schemaType(value.getClass());
-                if (schemaType == null)
-                    throw new DataException("Java class " + value.getClass() + " does not have corresponding schema type.");
-            } else {
-                schemaType = schema.type();
-            }
-            switch (schemaType) {
-                case INT8:
-                    return JsonNodeFactory.instance.numberNode((Byte) value);
-                case INT16:
-                    return JsonNodeFactory.instance.numberNode((Short) value);
-                case INT32:
-                    return JsonNodeFactory.instance.numberNode((Integer) value);
-                case INT64:
-                    return JsonNodeFactory.instance.numberNode((Long) value);
-                case FLOAT32:
-                    return JsonNodeFactory.instance.numberNode((Float) value);
-                case FLOAT64:
-                    return JsonNodeFactory.instance.numberNode((Double) value);
-                case BOOLEAN:
-                    return JsonNodeFactory.instance.booleanNode((Boolean) value);
-                case STRING:
-                    CharSequence charSeq = (CharSequence) value;
-                    return JsonNodeFactory.instance.textNode(charSeq.toString());
-                case BYTES:
-                    if (value instanceof byte[])
-                        return JsonNodeFactory.instance.binaryNode((byte[]) value);
-                    else if (value instanceof ByteBuffer)
-                        return JsonNodeFactory.instance.binaryNode(((ByteBuffer) value).array());
-                    else
-                        throw new DataException("Invalid type for bytes type: " + value.getClass());
-                case ARRAY: {
-                    Collection collection = (Collection) value;
-                    ArrayNode list = JsonNodeFactory.instance.arrayNode();
-                    for (Object elem : collection) {
-                        Schema valueSchema = schema == null ? null : schema.valueSchema();
-                        JsonNode fieldValue = convertToJson(valueSchema, elem);
-                        list.add(fieldValue);
-                    }
-                    return list;
-                }
-                case MAP: {
-                    Map<?, ?> map = (Map<?, ?>) value;
-                    // If true, using string keys and JSON object; if false, using non-string keys and Array-encoding
-                    boolean objectMode;
-                    if (schema == null) {
-                        objectMode = true;
-                        for (Map.Entry<?, ?> entry : map.entrySet()) {
-                            if (!(entry.getKey() instanceof String)) {
-                                objectMode = false;
-                                break;
-                            }
-                        }
-                    } else {
-                        objectMode = schema.keySchema().type() == Schema.Type.STRING;
-                    }
-                    ObjectNode obj = null;
-                    ArrayNode list = null;
-                    if (objectMode)
-                        obj = JsonNodeFactory.instance.objectNode();
-                    else
-                        list = JsonNodeFactory.instance.arrayNode();
-                    for (Map.Entry<?, ?> entry : map.entrySet()) {
-                        Schema keySchema = schema == null ? null : schema.keySchema();
-                        Schema valueSchema = schema == null ? null : schema.valueSchema();
-                        JsonNode mapKey = convertToJson(keySchema, entry.getKey());
-                        JsonNode mapValue = convertToJson(valueSchema, entry.getValue());
-
-                        if (objectMode)
-                            obj.set(mapKey.asText(), mapValue);
-                        else
-                            list.add(JsonNodeFactory.instance.arrayNode().add(mapKey).add(mapValue));
-                    }
-                    return objectMode ? obj : list;
-                }
-                case STRUCT: {
-                    Struct struct = (Struct) value;
-                    if (struct.schema() != schema)
-                        throw new DataException("Mismatching schema.");
-                    ObjectNode obj = JsonNodeFactory.instance.objectNode();
-                    for (Field field : schema.fields()) {
-                        obj.set(field.name(), convertToJson(field.schema(), struct.get(field)));
-                    }
-                    return obj;
-                }
-            }
-
-            throw new DataException("Couldn't convert " + value + " to JSON.");
-        } catch (ClassCastException e) {
-            throw new DataException("Invalid type for " + schema.type() + ": " + value.getClass());
-        }
-    }
-
-
-    private static Object convertToCopycat(Schema schema, JsonNode jsonValue) {
-        JsonToCopycatTypeConverter typeConverter;
-        final Schema.Type schemaType;
-        if (schema != null) {
-            schemaType = schema.type();
-        } else {
-            switch (jsonValue.getNodeType()) {
-                case NULL:
-                    // Special case. With no schema
-                    return null;
-                case BOOLEAN:
-                    schemaType = Schema.Type.BOOLEAN;
-                    break;
-                case NUMBER:
-                    if (jsonValue.isIntegralNumber())
-                        schemaType = Schema.Type.INT64;
-                    else
-                        schemaType = Schema.Type.FLOAT64;
-                    break;
-                case ARRAY:
-                    schemaType = Schema.Type.ARRAY;
-                    break;
-                case OBJECT:
-                    schemaType = Schema.Type.MAP;
-                    break;
-                case STRING:
-                    schemaType = Schema.Type.STRING;
-                    break;
-
-                case BINARY:
-                case MISSING:
-                case POJO:
-                default:
-                    schemaType = null;
-                    break;
-            }
-        }
-        typeConverter = TO_COPYCAT_CONVERTERS.get(schemaType);
-        if (typeConverter == null)
-            throw new DataException("Unknown schema type: " + schema.type());
-
-        Object converted = typeConverter.convert(schema, jsonValue);
-        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;
-    }
-
-
-    private interface JsonToCopycatTypeConverter {
-        Object convert(Schema schema, JsonNode value);
-    }
-
-    private interface LogicalTypeConverter {
-        Object convert(Schema schema, Object value);
-    }
-}

http://git-wip-us.apache.org/repos/asf/kafka/blob/f2031d40/copycat/json/src/main/java/org/apache/kafka/copycat/json/JsonDeserializer.java
----------------------------------------------------------------------
diff --git a/copycat/json/src/main/java/org/apache/kafka/copycat/json/JsonDeserializer.java b/copycat/json/src/main/java/org/apache/kafka/copycat/json/JsonDeserializer.java
deleted file mode 100644
index 1661754..0000000
--- a/copycat/json/src/main/java/org/apache/kafka/copycat/json/JsonDeserializer.java
+++ /dev/null
@@ -1,62 +0,0 @@
-/**
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements.  See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License.  You may obtain a copy of the License at
- * <p/>
- * http://www.apache.org/licenses/LICENSE-2.0
- * <p/>
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- **/
-package org.apache.kafka.copycat.json;
-
-import com.fasterxml.jackson.databind.JsonNode;
-import com.fasterxml.jackson.databind.ObjectMapper;
-import org.apache.kafka.common.errors.SerializationException;
-import org.apache.kafka.common.serialization.Deserializer;
-
-import java.util.Map;
-
-/**
- * JSON deserializer for Jackson's JsonNode tree model. Using the tree model allows it to work with arbitrarily
- * structured data without having associated Java classes. This deserializer also supports Copycat schemas.
- */
-public class JsonDeserializer implements Deserializer<JsonNode> {
-    private ObjectMapper objectMapper = new ObjectMapper();
-
-    /**
-     * Default constructor needed by Kafka
-     */
-    public JsonDeserializer() {
-    }
-
-    @Override
-    public void configure(Map<String, ?> props, boolean isKey) {
-    }
-
-    @Override
-    public JsonNode deserialize(String topic, byte[] bytes) {
-        if (bytes == null)
-            return null;
-
-        JsonNode data;
-        try {
-            data = objectMapper.readTree(bytes);
-        } catch (Exception e) {
-            throw new SerializationException(e);
-        }
-
-        return data;
-    }
-
-    @Override
-    public void close() {
-
-    }
-}

http://git-wip-us.apache.org/repos/asf/kafka/blob/f2031d40/copycat/json/src/main/java/org/apache/kafka/copycat/json/JsonSchema.java
----------------------------------------------------------------------
diff --git a/copycat/json/src/main/java/org/apache/kafka/copycat/json/JsonSchema.java b/copycat/json/src/main/java/org/apache/kafka/copycat/json/JsonSchema.java
deleted file mode 100644
index 78712f3..0000000
--- a/copycat/json/src/main/java/org/apache/kafka/copycat/json/JsonSchema.java
+++ /dev/null
@@ -1,82 +0,0 @@
-/**
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements.  See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License.  You may obtain a copy of the License at
- *
- *    http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- **/
-
-package org.apache.kafka.copycat.json;
-
-import com.fasterxml.jackson.databind.JsonNode;
-import com.fasterxml.jackson.databind.node.JsonNodeFactory;
-import com.fasterxml.jackson.databind.node.ObjectNode;
-
-public class JsonSchema {
-
-    static final String ENVELOPE_SCHEMA_FIELD_NAME = "schema";
-    static final String ENVELOPE_PAYLOAD_FIELD_NAME = "payload";
-    static final String SCHEMA_TYPE_FIELD_NAME = "type";
-    static final String SCHEMA_OPTIONAL_FIELD_NAME = "optional";
-    static final String SCHEMA_NAME_FIELD_NAME = "name";
-    static final String SCHEMA_VERSION_FIELD_NAME = "version";
-    static final String SCHEMA_DOC_FIELD_NAME = "doc";
-    static final String SCHEMA_PARAMETERS_FIELD_NAME = "parameters";
-    static final String SCHEMA_DEFAULT_FIELD_NAME = "default";
-    static final String ARRAY_ITEMS_FIELD_NAME = "items";
-    static final String MAP_KEY_FIELD_NAME = "keys";
-    static final String MAP_VALUE_FIELD_NAME = "values";
-    static final String STRUCT_FIELDS_FIELD_NAME = "fields";
-    static final String STRUCT_FIELD_NAME_FIELD_NAME = "field";
-    static final String BOOLEAN_TYPE_NAME = "boolean";
-    static final ObjectNode BOOLEAN_SCHEMA = JsonNodeFactory.instance.objectNode().put(SCHEMA_TYPE_FIELD_NAME, BOOLEAN_TYPE_NAME);
-    static final String INT8_TYPE_NAME = "int8";
-    static final ObjectNode INT8_SCHEMA = JsonNodeFactory.instance.objectNode().put(SCHEMA_TYPE_FIELD_NAME, INT8_TYPE_NAME);
-    static final String INT16_TYPE_NAME = "int16";
-    static final ObjectNode INT16_SCHEMA = JsonNodeFactory.instance.objectNode().put(SCHEMA_TYPE_FIELD_NAME, INT16_TYPE_NAME);
-    static final String INT32_TYPE_NAME = "int32";
-    static final ObjectNode INT32_SCHEMA = JsonNodeFactory.instance.objectNode().put(SCHEMA_TYPE_FIELD_NAME, INT32_TYPE_NAME);
-    static final String INT64_TYPE_NAME = "int64";
-    static final ObjectNode INT64_SCHEMA = JsonNodeFactory.instance.objectNode().put(SCHEMA_TYPE_FIELD_NAME, INT64_TYPE_NAME);
-    static final String FLOAT_TYPE_NAME = "float";
-    static final ObjectNode FLOAT_SCHEMA = JsonNodeFactory.instance.objectNode().put(SCHEMA_TYPE_FIELD_NAME, FLOAT_TYPE_NAME);
-    static final String DOUBLE_TYPE_NAME = "double";
-    static final ObjectNode DOUBLE_SCHEMA = JsonNodeFactory.instance.objectNode().put(SCHEMA_TYPE_FIELD_NAME, DOUBLE_TYPE_NAME);
-    static final String BYTES_TYPE_NAME = "bytes";
-    static final ObjectNode BYTES_SCHEMA = JsonNodeFactory.instance.objectNode().put(SCHEMA_TYPE_FIELD_NAME, BYTES_TYPE_NAME);
-    static final String STRING_TYPE_NAME = "string";
-    static final ObjectNode STRING_SCHEMA = JsonNodeFactory.instance.objectNode().put(SCHEMA_TYPE_FIELD_NAME, STRING_TYPE_NAME);
-    static final String ARRAY_TYPE_NAME = "array";
-    static final String MAP_TYPE_NAME = "map";
-    static final String STRUCT_TYPE_NAME = "struct";
-
-    public static ObjectNode envelope(JsonNode schema, JsonNode payload) {
-        ObjectNode result = JsonNodeFactory.instance.objectNode();
-        result.set(ENVELOPE_SCHEMA_FIELD_NAME, schema);
-        result.set(ENVELOPE_PAYLOAD_FIELD_NAME, payload);
-        return result;
-    }
-
-    static class Envelope {
-        public JsonNode schema;
-        public JsonNode payload;
-
-        public Envelope(JsonNode schema, JsonNode payload) {
-            this.schema = schema;
-            this.payload = payload;
-        }
-
-        public ObjectNode toJsonNode() {
-            return envelope(schema, payload);
-        }
-    }
-}

http://git-wip-us.apache.org/repos/asf/kafka/blob/f2031d40/copycat/json/src/main/java/org/apache/kafka/copycat/json/JsonSerializer.java
----------------------------------------------------------------------
diff --git a/copycat/json/src/main/java/org/apache/kafka/copycat/json/JsonSerializer.java b/copycat/json/src/main/java/org/apache/kafka/copycat/json/JsonSerializer.java
deleted file mode 100644
index 129d14b..0000000
--- a/copycat/json/src/main/java/org/apache/kafka/copycat/json/JsonSerializer.java
+++ /dev/null
@@ -1,60 +0,0 @@
-/**
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements.  See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License.  You may obtain a copy of the License at
- * <p/>
- * http://www.apache.org/licenses/LICENSE-2.0
- * <p/>
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- **/
-package org.apache.kafka.copycat.json;
-
-import com.fasterxml.jackson.databind.JsonNode;
-import com.fasterxml.jackson.databind.ObjectMapper;
-import org.apache.kafka.common.errors.SerializationException;
-import org.apache.kafka.common.serialization.Serializer;
-
-import java.util.Map;
-
-/**
- * Serialize Jackson JsonNode tree model objects to UTF-8 JSON. Using the tree model allows handling arbitrarily
- * structured data without corresponding Java classes. This serializer also supports Copycat schemas.
- */
-public class JsonSerializer implements Serializer<JsonNode> {
-    private final ObjectMapper objectMapper = new ObjectMapper();
-
-    /**
-     * Default constructor needed by Kafka
-     */
-    public JsonSerializer() {
-
-    }
-
-    @Override
-    public void configure(Map<String, ?> config, boolean isKey) {
-    }
-
-    @Override
-    public byte[] serialize(String topic, JsonNode data) {
-        if (data == null)
-            return null;
-
-        try {
-            return objectMapper.writeValueAsBytes(data);
-        } catch (Exception e) {
-            throw new SerializationException("Error serializing JSON message", e);
-        }
-    }
-
-    @Override
-    public void close() {
-    }
-
-}

http://git-wip-us.apache.org/repos/asf/kafka/blob/f2031d40/copycat/json/src/test/java/org/apache/kafka/copycat/json/JsonConverterTest.java
----------------------------------------------------------------------
diff --git a/copycat/json/src/test/java/org/apache/kafka/copycat/json/JsonConverterTest.java b/copycat/json/src/test/java/org/apache/kafka/copycat/json/JsonConverterTest.java
deleted file mode 100644
index 6b40046..0000000
--- a/copycat/json/src/test/java/org/apache/kafka/copycat/json/JsonConverterTest.java
+++ /dev/null
@@ -1,644 +0,0 @@
-/**
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements.  See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License.  You may obtain a copy of the License at
- *
- *    http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- **/
-
-package org.apache.kafka.copycat.json;
-
-import com.fasterxml.jackson.databind.JsonNode;
-import com.fasterxml.jackson.databind.ObjectMapper;
-import com.fasterxml.jackson.databind.node.ArrayNode;
-import com.fasterxml.jackson.databind.node.JsonNodeFactory;
-import org.apache.kafka.copycat.data.Date;
-import org.apache.kafka.copycat.data.Decimal;
-import com.fasterxml.jackson.databind.node.ObjectNode;
-import org.apache.kafka.common.cache.Cache;
-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.junit.Before;
-import org.junit.Test;
-import org.powermock.reflect.Whitebox;
-
-import java.io.IOException;
-import java.io.UnsupportedEncodingException;
-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.HashMap;
-import java.util.HashSet;
-import java.util.Map;
-import java.util.Set;
-import java.util.TimeZone;
-
-import static org.junit.Assert.assertArrayEquals;
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertNotNull;
-import static org.junit.Assert.assertTrue;
-import static org.junit.Assert.fail;
-
-public class JsonConverterTest {
-    private static final String TOPIC = "topic";
-
-    ObjectMapper objectMapper = new ObjectMapper();
-    JsonConverter converter = new JsonConverter();
-
-    @Before
-    public void setUp() {
-        converter.configure(Collections.EMPTY_MAP, false);
-    }
-
-    // Schema metadata
-
-    @Test
-    public void testCopycatSchemaMetadataTranslation() {
-        // this validates the non-type fields are translated and handled properly
-        assertEquals(new SchemaAndValue(Schema.BOOLEAN_SCHEMA, true), converter.toCopycatData(TOPIC, "{ \"schema\": { \"type\": \"boolean\" }, \"payload\": true }".getBytes()));
-        assertEquals(new SchemaAndValue(Schema.OPTIONAL_BOOLEAN_SCHEMA, null), converter.toCopycatData(TOPIC, "{ \"schema\": { \"type\": \"boolean\", \"optional\": true }, \"payload\": null }".getBytes()));
-        assertEquals(new SchemaAndValue(SchemaBuilder.bool().defaultValue(true).build(), true),
-                converter.toCopycatData(TOPIC, "{ \"schema\": { \"type\": \"boolean\", \"default\": true }, \"payload\": null }".getBytes()));
-        assertEquals(new SchemaAndValue(SchemaBuilder.bool().required().name("bool").version(2).doc("the documentation").parameter("foo", "bar").build(), true),
-                converter.toCopycatData(TOPIC, "{ \"schema\": { \"type\": \"boolean\", \"optional\": false, \"name\": \"bool\", \"version\": 2, \"doc\": \"the documentation\", \"parameters\": { \"foo\": \"bar\" }}, \"payload\": true }".getBytes()));
-    }
-
-    // Schema types
-
-    @Test
-    public void booleanToCopycat() {
-        assertEquals(new SchemaAndValue(Schema.BOOLEAN_SCHEMA, true), converter.toCopycatData(TOPIC, "{ \"schema\": { \"type\": \"boolean\" }, \"payload\": true }".getBytes()));
-        assertEquals(new SchemaAndValue(Schema.BOOLEAN_SCHEMA, false), converter.toCopycatData(TOPIC, "{ \"schema\": { \"type\": \"boolean\" }, \"payload\": false }".getBytes()));
-    }
-
-    @Test
-    public void byteToCopycat() {
-        assertEquals(new SchemaAndValue(Schema.INT8_SCHEMA, (byte) 12), converter.toCopycatData(TOPIC, "{ \"schema\": { \"type\": \"int8\" }, \"payload\": 12 }".getBytes()));
-    }
-
-    @Test
-    public void shortToCopycat() {
-        assertEquals(new SchemaAndValue(Schema.INT16_SCHEMA, (short) 12), converter.toCopycatData(TOPIC, "{ \"schema\": { \"type\": \"int16\" }, \"payload\": 12 }".getBytes()));
-    }
-
-    @Test
-    public void intToCopycat() {
-        assertEquals(new SchemaAndValue(Schema.INT32_SCHEMA, 12), converter.toCopycatData(TOPIC, "{ \"schema\": { \"type\": \"int32\" }, \"payload\": 12 }".getBytes()));
-    }
-
-    @Test
-    public void longToCopycat() {
-        assertEquals(new SchemaAndValue(Schema.INT64_SCHEMA, 12L), converter.toCopycatData(TOPIC, "{ \"schema\": { \"type\": \"int64\" }, \"payload\": 12 }".getBytes()));
-        assertEquals(new SchemaAndValue(Schema.INT64_SCHEMA, 4398046511104L), converter.toCopycatData(TOPIC, "{ \"schema\": { \"type\": \"int64\" }, \"payload\": 4398046511104 }".getBytes()));
-    }
-
-    @Test
-    public void floatToCopycat() {
-        assertEquals(new SchemaAndValue(Schema.FLOAT32_SCHEMA, 12.34f), converter.toCopycatData(TOPIC, "{ \"schema\": { \"type\": \"float\" }, \"payload\": 12.34 }".getBytes()));
-    }
-
-    @Test
-    public void doubleToCopycat() {
-        assertEquals(new SchemaAndValue(Schema.FLOAT64_SCHEMA, 12.34), converter.toCopycatData(TOPIC, "{ \"schema\": { \"type\": \"double\" }, \"payload\": 12.34 }".getBytes()));
-    }
-
-
-    @Test
-    public void bytesToCopycat() throws UnsupportedEncodingException {
-        ByteBuffer reference = ByteBuffer.wrap("test-string".getBytes("UTF-8"));
-        String msg = "{ \"schema\": { \"type\": \"bytes\" }, \"payload\": \"dGVzdC1zdHJpbmc=\" }";
-        SchemaAndValue schemaAndValue = converter.toCopycatData(TOPIC, msg.getBytes());
-        ByteBuffer converted = ByteBuffer.wrap((byte[]) schemaAndValue.value());
-        assertEquals(reference, converted);
-    }
-
-    @Test
-    public void stringToCopycat() {
-        assertEquals(new SchemaAndValue(Schema.STRING_SCHEMA, "foo-bar-baz"), converter.toCopycatData(TOPIC, "{ \"schema\": { \"type\": \"string\" }, \"payload\": \"foo-bar-baz\" }".getBytes()));
-    }
-
-    @Test
-    public void arrayToCopycat() {
-        byte[] arrayJson = "{ \"schema\": { \"type\": \"array\", \"items\": { \"type\" : \"int32\" } }, \"payload\": [1, 2, 3] }".getBytes();
-        assertEquals(new SchemaAndValue(SchemaBuilder.array(Schema.INT32_SCHEMA).build(), Arrays.asList(1, 2, 3)), converter.toCopycatData(TOPIC, arrayJson));
-    }
-
-    @Test
-    public void mapToCopycatStringKeys() {
-        byte[] mapJson = "{ \"schema\": { \"type\": \"map\", \"keys\": { \"type\" : \"string\" }, \"values\": { \"type\" : \"int32\" } }, \"payload\": { \"key1\": 12, \"key2\": 15} }".getBytes();
-        Map<String, Integer> expected = new HashMap<>();
-        expected.put("key1", 12);
-        expected.put("key2", 15);
-        assertEquals(new SchemaAndValue(SchemaBuilder.map(Schema.STRING_SCHEMA, Schema.INT32_SCHEMA).build(), expected), converter.toCopycatData(TOPIC, mapJson));
-    }
-
-    @Test
-    public void mapToCopycatNonStringKeys() {
-        byte[] mapJson = "{ \"schema\": { \"type\": \"map\", \"keys\": { \"type\" : \"int32\" }, \"values\": { \"type\" : \"int32\" } }, \"payload\": [ [1, 12], [2, 15] ] }".getBytes();
-        Map<Integer, Integer> expected = new HashMap<>();
-        expected.put(1, 12);
-        expected.put(2, 15);
-        assertEquals(new SchemaAndValue(SchemaBuilder.map(Schema.INT32_SCHEMA, Schema.INT32_SCHEMA).build(), expected), converter.toCopycatData(TOPIC, mapJson));
-    }
-
-    @Test
-    public void structToCopycat() {
-        byte[] structJson = "{ \"schema\": { \"type\": \"struct\", \"fields\": [{ \"field\": \"field1\", \"type\": \"boolean\" }, { \"field\": \"field2\", \"type\": \"string\" }] }, \"payload\": { \"field1\": true, \"field2\": \"string\" } }".getBytes();
-        Schema expectedSchema = SchemaBuilder.struct().field("field1", Schema.BOOLEAN_SCHEMA).field("field2", Schema.STRING_SCHEMA).build();
-        Struct expected = new Struct(expectedSchema).put("field1", true).put("field2", "string");
-        SchemaAndValue converted = converter.toCopycatData(TOPIC, structJson);
-        assertEquals(new SchemaAndValue(expectedSchema, expected), converted);
-    }
-
-    @Test(expected = DataException.class)
-    public void nullToCopycat() {
-        // When schemas are enabled, trying to decode a null should be an error -- we should *always* have the envelope
-        assertEquals(SchemaAndValue.NULL, converter.toCopycatData(TOPIC, null));
-    }
-
-    @Test
-    public void nullSchemaPrimitiveToCopycat() {
-        SchemaAndValue converted = converter.toCopycatData(TOPIC, "{ \"schema\": null, \"payload\": null }".getBytes());
-        assertEquals(SchemaAndValue.NULL, converted);
-
-        converted = converter.toCopycatData(TOPIC, "{ \"schema\": null, \"payload\": true }".getBytes());
-        assertEquals(new SchemaAndValue(null, true), converted);
-
-        // Integers: Copycat has more data types, and JSON unfortunately mixes all number types. We try to preserve
-        // info as best we can, so we always use the largest integer and floating point numbers we can and have Jackson
-        // determine if it's an integer or not
-        converted = converter.toCopycatData(TOPIC, "{ \"schema\": null, \"payload\": 12 }".getBytes());
-        assertEquals(new SchemaAndValue(null, 12L), converted);
-
-        converted = converter.toCopycatData(TOPIC, "{ \"schema\": null, \"payload\": 12.24 }".getBytes());
-        assertEquals(new SchemaAndValue(null, 12.24), converted);
-
-        converted = converter.toCopycatData(TOPIC, "{ \"schema\": null, \"payload\": \"a string\" }".getBytes());
-        assertEquals(new SchemaAndValue(null, "a string"), converted);
-
-        converted = converter.toCopycatData(TOPIC, "{ \"schema\": null, \"payload\": [1, \"2\", 3] }".getBytes());
-        assertEquals(new SchemaAndValue(null, Arrays.asList(1L, "2", 3L)), converted);
-
-        converted = converter.toCopycatData(TOPIC, "{ \"schema\": null, \"payload\": { \"field1\": 1, \"field2\": 2} }".getBytes());
-        Map<String, Long> obj = new HashMap<>();
-        obj.put("field1", 1L);
-        obj.put("field2", 2L);
-        assertEquals(new SchemaAndValue(null, obj), converted);
-    }
-
-    @Test
-    public void decimalToCopycat() {
-        Schema schema = Decimal.schema(2);
-        BigDecimal reference = new BigDecimal(new BigInteger("156"), 2);
-        // Payload is base64 encoded byte[]{0, -100}, which is the two's complement encoding of 156.
-        String msg = "{ \"schema\": { \"type\": \"bytes\", \"name\": \"org.apache.kafka.copycat.data.Decimal\", \"version\": 1, \"parameters\": { \"scale\": \"2\" } }, \"payload\": \"AJw=\" }";
-        SchemaAndValue schemaAndValue = converter.toCopycatData(TOPIC, msg.getBytes());
-        BigDecimal converted = (BigDecimal) schemaAndValue.value();
-        assertEquals(schema, schemaAndValue.schema());
-        assertEquals(reference, converted);
-    }
-
-    @Test
-    public void dateToCopycat() {
-        Schema schema = Date.SCHEMA;
-        GregorianCalendar calendar = new GregorianCalendar(1970, Calendar.JANUARY, 1, 0, 0, 0);
-        calendar.setTimeZone(TimeZone.getTimeZone("UTC"));
-        calendar.add(Calendar.DATE, 10000);
-        java.util.Date reference = calendar.getTime();
-        String msg = "{ \"schema\": { \"type\": \"int32\", \"name\": \"org.apache.kafka.copycat.data.Date\", \"version\": 1 }, \"payload\": 10000 }";
-        SchemaAndValue schemaAndValue = converter.toCopycatData(TOPIC, msg.getBytes());
-        java.util.Date converted = (java.util.Date) schemaAndValue.value();
-        assertEquals(schema, schemaAndValue.schema());
-        assertEquals(reference, converted);
-    }
-
-    @Test
-    public void timeToCopycat() {
-        Schema schema = Time.SCHEMA;
-        GregorianCalendar calendar = new GregorianCalendar(1970, Calendar.JANUARY, 1, 0, 0, 0);
-        calendar.setTimeZone(TimeZone.getTimeZone("UTC"));
-        calendar.add(Calendar.MILLISECOND, 14400000);
-        java.util.Date reference = calendar.getTime();
-        String msg = "{ \"schema\": { \"type\": \"int32\", \"name\": \"org.apache.kafka.copycat.data.Time\", \"version\": 1 }, \"payload\": 14400000 }";
-        SchemaAndValue schemaAndValue = converter.toCopycatData(TOPIC, msg.getBytes());
-        java.util.Date converted = (java.util.Date) schemaAndValue.value();
-        assertEquals(schema, schemaAndValue.schema());
-        assertEquals(reference, converted);
-    }
-
-    @Test
-    public void timestampToCopycat() {
-        Schema schema = Timestamp.SCHEMA;
-        GregorianCalendar calendar = new GregorianCalendar(1970, Calendar.JANUARY, 1, 0, 0, 0);
-        calendar.setTimeZone(TimeZone.getTimeZone("UTC"));
-        calendar.add(Calendar.MILLISECOND, 2000000000);
-        calendar.add(Calendar.MILLISECOND, 2000000000);
-        java.util.Date reference = calendar.getTime();
-        String msg = "{ \"schema\": { \"type\": \"int64\", \"name\": \"org.apache.kafka.copycat.data.Timestamp\", \"version\": 1 }, \"payload\": 4000000000 }";
-        SchemaAndValue schemaAndValue = converter.toCopycatData(TOPIC, msg.getBytes());
-        java.util.Date converted = (java.util.Date) schemaAndValue.value();
-        assertEquals(schema, schemaAndValue.schema());
-        assertEquals(reference, converted);
-    }
-
-    // Schema metadata
-
-    @Test
-    public void testJsonSchemaMetadataTranslation() {
-        JsonNode converted = parse(converter.fromCopycatData(TOPIC, Schema.BOOLEAN_SCHEMA, true));
-        validateEnvelope(converted);
-        assertEquals(parse("{ \"type\": \"boolean\", \"optional\": false }"), converted.get(JsonSchema.ENVELOPE_SCHEMA_FIELD_NAME));
-        assertEquals(true, converted.get(JsonSchema.ENVELOPE_PAYLOAD_FIELD_NAME).booleanValue());
-
-        converted = parse(converter.fromCopycatData(TOPIC, Schema.OPTIONAL_BOOLEAN_SCHEMA, null));
-        validateEnvelope(converted);
-        assertEquals(parse("{ \"type\": \"boolean\", \"optional\": true }"), converted.get(JsonSchema.ENVELOPE_SCHEMA_FIELD_NAME));
-        assertTrue(converted.get(JsonSchema.ENVELOPE_PAYLOAD_FIELD_NAME).isNull());
-
-        converted = parse(converter.fromCopycatData(TOPIC, SchemaBuilder.bool().defaultValue(true).build(), true));
-        validateEnvelope(converted);
-        assertEquals(parse("{ \"type\": \"boolean\", \"optional\": false, \"default\": true }"), converted.get(JsonSchema.ENVELOPE_SCHEMA_FIELD_NAME));
-        assertEquals(true, converted.get(JsonSchema.ENVELOPE_PAYLOAD_FIELD_NAME).booleanValue());
-
-        converted = parse(converter.fromCopycatData(TOPIC, SchemaBuilder.bool().required().name("bool").version(3).doc("the documentation").parameter("foo", "bar").build(), true));
-        validateEnvelope(converted);
-        assertEquals(parse("{ \"type\": \"boolean\", \"optional\": false, \"name\": \"bool\", \"version\": 3, \"doc\": \"the documentation\", \"parameters\": { \"foo\": \"bar\" }}"),
-                converted.get(JsonSchema.ENVELOPE_SCHEMA_FIELD_NAME));
-        assertEquals(true, converted.get(JsonSchema.ENVELOPE_PAYLOAD_FIELD_NAME).booleanValue());
-    }
-
-
-    @Test
-    public void testCacheSchemaToCopycatConversion() {
-        Cache<JsonNode, Schema> cache = Whitebox.getInternalState(converter, "toCopycatSchemaCache");
-        assertEquals(0, cache.size());
-
-        converter.toCopycatData(TOPIC, "{ \"schema\": { \"type\": \"boolean\" }, \"payload\": true }".getBytes());
-        assertEquals(1, cache.size());
-
-        converter.toCopycatData(TOPIC, "{ \"schema\": { \"type\": \"boolean\" }, \"payload\": true }".getBytes());
-        assertEquals(1, cache.size());
-
-        // Different schema should also get cached
-        converter.toCopycatData(TOPIC, "{ \"schema\": { \"type\": \"boolean\", \"optional\": true }, \"payload\": true }".getBytes());
-        assertEquals(2, cache.size());
-
-        // Even equivalent, but different JSON encoding of schema, should get different cache entry
-        converter.toCopycatData(TOPIC, "{ \"schema\": { \"type\": \"boolean\", \"optional\": false }, \"payload\": true }".getBytes());
-        assertEquals(3, cache.size());
-    }
-
-    // Schema types
-
-    @Test
-    public void booleanToJson() {
-        JsonNode converted = parse(converter.fromCopycatData(TOPIC, Schema.BOOLEAN_SCHEMA, true));
-        validateEnvelope(converted);
-        assertEquals(parse("{ \"type\": \"boolean\", \"optional\": false }"), converted.get(JsonSchema.ENVELOPE_SCHEMA_FIELD_NAME));
-        assertEquals(true, converted.get(JsonSchema.ENVELOPE_PAYLOAD_FIELD_NAME).booleanValue());
-    }
-
-    @Test
-    public void byteToJson() {
-        JsonNode converted = parse(converter.fromCopycatData(TOPIC, Schema.INT8_SCHEMA, (byte) 12));
-        validateEnvelope(converted);
-        assertEquals(parse("{ \"type\": \"int8\", \"optional\": false }"), converted.get(JsonSchema.ENVELOPE_SCHEMA_FIELD_NAME));
-        assertEquals(12, converted.get(JsonSchema.ENVELOPE_PAYLOAD_FIELD_NAME).intValue());
-    }
-
-    @Test
-    public void shortToJson() {
-        JsonNode converted = parse(converter.fromCopycatData(TOPIC, Schema.INT16_SCHEMA, (short) 12));
-        validateEnvelope(converted);
-        assertEquals(parse("{ \"type\": \"int16\", \"optional\": false }"), converted.get(JsonSchema.ENVELOPE_SCHEMA_FIELD_NAME));
-        assertEquals(12, converted.get(JsonSchema.ENVELOPE_PAYLOAD_FIELD_NAME).intValue());
-    }
-
-    @Test
-    public void intToJson() {
-        JsonNode converted = parse(converter.fromCopycatData(TOPIC, Schema.INT32_SCHEMA, 12));
-        validateEnvelope(converted);
-        assertEquals(parse("{ \"type\": \"int32\", \"optional\": false }"), converted.get(JsonSchema.ENVELOPE_SCHEMA_FIELD_NAME));
-        assertEquals(12, converted.get(JsonSchema.ENVELOPE_PAYLOAD_FIELD_NAME).intValue());
-    }
-
-    @Test
-    public void longToJson() {
-        JsonNode converted = parse(converter.fromCopycatData(TOPIC, Schema.INT64_SCHEMA, 4398046511104L));
-        validateEnvelope(converted);
-        assertEquals(parse("{ \"type\": \"int64\", \"optional\": false }"), converted.get(JsonSchema.ENVELOPE_SCHEMA_FIELD_NAME));
-        assertEquals(4398046511104L, converted.get(JsonSchema.ENVELOPE_PAYLOAD_FIELD_NAME).longValue());
-    }
-
-    @Test
-    public void floatToJson() {
-        JsonNode converted = parse(converter.fromCopycatData(TOPIC, Schema.FLOAT32_SCHEMA, 12.34f));
-        validateEnvelope(converted);
-        assertEquals(parse("{ \"type\": \"float\", \"optional\": false }"), converted.get(JsonSchema.ENVELOPE_SCHEMA_FIELD_NAME));
-        assertEquals(12.34f, converted.get(JsonSchema.ENVELOPE_PAYLOAD_FIELD_NAME).floatValue(), 0.001);
-    }
-
-    @Test
-    public void doubleToJson() {
-        JsonNode converted = parse(converter.fromCopycatData(TOPIC, Schema.FLOAT64_SCHEMA, 12.34));
-        validateEnvelope(converted);
-        assertEquals(parse("{ \"type\": \"double\", \"optional\": false }"), converted.get(JsonSchema.ENVELOPE_SCHEMA_FIELD_NAME));
-        assertEquals(12.34, converted.get(JsonSchema.ENVELOPE_PAYLOAD_FIELD_NAME).doubleValue(), 0.001);
-    }
-
-    @Test
-    public void bytesToJson() throws IOException {
-        JsonNode converted = parse(converter.fromCopycatData(TOPIC, Schema.BYTES_SCHEMA, "test-string".getBytes()));
-        validateEnvelope(converted);
-        assertEquals(parse("{ \"type\": \"bytes\", \"optional\": false }"), converted.get(JsonSchema.ENVELOPE_SCHEMA_FIELD_NAME));
-        assertEquals(ByteBuffer.wrap("test-string".getBytes()),
-                ByteBuffer.wrap(converted.get(JsonSchema.ENVELOPE_PAYLOAD_FIELD_NAME).binaryValue()));
-    }
-
-    @Test
-    public void stringToJson() {
-        JsonNode converted = parse(converter.fromCopycatData(TOPIC, Schema.STRING_SCHEMA, "test-string"));
-        validateEnvelope(converted);
-        assertEquals(parse("{ \"type\": \"string\", \"optional\": false }"), converted.get(JsonSchema.ENVELOPE_SCHEMA_FIELD_NAME));
-        assertEquals("test-string", converted.get(JsonSchema.ENVELOPE_PAYLOAD_FIELD_NAME).textValue());
-    }
-
-    @Test
-    public void arrayToJson() {
-        Schema int32Array = SchemaBuilder.array(Schema.INT32_SCHEMA).build();
-        JsonNode converted = parse(converter.fromCopycatData(TOPIC, int32Array, Arrays.asList(1, 2, 3)));
-        validateEnvelope(converted);
-        assertEquals(parse("{ \"type\": \"array\", \"items\": { \"type\": \"int32\", \"optional\": false }, \"optional\": false }"),
-                converted.get(JsonSchema.ENVELOPE_SCHEMA_FIELD_NAME));
-        assertEquals(JsonNodeFactory.instance.arrayNode().add(1).add(2).add(3),
-                converted.get(JsonSchema.ENVELOPE_PAYLOAD_FIELD_NAME));
-    }
-
-    @Test
-    public void mapToJsonStringKeys() {
-        Schema stringIntMap = SchemaBuilder.map(Schema.STRING_SCHEMA, Schema.INT32_SCHEMA).build();
-        Map<String, Integer> input = new HashMap<>();
-        input.put("key1", 12);
-        input.put("key2", 15);
-        JsonNode converted = parse(converter.fromCopycatData(TOPIC, stringIntMap, input));
-        validateEnvelope(converted);
-        assertEquals(parse("{ \"type\": \"map\", \"keys\": { \"type\" : \"string\", \"optional\": false }, \"values\": { \"type\" : \"int32\", \"optional\": false }, \"optional\": false }"),
-                converted.get(JsonSchema.ENVELOPE_SCHEMA_FIELD_NAME));
-        assertEquals(JsonNodeFactory.instance.objectNode().put("key1", 12).put("key2", 15),
-                converted.get(JsonSchema.ENVELOPE_PAYLOAD_FIELD_NAME));
-    }
-
-    @Test
-    public void mapToJsonNonStringKeys() {
-        Schema intIntMap = SchemaBuilder.map(Schema.INT32_SCHEMA, Schema.INT32_SCHEMA).build();
-        Map<Integer, Integer> input = new HashMap<>();
-        input.put(1, 12);
-        input.put(2, 15);
-        JsonNode converted = parse(converter.fromCopycatData(TOPIC, intIntMap, input));
-        validateEnvelope(converted);
-        assertEquals(parse("{ \"type\": \"map\", \"keys\": { \"type\" : \"int32\", \"optional\": false }, \"values\": { \"type\" : \"int32\", \"optional\": false }, \"optional\": false }"),
-                converted.get(JsonSchema.ENVELOPE_SCHEMA_FIELD_NAME));
-
-        assertTrue(converted.get(JsonSchema.ENVELOPE_PAYLOAD_FIELD_NAME).isArray());
-        ArrayNode payload = (ArrayNode) converted.get(JsonSchema.ENVELOPE_PAYLOAD_FIELD_NAME);
-        assertEquals(2, payload.size());
-        Set<JsonNode> payloadEntries = new HashSet<>();
-        for (JsonNode elem : payload)
-            payloadEntries.add(elem);
-        assertEquals(new HashSet<>(Arrays.asList(JsonNodeFactory.instance.arrayNode().add(1).add(12),
-                        JsonNodeFactory.instance.arrayNode().add(2).add(15))),
-                payloadEntries
-        );
-    }
-
-    @Test
-    public void structToJson() {
-        Schema schema = SchemaBuilder.struct().field("field1", Schema.BOOLEAN_SCHEMA).field("field2", Schema.STRING_SCHEMA).build();
-        Struct input = new Struct(schema).put("field1", true).put("field2", "string");
-        JsonNode converted = parse(converter.fromCopycatData(TOPIC, schema, input));
-        validateEnvelope(converted);
-        assertEquals(parse("{ \"type\": \"struct\", \"optional\": false, \"fields\": [{ \"field\": \"field1\", \"type\": \"boolean\", \"optional\": false }, { \"field\": \"field2\", \"type\": \"string\", \"optional\": false }] }"),
-                converted.get(JsonSchema.ENVELOPE_SCHEMA_FIELD_NAME));
-        assertEquals(JsonNodeFactory.instance.objectNode()
-                        .put("field1", true)
-                        .put("field2", "string"),
-                converted.get(JsonSchema.ENVELOPE_PAYLOAD_FIELD_NAME));
-    }
-
-
-    @Test
-    public void decimalToJson() throws IOException {
-        JsonNode converted = parse(converter.fromCopycatData(TOPIC, Decimal.schema(2), new BigDecimal(new BigInteger("156"), 2)));
-        validateEnvelope(converted);
-        assertEquals(parse("{ \"type\": \"bytes\", \"optional\": false, \"name\": \"org.apache.kafka.copycat.data.Decimal\", \"version\": 1, \"parameters\": { \"scale\": \"2\" } }"),
-                converted.get(JsonSchema.ENVELOPE_SCHEMA_FIELD_NAME));
-        assertArrayEquals(new byte[]{0, -100}, converted.get(JsonSchema.ENVELOPE_PAYLOAD_FIELD_NAME).binaryValue());
-    }
-
-    @Test
-    public void dateToJson() throws IOException {
-        GregorianCalendar calendar = new GregorianCalendar(1970, Calendar.JANUARY, 1, 0, 0, 0);
-        calendar.setTimeZone(TimeZone.getTimeZone("UTC"));
-        calendar.add(Calendar.DATE, 10000);
-        java.util.Date date = calendar.getTime();
-
-        JsonNode converted = parse(converter.fromCopycatData(TOPIC, Date.SCHEMA, date));
-        validateEnvelope(converted);
-        assertEquals(parse("{ \"type\": \"int32\", \"optional\": false, \"name\": \"org.apache.kafka.copycat.data.Date\", \"version\": 1 }"),
-                converted.get(JsonSchema.ENVELOPE_SCHEMA_FIELD_NAME));
-        JsonNode payload = converted.get(JsonSchema.ENVELOPE_PAYLOAD_FIELD_NAME);
-        assertTrue(payload.isInt());
-        assertEquals(10000, payload.intValue());
-    }
-
-    @Test
-    public void timeToJson() throws IOException {
-        GregorianCalendar calendar = new GregorianCalendar(1970, Calendar.JANUARY, 1, 0, 0, 0);
-        calendar.setTimeZone(TimeZone.getTimeZone("UTC"));
-        calendar.add(Calendar.MILLISECOND, 14400000);
-        java.util.Date date = calendar.getTime();
-
-        JsonNode converted = parse(converter.fromCopycatData(TOPIC, Time.SCHEMA, date));
-        validateEnvelope(converted);
-        assertEquals(parse("{ \"type\": \"int32\", \"optional\": false, \"name\": \"org.apache.kafka.copycat.data.Time\", \"version\": 1 }"),
-                converted.get(JsonSchema.ENVELOPE_SCHEMA_FIELD_NAME));
-        JsonNode payload = converted.get(JsonSchema.ENVELOPE_PAYLOAD_FIELD_NAME);
-        assertTrue(payload.isInt());
-        assertEquals(14400000, payload.longValue());
-    }
-
-    @Test
-    public void timestampToJson() throws IOException {
-        GregorianCalendar calendar = new GregorianCalendar(1970, Calendar.JANUARY, 1, 0, 0, 0);
-        calendar.setTimeZone(TimeZone.getTimeZone("UTC"));
-        calendar.add(Calendar.MILLISECOND, 2000000000);
-        calendar.add(Calendar.MILLISECOND, 2000000000);
-        java.util.Date date = calendar.getTime();
-
-        JsonNode converted = parse(converter.fromCopycatData(TOPIC, Timestamp.SCHEMA, date));
-        validateEnvelope(converted);
-        assertEquals(parse("{ \"type\": \"int64\", \"optional\": false, \"name\": \"org.apache.kafka.copycat.data.Timestamp\", \"version\": 1 }"),
-                converted.get(JsonSchema.ENVELOPE_SCHEMA_FIELD_NAME));
-        JsonNode payload = converted.get(JsonSchema.ENVELOPE_PAYLOAD_FIELD_NAME);
-        assertTrue(payload.isLong());
-        assertEquals(4000000000L, payload.longValue());
-    }
-
-
-    @Test
-    public void nullSchemaAndPrimitiveToJson() {
-        // This still needs to do conversion of data, null schema means "anything goes"
-        JsonNode converted = parse(converter.fromCopycatData(TOPIC, null, true));
-        validateEnvelopeNullSchema(converted);
-        assertTrue(converted.get(JsonSchema.ENVELOPE_SCHEMA_FIELD_NAME).isNull());
-        assertEquals(true, converted.get(JsonSchema.ENVELOPE_PAYLOAD_FIELD_NAME).booleanValue());
-    }
-
-    @Test
-    public void nullSchemaAndArrayToJson() {
-        // This still needs to do conversion of data, null schema means "anything goes". Make sure we mix and match
-        // types to verify conversion still works.
-        JsonNode converted = parse(converter.fromCopycatData(TOPIC, null, Arrays.asList(1, "string", true)));
-        validateEnvelopeNullSchema(converted);
-        assertTrue(converted.get(JsonSchema.ENVELOPE_SCHEMA_FIELD_NAME).isNull());
-        assertEquals(JsonNodeFactory.instance.arrayNode().add(1).add("string").add(true),
-                converted.get(JsonSchema.ENVELOPE_PAYLOAD_FIELD_NAME));
-    }
-
-    @Test
-    public void nullSchemaAndMapToJson() {
-        // This still needs to do conversion of data, null schema means "anything goes". Make sure we mix and match
-        // types to verify conversion still works.
-        Map<String, Object> input = new HashMap<>();
-        input.put("key1", 12);
-        input.put("key2", "string");
-        input.put("key3", true);
-        JsonNode converted = parse(converter.fromCopycatData(TOPIC, null, input));
-        validateEnvelopeNullSchema(converted);
-        assertTrue(converted.get(JsonSchema.ENVELOPE_SCHEMA_FIELD_NAME).isNull());
-        assertEquals(JsonNodeFactory.instance.objectNode().put("key1", 12).put("key2", "string").put("key3", true),
-                converted.get(JsonSchema.ENVELOPE_PAYLOAD_FIELD_NAME));
-    }
-
-    @Test
-    public void nullSchemaAndMapNonStringKeysToJson() {
-        // This still needs to do conversion of data, null schema means "anything goes". Make sure we mix and match
-        // types to verify conversion still works.
-        Map<Object, Object> input = new HashMap<>();
-        input.put("string", 12);
-        input.put(52, "string");
-        input.put(false, true);
-        JsonNode converted = parse(converter.fromCopycatData(TOPIC, null, input));
-        validateEnvelopeNullSchema(converted);
-        assertTrue(converted.get(JsonSchema.ENVELOPE_SCHEMA_FIELD_NAME).isNull());
-        assertTrue(converted.get(JsonSchema.ENVELOPE_PAYLOAD_FIELD_NAME).isArray());
-        ArrayNode payload = (ArrayNode) converted.get(JsonSchema.ENVELOPE_PAYLOAD_FIELD_NAME);
-        assertEquals(3, payload.size());
-        Set<JsonNode> payloadEntries = new HashSet<>();
-        for (JsonNode elem : payload)
-            payloadEntries.add(elem);
-        assertEquals(new HashSet<>(Arrays.asList(JsonNodeFactory.instance.arrayNode().add("string").add(12),
-                        JsonNodeFactory.instance.arrayNode().add(52).add("string"),
-                        JsonNodeFactory.instance.arrayNode().add(false).add(true))),
-                payloadEntries
-        );
-    }
-
-
-    @Test(expected = DataException.class)
-    public void mismatchSchemaJson() {
-        // If we have mismatching schema info, we should properly convert to a DataException
-        converter.fromCopycatData(TOPIC, Schema.FLOAT64_SCHEMA, true);
-    }
-
-
-
-    @Test
-    public void noSchemaToCopycat() {
-        Map<String, Boolean> props = Collections.singletonMap("schemas.enable", false);
-        converter.configure(props, true);
-        assertEquals(new SchemaAndValue(null, true), converter.toCopycatData(TOPIC, "true".getBytes()));
-    }
-
-    @Test
-    public void noSchemaToJson() {
-        Map<String, Boolean> props = Collections.singletonMap("schemas.enable", false);
-        converter.configure(props, true);
-        JsonNode converted = parse(converter.fromCopycatData(TOPIC, null, true));
-        assertTrue(converted.isBoolean());
-        assertEquals(true, converted.booleanValue());
-    }
-
-    @Test
-    public void testCacheSchemaToJsonConversion() {
-        Cache<Schema, ObjectNode> cache = Whitebox.getInternalState(converter, "fromCopycatSchemaCache");
-        assertEquals(0, cache.size());
-
-        // Repeated conversion of the same schema, even if the schema object is different should return the same Java
-        // object
-        converter.fromCopycatData(TOPIC, SchemaBuilder.bool().build(), true);
-        assertEquals(1, cache.size());
-
-        converter.fromCopycatData(TOPIC, SchemaBuilder.bool().build(), true);
-        assertEquals(1, cache.size());
-
-        // Validate that a similar, but different schema correctly returns a different schema.
-        converter.fromCopycatData(TOPIC, SchemaBuilder.bool().optional().build(), true);
-        assertEquals(2, cache.size());
-    }
-
-
-    private JsonNode parse(byte[] json) {
-        try {
-            return objectMapper.readTree(json);
-        } catch (IOException e) {
-            fail("IOException during JSON parse: " + e.getMessage());
-            throw new RuntimeException("failed");
-        }
-    }
-
-    private JsonNode parse(String json) {
-        try {
-            return objectMapper.readTree(json);
-        } catch (IOException e) {
-            fail("IOException during JSON parse: " + e.getMessage());
-            throw new RuntimeException("failed");
-        }
-    }
-
-    private void validateEnvelope(JsonNode env) {
-        assertNotNull(env);
-        assertTrue(env.isObject());
-        assertEquals(2, env.size());
-        assertTrue(env.has(JsonSchema.ENVELOPE_SCHEMA_FIELD_NAME));
-        assertTrue(env.get(JsonSchema.ENVELOPE_SCHEMA_FIELD_NAME).isObject());
-        assertTrue(env.has(JsonSchema.ENVELOPE_PAYLOAD_FIELD_NAME));
-    }
-
-    private void validateEnvelopeNullSchema(JsonNode env) {
-        assertNotNull(env);
-        assertTrue(env.isObject());
-        assertEquals(2, env.size());
-        assertTrue(env.has(JsonSchema.ENVELOPE_SCHEMA_FIELD_NAME));
-        assertTrue(env.get(JsonSchema.ENVELOPE_SCHEMA_FIELD_NAME).isNull());
-        assertTrue(env.has(JsonSchema.ENVELOPE_PAYLOAD_FIELD_NAME));
-    }
-}

http://git-wip-us.apache.org/repos/asf/kafka/blob/f2031d40/copycat/runtime/src/main/java/org/apache/kafka/copycat/cli/CopycatDistributed.java
----------------------------------------------------------------------
diff --git a/copycat/runtime/src/main/java/org/apache/kafka/copycat/cli/CopycatDistributed.java b/copycat/runtime/src/main/java/org/apache/kafka/copycat/cli/CopycatDistributed.java
deleted file mode 100644
index 8dfefaa..0000000
--- a/copycat/runtime/src/main/java/org/apache/kafka/copycat/cli/CopycatDistributed.java
+++ /dev/null
@@ -1,67 +0,0 @@
-/**
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements.  See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License.  You may obtain a copy of the License at
- *
- *    http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- **/
-
-package org.apache.kafka.copycat.cli;
-
-import org.apache.kafka.common.annotation.InterfaceStability;
-import org.apache.kafka.common.utils.Utils;
-import org.apache.kafka.copycat.runtime.Copycat;
-import org.apache.kafka.copycat.runtime.Worker;
-import org.apache.kafka.copycat.runtime.distributed.DistributedConfig;
-import org.apache.kafka.copycat.runtime.distributed.DistributedHerder;
-import org.apache.kafka.copycat.runtime.rest.RestServer;
-import org.apache.kafka.copycat.storage.KafkaOffsetBackingStore;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import java.util.Collections;
-import java.util.Map;
-
-/**
- * <p>
- * Command line utility that runs Copycat in distributed mode. In this mode, the process joints a group of other workers
- * and work is distributed among them. This is useful for running Copycat as a service, where connectors can be
- * submitted to the cluster to be automatically executed in a scalable, distributed fashion. This also allows you to
- * easily scale out horizontally, elastically adding or removing capacity simply by starting or stopping worker
- * instances.
- * </p>
- */
-@InterfaceStability.Unstable
-public class CopycatDistributed {
-    private static final Logger log = LoggerFactory.getLogger(CopycatDistributed.class);
-
-    public static void main(String[] args) throws Exception {
-        if (args.length < 1) {
-            log.info("Usage: CopycatDistributed worker.properties");
-            System.exit(1);
-        }
-
-        String workerPropsFile = args[0];
-        Map<String, String> workerProps = !workerPropsFile.isEmpty() ?
-                Utils.propsToStringMap(Utils.loadProps(workerPropsFile)) : Collections.<String, String>emptyMap();
-
-        DistributedConfig config = new DistributedConfig(workerProps);
-        Worker worker = new Worker(config, new KafkaOffsetBackingStore());
-        RestServer rest = new RestServer(config);
-        DistributedHerder herder = new DistributedHerder(config, worker, rest.advertisedUrl());
-        final Copycat copycat = new Copycat(worker, herder, rest);
-        copycat.start();
-
-        // Shutdown will be triggered by Ctrl-C or via HTTP shutdown request
-        copycat.awaitStop();
-    }
-}

http://git-wip-us.apache.org/repos/asf/kafka/blob/f2031d40/copycat/runtime/src/main/java/org/apache/kafka/copycat/cli/CopycatStandalone.java
----------------------------------------------------------------------
diff --git a/copycat/runtime/src/main/java/org/apache/kafka/copycat/cli/CopycatStandalone.java b/copycat/runtime/src/main/java/org/apache/kafka/copycat/cli/CopycatStandalone.java
deleted file mode 100644
index 3869552..0000000
--- a/copycat/runtime/src/main/java/org/apache/kafka/copycat/cli/CopycatStandalone.java
+++ /dev/null
@@ -1,98 +0,0 @@
-/**
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements.  See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License.  You may obtain a copy of the License at
- * <p/>
- * http://www.apache.org/licenses/LICENSE-2.0
- * <p/>
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- **/
-
-package org.apache.kafka.copycat.cli;
-
-import org.apache.kafka.common.annotation.InterfaceStability;
-import org.apache.kafka.common.utils.Utils;
-import org.apache.kafka.copycat.runtime.ConnectorConfig;
-import org.apache.kafka.copycat.runtime.Copycat;
-import org.apache.kafka.copycat.runtime.Herder;
-import org.apache.kafka.copycat.runtime.Worker;
-import org.apache.kafka.copycat.runtime.rest.RestServer;
-import org.apache.kafka.copycat.runtime.rest.entities.ConnectorInfo;
-import org.apache.kafka.copycat.runtime.standalone.StandaloneConfig;
-import org.apache.kafka.copycat.runtime.standalone.StandaloneHerder;
-import org.apache.kafka.copycat.storage.FileOffsetBackingStore;
-import org.apache.kafka.copycat.util.Callback;
-import org.apache.kafka.copycat.util.FutureCallback;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import java.util.Arrays;
-import java.util.Collections;
-import java.util.Map;
-
-/**
- * <p>
- * Command line utility that runs Copycat as a standalone process. In this mode, work is not
- * distributed. Instead, all the normal Copycat machinery works within a single process. This is
- * useful for ad hoc, small, or experimental jobs.
- * </p>
- * <p>
- * By default, no job configs or offset data is persistent. You can make jobs persistent and
- * fault tolerant by overriding the settings to use file storage for both.
- * </p>
- */
-@InterfaceStability.Unstable
-public class CopycatStandalone {
-    private static final Logger log = LoggerFactory.getLogger(CopycatStandalone.class);
-
-    public static void main(String[] args) throws Exception {
-
-        if (args.length < 2) {
-            log.info("Usage: CopycatStandalone worker.properties connector1.properties [connector2.properties ...]");
-            System.exit(1);
-        }
-
-        String workerPropsFile = args[0];
-        Map<String, String> workerProps = !workerPropsFile.isEmpty() ?
-                Utils.propsToStringMap(Utils.loadProps(workerPropsFile)) : Collections.<String, String>emptyMap();
-
-        StandaloneConfig config = new StandaloneConfig(workerProps);
-        Worker worker = new Worker(config, new FileOffsetBackingStore());
-        RestServer rest = new RestServer(config);
-        Herder herder = new StandaloneHerder(worker);
-        final Copycat copycat = new Copycat(worker, herder, rest);
-        copycat.start();
-
-        try {
-            for (final String connectorPropsFile : Arrays.copyOfRange(args, 1, args.length)) {
-                Map<String, String> connectorProps = Utils.propsToStringMap(Utils.loadProps(connectorPropsFile));
-                FutureCallback<Herder.Created<ConnectorInfo>> cb = new FutureCallback<>(new Callback<Herder.Created<ConnectorInfo>>() {
-                    @Override
-                    public void onCompletion(Throwable error, Herder.Created<ConnectorInfo> info) {
-                        if (error != null)
-                            log.error("Failed to create job for {}", connectorPropsFile);
-                        else
-                            log.info("Created connector {}", info.result().name());
-                    }
-                });
-                herder.putConnectorConfig(
-                        connectorProps.get(ConnectorConfig.NAME_CONFIG),
-                        connectorProps, false, cb);
-                cb.get();
-            }
-        } catch (Throwable t) {
-            log.error("Stopping after connector error", t);
-            copycat.stop();
-        }
-
-        // Shutdown will be triggered by Ctrl-C or via HTTP shutdown request
-        copycat.awaitStop();
-    }
-}


Mime
View raw message