static Schema<?> buildSchema(String keySchema, String schema, String keyValueEncodingType) {
        switch (keyValueEncodingType) {
            case KEY_VALUE_ENCODING_TYPE_NOT_SET:
                return buildComponentSchema(schema);
            case KEY_VALUE_ENCODING_TYPE_SEPARATED:
                return Schema.KeyValue(buildComponentSchema(keySchema), buildComponentSchema(schema),
                        KeyValueEncodingType.SEPARATED);
            case KEY_VALUE_ENCODING_TYPE_INLINE:
                return Schema.KeyValue(buildComponentSchema(keySchema), buildComponentSchema(schema),
                        KeyValueEncodingType.INLINE);
            default:
                throw new IllegalArgumentException("Invalid KeyValueEncodingType "
                        + keyValueEncodingType + ", only: 'none','separated' and 'inline");
        }
    }
@Test
    public void testBuildSchema() {
        // default
        assertEquals(SchemaType.BYTES, CmdProduce.buildSchema("string", "bytes", CmdProduce.KEY_VALUE_ENCODING_TYPE_NOT_SET).getSchemaInfo().getType());

        // simple key value
        assertEquals(SchemaType.KEY_VALUE, CmdProduce.buildSchema("string", "string", "separated").getSchemaInfo().getType());
        assertEquals(SchemaType.KEY_VALUE, CmdProduce.buildSchema("string", "string", "inline").getSchemaInfo().getType());

        KeyValueSchema<?, ?> composite1 = (KeyValueSchema<?, ?>) CmdProduce.buildSchema("string",
                "json:{\"type\": \"record\",\"namespace\": \"com.example\",\"name\": \"FullName\", \"fields\": [{ \"name\": \"a\", \"type\": \"string\" }]}",
                "inline");
        assertEquals(KeyValueEncodingType.INLINE, composite1.getKeyValueEncodingType());
        assertEquals(SchemaType.STRING, composite1.getKeySchema().getSchemaInfo().getType());
        assertEquals(SchemaType.JSON, composite1.getValueSchema().getSchemaInfo().getType());

        KeyValueSchema<?, ?> composite2 = (KeyValueSchema<?, ?>) CmdProduce.buildSchema(
                "json:{\"type\": \"record\",\"namespace\": \"com.example\",\"name\": \"FullName\", \"fields\": [{ \"name\": \"a\", \"type\": \"string\" }]}",
                "avro:{\"type\": \"record\",\"namespace\": \"com.example\",\"name\": \"FullName\", \"fields\": [{ \"name\": \"a\", \"type\": \"string\" }]}",
                "inline");
        assertEquals(KeyValueEncodingType.INLINE, composite2.getKeyValueEncodingType());
        assertEquals(SchemaType.JSON, composite2.getKeySchema().getSchemaInfo().getType());
        assertEquals(SchemaType.AVRO, composite2.getValueSchema().getSchemaInfo().getType());
    }