java - 如何使用 Horton 架构注册表修复 : "Unrecognized token ' Unrecognized': was expecting ('true' , 'false' 或 'null' )"

Question

我正在尝试使用 horton 模式注册表在 kafka 中使用 avro 格式的消息。问题是，当我尝试发布 avro 消息时，我收到此错误:

Caused by: com.fasterxml.jackson.core.JsonParseException: Unrecognized token 'Unrecognized': was expecting ('true', 'false' or 'null')
 at [Source: (String)"Unrecognized field "initialState" (class com.hortonworks.registries.schemaregistry.SchemaVersion), not marked as ignorable (2 known properties: "description", "schemaText"])
 at [Source: org.glassfish.jersey.message.internal.ReaderInterceptorExecutor$UnCloseableInputStream@3daa4db5; line: 1, column: 321] (through reference chain: com.hortonworks.registries.schemaregistry.SchemaVersion["initialState"])"; line: 1, column: 13]
    at com.fasterxml.jackson.core.JsonParser._constructError(JsonParser.java:1804)
    at com.fasterxml.jackson.core.base.ParserMinimalBase._reportError(ParserMinimalBase.java:703)
    at com.fasterxml.jackson.core.json.ReaderBasedJsonParser._reportInvalidToken(ReaderBasedJsonParser.java:2853)
    at com.fasterxml.jackson.core.json.ReaderBasedJsonParser._handleOddValue(ReaderBasedJsonParser.java:1899)
    at com.fasterxml.jackson.core.json.ReaderBasedJsonParser.nextToken(ReaderBasedJsonParser.java:757)
    at com.fasterxml.jackson.databind.ObjectMapper._readTreeAndClose(ObjectMapper.java:4042)
    at com.fasterxml.jackson.databind.ObjectMapper.readTree(ObjectMapper.java:2551)
    at com.hortonworks.registries.schemaregistry.client.SchemaRegistryClient.readCatalogResponse(SchemaRegistryClient.java:644)

我查看了类定义 (SchemaVersion)，发现有注释:@JsonIgnoreProperties(ignoreUnknown = true) 但我仍然收到此错误。

我还使用 gradle 作为构建工具:

compile(group: 'org.apache.avro', name: 'avro', version: '1.8.2')
compile(group: 'com.hortonworks.registries', name: 'schema-registry-serdes', version: '0.7.0')
compile(group: 'com.hortonworks.registries', name: 'schema-registry-client', version: '0.7.0')
// confluent platform 5.1.1 provided with kafka 2.1.0
compile(group: 'org.apache.kafka', name: 'kafka-clients', version: '2.1.0')

compile(group: 'org.glassfish.jersey.core', name: 'jersey-client', version: '2.28')
compile(group: 'org.glassfish.jersey.inject', name: 'jersey-hk2', version: '2.28')

导致错误的代码:打包项目；

import com.hortonworks.registries.schemaregistry.client.SchemaRegistryClient;
import org.apache.avro.generic.GenericRecord;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.*;

import java.util.Properties;
import java.util.concurrent.ExecutionException;

public class HortonToConfluentTest {
    private static final String HORTON_SCHEMA_REGISTRY_URL = "http://localhost:9090/api/v1";
    private static final String BOOTSTRAP_SERVER = "localhost:29092";

    private static final String AVRO_SOURCE_TOPIC = "avro_topic";

    public static void main(String[] args) throws ExecutionException, InterruptedException {
        Properties hortonProducerProperties = new Properties();
        hortonProducerProperties.put(ProducerConfig.CLIENT_ID_CONFIG, "horton-producer");
        hortonProducerProperties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, BOOTSTRAP_SERVER);
        hortonProducerProperties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        hortonProducerProperties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, com.hortonworks.registries.schemaregistry.serdes.avro.kafka.KafkaAvroSerializer.class);
        hortonProducerProperties.put(SchemaRegistryClient.Configuration.SCHEMA_REGISTRY_URL.name(), HORTON_SCHEMA_REGISTRY_URL);


        KafkaProducer<String, GenericRecord> hortonProducer = new KafkaProducer<>(hortonProducerProperties);
        hortonProducer.send(new ProducerRecord<>(AVRO_SOURCE_TOPIC, GenerateRecord.generate(1, "body"))).get();

        hortonProducer.flush();
        hortonProducer.close();
    }
}

架构注册表和使用 docker 运行的 kafka ( https://github.com/TheBookPeople/hortonworks-registry-docker ):

version: '3' services:   db:
    image: mysql:5.7.17
    container_name: db
    hostname: db
    ports:
      - 3306:3306
    environment:
      MYSQL_ROOT_PASSWORD: password
      MYSQL_DATABASE: hortonworks
      MYSQL_USER: hortonworks
      MYSQL_PASSWORD: password   horton-registry:
    image: thebookpeople/hortonworks-registry:latest
    container_name: horton-registry
    hostname: horton-registry
    depends_on:
      - db
    ports:
      - 9090:9090
    environment:
      DB_NAME: hortonworks
      DB_USER: hortonworks
      DB_PASSWORD: password
      DB_PORT: 3306
      DB_HOST: db   zookeeper:
    image: confluentinc/cp-zookeeper:5.1.1
    hostname: zookeeper
    container_name: zookeeper
    ports:
      - 22181:22181
    environment:
      - ZOOKEEPER_CLIENT_PORT=22181   kafka:
    image: confluentinc/cp-enterprise-kafka:5.1.1
    hostname: kafka
    container_name: kafka
    ports:
      - 29092:29092
    depends_on:
      - zookeeper
    environment:
      - KAFKA_ZOOKEEPER_CONNECT=zookeeper:22181
      - KAFKA_LISTENER_SECURITY_PROTOCOL_MAP=PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT
      - KAFKA_ADVERTISED_LISTENERS=PLAINTEXT://kafka:9092,PLAINTEXT_HOST://localhost:29092
      - KAFKA_BROKER_ID=1
      - KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR=1
      - KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS=0

我已经尝试过:

1. 在架构注册表中手动定义架构

2. 更改horton-schema-registry的依赖版本

3. 显式定义对 jackson 的依赖(使用最新可用版本)

Answer 1

问题出在依赖关系上(根据https://hortonworks.com/tutorial/schema-registry-in-trucking-iot-on-hdf/section/4/):

compile(group: 'com.hortonworks.registries', name: 'schema-registry-serdes', version: '0.3.0.3.0.1.1-5')
compile(group: 'javax.xml.bind', name: 'jaxb-api', version: '2.3.0')

而不是:

compile(group: 'com.hortonworks.registries', name: 'schema-registry-serdes', version: '0.7.0')
compile(group: 'com.hortonworks.registries', name: 'schema-registry-client', version: '0.7.0')

最终依赖列表:

compile(group: 'org.apache.avro', name: 'avro', version: '1.8.2')
compile(group: 'com.hortonworks.registries', name: 'schema-registry-serdes', version: '0.3.0.3.0.1.1-5')
// confluent platform 5.1.1 provided with kafka 2.1.0
compile(group: 'org.apache.kafka', name: 'kafka-clients', version: '2.1.0')
compile(group: 'javax.xml.bind', name: 'jaxb-api', version: '2.3.0')