我有一個(gè) SchemaRegistry 和一個(gè) KafkaBroker，我使用 Avro v1.8.1 從中提取數(shù)據(jù)。對(duì)于反序列化，我一直在使用 Confluent 的KafkaAvroDeserializer。現(xiàn)在我打算重構(gòu)我的代碼以使用 Alpakka 提供的Elasticsearch API，但不幸的是這會(huì)破壞反序列化，因?yàn)樗鼤?huì)導(dǎo)致 NullPointerExceptions：線程“main”中的異常 org.apache.kafka.common.errors.SerializationException：在偏移量 0 處反序列化分區(qū) topic-0 的鍵/值時(shí)出錯(cuò)。如果需要，請(qǐng)尋找過去的記錄以繼續(xù)消費(fèi)。原因：org.apache.kafka.common.errors.SerializationException：反序列化 id 2 的 Avro 消息時(shí)出錯(cuò) 原因：io.confluent.kafka.serializers.AbstractKafkaAvroDeserializer.deserialize（AbstractKafkaAvroDeserializer.java:116）處的 java.lang.NullPointerException io.confluent.kafka.serializers.AbstractKafkaAvroDeserializer.deserialize(AbstractKafkaAvroDeserializer.java:88) 在 io.confluent.kafka.serializers.KafkaAvroDeserializer.deserialize(KafkaAvroDeserializer.java:55) 在 org.apache.kafka.Deserialization.Deserializer.common.serializer在 org.apache.kafka.clients.consumer 中反序列化（Deserializer.java:58）。我一直在使用 Alpakka 的 ConsumerSettings API，如本例中所述：val system = ActorSystem.create();// necessary to convert timestamps correctly in Avro Version 1.8.1 to avoid ClassCastExceptionsSpecificData.get().addLogicalTypeConversion(new TimeConversions.TimestampConversion());val consumerSettings = ConsumerSettings.create(system, new StringDeserializer(), new KafkaAvroDeserializer())    .withBootstrapServers(kafkaBootstrapServerUrl)    .withClientId(InetAddress.getLocalHost().getHostName())    .withGroupId("" + new Random().nextInt())    .withProperty(AbstractKafkaAvroSerDeConfig.SCHEMA_REGISTRY_URL_CONFIG, schemaRegistryUrl)    .withProperty(KafkaAvroDeserializerConfig.SPECIFIC_AVRO_READER_CONFIG, "true")    .withProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest")    .withStopTimeout(Duration.ofSeconds(5));