Ошибка десериализации сообщения avro kafka из JDBC-коннектора - PullRequest
Новая
       384

Ошибка десериализации сообщения avro kafka из JDBC-коннектора

0 голосов
/ 15 марта 2019

Я пытаюсь выслушать тему, на которую я отправил сообщение, используя функцию подключения kafka к confluent. Однако я не могу десериализовать его. Я считаю, что его сериализация avro, но не смог найти правильный десериализатор.

сообщение похоже на показанное ниже в теме консоли 

null    {"c1":{"int":10},"c2":{"string":"foo"},"create_ts":1552598863000,"update_ts":1552598863000}

Ниже приведен десериализатор 

public class AvroDeserializer<T extends SpecificRecordBase> implements Deserializer<T> {

    private static final Logger LOGGER = LoggerFactory.getLogger(AvroDeserializer.class);

    protected final Class<T> targetType;

    public AvroDeserializer(Class<T> targetType) {
        this.targetType = targetType;
    }

    @Override
    public void close() {
        // No-op
    }

    @Override
    public void configure(Map<String, ?> arg0, boolean arg1) {
        // No-op
    }

    @SuppressWarnings("unchecked")
    @Override
    public T deserialize(String topic, byte[] data) {
        try {
            T result = null;

            if (data != null) {
                LOGGER.debug("data='{}'", DatatypeConverter.printHexBinary(data));

                DatumReader<GenericRecord> datumReader =
                        new SpecificDatumReader<>(targetType.newInstance().getSchema());
                Decoder decoder = DecoderFactory.get().binaryDecoder(data, null);

                result = (T) datumReader.read(null, decoder);
                LOGGER.debug("deserialized data='{}'", result);
            }
            return result;
        } catch (Exception ex) {
            throw new SerializationException(
                    "Can't deserialize data '" + Arrays.toString(data) + "' from topic '" + topic + "'", ex);
        }
    }
}

Исключение 

org.apache.kafka.common.errors.SerializationException: Error deserializing key/value for partition mysql-foobar-0 at offset 10. If needed, please seek past the record to continue consumption.
Caused by: org.apache.kafka.common.errors.SerializationException: Can't deserialize data '[0, 0, 0, 0, 21, 2, 20, 2, 6, 102, 111, 111, -80, -78, -44, -31, -81, 90, -80, -78, -44, -31, -81, 90]' from topic 'mysql-foobar'
Caused by: java.lang.InstantiationException: null
    at sun.reflect.InstantiationExceptionConstructorAccessorImpl.newInstance(InstantiationExceptionConstructorAccessorImpl.java:48) ~[na:1.8.0_131]
    at java.lang.reflect.Constructor.newInstance(Constructor.java:423) ~[na:1.8.0_131]
    at java.lang.Class.newInstance(Class.java:442) ~[na:1.8.0_131]
    at com.spring.kafkaexample.springbootkafkaconsumer.config.AvroDeserializer.deserialize(AvroDeserializer.java:48) ~[classes/:na]
    at com.spring.kafkaexample.springbootkafkaconsumer.config.AvroDeserializer.deserialize(AvroDeserializer.java:18) ~[classes/:na]

1 Ответ

0 голосов
/ 16 марта 2019

, как показано ниже в теме консоли

Не ясно, использовали ли вы kafka-avro-console-consumer или просто kafka-console-consumer. Чтобы узнать, являются ли ваши данные Avro, нужно посмотреть конфигурацию источника / соединителя.

Нет необходимости писать собственный десериализатор. Кроме того, Confluent не использует соглашение Avro schema + message, которое использовался бы в вашем коде (таким образом, почему вы получаете эту ошибку). Сначала вам нужно найти схему в реестре схем.

Добавить репозиторий Confluent Maven 

<repositories>

  <repository>
    <id>confluent</id>
    <url>https://packages.confluent.io/maven/</url>
  </repository>

</repositories>

Затем добавьте зависимость Confluent serializer 

<dependency>
  <groupId>io.confluent</groupId>
  <artifactId>kafka-avro-serializer</artifactId>
  <version>${confluent.version}</version>
</dependency>

Тогда import io.confluent.kafka.serializers.KafkaAvroDeserializer, или используйте этот класс в своих пользовательских конфигах

https://docs.confluent.io/current/clients/install.html#java

ИЛИ , вы можете переключить ваш коннектор MySQL, чтобы не использовать Avro Converter

...