当我尝试在 Avro 模式中处理和接收数据时出现空指针异常/未找到异常

Question

我正在使用处理器来使用来自主题的字节数组 serdes 的字节数组数据，将它们处理成通用记录（基于我从 HTTP GET 请求中获得的模式）并将它们发送到具有格式化 avro 模式的主题注册表。

我从 HTTP GET 请求中检索架构并根据它映射我的数据以生成遵循架构的通用记录没有问题。但是，当我尝试将它沉入主题时，我得到一个空指针异常：

org.apache.kafka.common.errors.SerializationException: Error serializing Avro message
    Caused by: java.lang.NullPointerException
atio.confluent.kafka.serializers.AbstractKafkaAvroSerializer.serializeImpl(AbstractKafkaAvroSerializer.
    java:72        )
at io.confluent.kafka.serializers.KafkaAvroSerializer.serialize(KafkaAvroSerializer.java:54)
at 
   org.apache.kafka.streams.processor.internals.RecordCollectorImpl.send(RecordCollectorImpl.java:78)
at org.apache.kafka.streams.processor.internals.SinkNode.process(SinkNode.java:79)
atorg.apache.kafka.streams.processor.internals.ProcessorContextImpl.forward(ProcessorContextImpl
     .java:83)
at streamProcessor.XXXXprocessor.process(XXXXprocessor.java:80)
at streamProcessor.XXXXprocessor.process(XXXXprocessor.java:1)
at org.apache.kafka.streams.processor.internals.ProcessorNode$1.run(ProcessorNode.java:48)
atorg.apache.kafka.streams.processor.internals.StreamsMetricsImpl.measureLatencyNs(StreamsMetr
    icsImpl.java:188)
at org.apache.kafka.streams.processor.internals.ProcessorNode.process(ProcessorNode.java:134)
atorg.apache.kafka.streams.processor.internals.ProcessorContextImpl.forward(ProcessorContextImpl
    .java:111)
at streamProcessor.SelectorProcessor.process(SelectorProcessor.java:33)
at streamProcessor.SelectorProcessor.process(SelectorProcessor.java:1)
at org.apache.kafka.streams.processor.internals.ProcessorNode$1.run(ProcessorNode.java:48)
atorg.apache.kafka.streams.processor.internals.StreamsMetricsImpl.measureLatencyNs(StreamsMetr
    icsImpl.java:188)
at org.apache.kafka.streams.processor.internals.ProcessorNode.process(ProcessorNode.java:134)
atorg.apache.kafka.streams.processor.internals.ProcessorContextImpl.forward(ProcessorContextImpl
    .java:83)
at org.apache.kafka.streams.processor.internals.SourceNode.process(SourceNode.java:70)
at org.apache.kafka.streams.processor.internals.StreamTask.process(StreamTask.java:197)
at org.apache.kafka.streams.processor.internals.StreamThread.runLoop(StreamThread.java:627)
at org.apache.kafka.streams.processor.internals.StreamThread.run(StreamThread.java:361)

这是我的拓扑代码：

//Stream Properties
Properties config = new Properties();
config.put(StreamsConfig.APPLICATION_ID_CONFIG, "processor-kafka-streams234");
config.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "xxxxxxxxxxxxxxxxxxxxxx:xxxx");
config.put(StreamsConfig.KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass().getName());
config.put(StreamsConfig.VALUE_SERDE_CLASS_CONFIG, 
        Serdes.ByteArray().getClass().getName());

config.put(StreamsConfig.TIMESTAMP_EXTRACTOR_CLASS_CONFIG, 
        WallclockTimestampExtractor.class);



//Build topology
TopologyBuilder builder = new TopologyBuilder();
builder.addSource("messages-source", "mytest2");
builder.addProcessor("selector-processor", () -> new SelectorProcessor(), "messages-source");

builder.addProcessor("XXXX-processor", () -> new XXXXprocessor(), "selector-processor");
builder.addSink("XXXX-sink", "XXXXavrotest", new KafkaAvroSerializer(), new               
        KafkaAvroSerializer(), "XXXX-processor");



//Start Streaming
KafkaStreams streaming = new KafkaStreams(builder, config);
streaming.start();
System.out.println("processor streaming...");

在问题论坛上阅读了一些内容后，我发现在创建 KafkaAvroSerializers 时可能需要注入客户端，因此我将那行更改为：

  SchemaRegistryClient client = new 
  CachedSchemaRegistryClient("xxxxxxxxxxxxxxxxxxxxxx:xxxx/subjects/xxxxschemas/versions", 1000);
  builder.addSink("XXXX-sink", "XXXXavrotest", new KafkaAvroSerializer(client), new 
  KafkaAvroSerializer(client), "XXXX-processor");

导致 HTTP 404 Not Found 异常 ...

Answer 1

我的网址有误：P

由于我的主题中的 cleanup.policy 设置，因此密钥必须初始化为 null 之外的其他内容。