Агрегат кафки с материализованной специфической авро подачей дает исключение NullPointerException - PullRequest
0 голосов
/ 16 апреля 2019

Я пытаюсь создать окно для потока данных, для каждого окна мне нужен список значений в этом окне, и для этого я создал собственную схему avro, которая имеет поле records, которое представляет собой список Input. Агрегатная функция имеет часть Materialized, потому что у меня была эта проблема.

    KStream<String, Input> windowedStream = timestampFilteredStream
            .groupByKey()
            .windowedBy(TimeWindows.of(Duration.ofSeconds(10)).grace(Duration.ofSeconds(5)))
            .aggregate(
                    () -> new InputList(new ArrayList<>()),
                    (key, value, aggregate) -> {
                        if (value != null) {
                            aggregate.getRecords().add(value);
                        }
                        return aggregate;
                    },
                    Materialized.with(Serdes.String(), new SpecificAvroSerde<>())
            )
            .suppress(Suppressed.untilWindowCloses(Suppressed.BufferConfig.unbounded()))
            .toStream()
            .map((window, inputs) -> {
                long windowEnd = window.window().endTime().toEpochMilli();
                String sensorId = window.key();
                Double weightInc = inputs.getRecords().stream().mapToDouble(Input::getWeightActual).reduce(0, (a, b) -> b - a);
                Double lengthInc = inputs.getRecords().stream().mapToDouble(Input::getLengthActual).reduce(0, (a, b) -> b - a);
                Double unitsInc = inputs.getRecords().stream().mapToDouble(Input::getUnitsActual).reduce(0, (a, b) -> b - a);
                Double avgSpeed = inputs.getRecords().stream().mapToDouble(Input::getSpeedActual).average().orElse(0);
                return KeyValue.pair(sensorId, new Input(windowEnd, weightInc, lengthInc, unitsInc, avgSpeed));
            });

    windowedStream.foreach((sensorId, input) -> {
        System.out.println(sensorId + " with computed input " + input);
    });

Трассировка стека:

Exception in thread "sensors-pipeline-35077a4f-40f8-4356-9e56-53938f52c321-StreamThread-1" org.apache.kafka.streams.errors.StreamsException: Exception caught in process. taskId=0_0, processor=KSTREAM-SOURCE-0000000000, topic=sensors, partition=0, offset=0, stacktrace=org.apache.kafka.common.errors.SerializationException: Error serializing Avro message
Caused by: java.lang.NullPointerException
    at io.confluent.kafka.serializers.AbstractKafkaAvroSerializer.serializeImpl(AbstractKafkaAvroSerializer.java:82)
    at io.confluent.kafka.serializers.KafkaAvroSerializer.serialize(KafkaAvroSerializer.java:53)
    at io.confluent.kafka.streams.serdes.avro.SpecificAvroSerializer.serialize(SpecificAvroSerializer.java:65)
    at io.confluent.kafka.streams.serdes.avro.SpecificAvroSerializer.serialize(SpecificAvroSerializer.java:38)
    at org.apache.kafka.streams.state.StateSerdes.rawValue(StateSerdes.java:191)
    at org.apache.kafka.streams.state.internals.MeteredWindowStore.put(MeteredWindowStore.java:117)
    at org.apache.kafka.streams.processor.internals.ProcessorContextImpl$WindowStoreReadWriteDecorator.put(ProcessorContextImpl.java:484)
    at org.apache.kafka.streams.kstream.internals.KStreamWindowAggregate$KStreamWindowAggregateProcessor.process(KStreamWindowAggregate.java:127)
    at org.apache.kafka.streams.processor.internals.ProcessorNode.process(ProcessorNode.java:117)
    at org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forward(ProcessorContextImpl.java:183)
    at org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forward(ProcessorContextImpl.java:162)
    at org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forward(ProcessorContextImpl.java:122)
    at org.apache.kafka.streams.kstream.internals.KStreamTransformValues$KStreamTransformValuesProcessor.process(KStreamTransformValues.java:56)
    at org.apache.kafka.streams.processor.internals.ProcessorNode.process(ProcessorNode.java:117)
    at org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forward(ProcessorContextImpl.java:183)
    at org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forward(ProcessorContextImpl.java:162)
    at org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forward(ProcessorContextImpl.java:122)
    at org.apache.kafka.streams.processor.internals.SourceNode.process(SourceNode.java:87)
    at org.apache.kafka.streams.processor.internals.StreamTask.process(StreamTask.java:364)
    at org.apache.kafka.streams.processor.internals.AssignedStreamsTasks.process(AssignedStreamsTasks.java:199)
    at org.apache.kafka.streams.processor.internals.TaskManager.process(TaskManager.java:420)
    at org.apache.kafka.streams.processor.internals.StreamThread.runOnce(StreamThread.java:890)
    at org.apache.kafka.streams.processor.internals.StreamThread.runLoop(StreamThread.java:805)
    at org.apache.kafka.streams.processor.internals.StreamThread.run(StreamThread.java:774)

    at org.apache.kafka.streams.processor.internals.StreamTask.process(StreamTask.java:381)
    at org.apache.kafka.streams.processor.internals.AssignedStreamsTasks.process(AssignedStreamsTasks.java:199)
    at org.apache.kafka.streams.processor.internals.TaskManager.process(TaskManager.java:420)
    at org.apache.kafka.streams.processor.internals.StreamThread.runOnce(StreamThread.java:890)
    at org.apache.kafka.streams.processor.internals.StreamThread.runLoop(StreamThread.java:805)
    at org.apache.kafka.streams.processor.internals.StreamThread.run(StreamThread.java:774)
Caused by: org.apache.kafka.common.errors.SerializationException: Error serializing Avro message
Caused by: java.lang.NullPointerException
    at io.confluent.kafka.serializers.AbstractKafkaAvroSerializer.serializeImpl(AbstractKafkaAvroSerializer.java:82)
    at io.confluent.kafka.serializers.KafkaAvroSerializer.serialize(KafkaAvroSerializer.java:53)
    at io.confluent.kafka.streams.serdes.avro.SpecificAvroSerializer.serialize(SpecificAvroSerializer.java:65)
    at io.confluent.kafka.streams.serdes.avro.SpecificAvroSerializer.serialize(SpecificAvroSerializer.java:38)
    at org.apache.kafka.streams.state.StateSerdes.rawValue(StateSerdes.java:191)
    at org.apache.kafka.streams.state.internals.MeteredWindowStore.put(MeteredWindowStore.java:117)
    at org.apache.kafka.streams.processor.internals.ProcessorContextImpl$WindowStoreReadWriteDecorator.put(ProcessorContextImpl.java:484)
    at org.apache.kafka.streams.kstream.internals.KStreamWindowAggregate$KStreamWindowAggregateProcessor.process(KStreamWindowAggregate.java:127)
    at org.apache.kafka.streams.processor.internals.ProcessorNode.process(ProcessorNode.java:117)
    at org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forward(ProcessorContextImpl.java:183)
    at org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forward(ProcessorContextImpl.java:162)
    at org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forward(ProcessorContextImpl.java:122)
    at org.apache.kafka.streams.kstream.internals.KStreamTransformValues$KStreamTransformValuesProcessor.process(KStreamTransformValues.java:56)
    at org.apache.kafka.streams.processor.internals.ProcessorNode.process(ProcessorNode.java:117)
    at org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forward(ProcessorContextImpl.java:183)
    at org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forward(ProcessorContextImpl.java:162)
    at org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forward(ProcessorContextImpl.java:122)
    at org.apache.kafka.streams.processor.internals.SourceNode.process(SourceNode.java:87)
    at org.apache.kafka.streams.processor.internals.StreamTask.process(StreamTask.java:364)
    at org.apache.kafka.streams.processor.internals.AssignedStreamsTasks.process(AssignedStreamsTasks.java:199)
    at org.apache.kafka.streams.processor.internals.TaskManager.process(TaskManager.java:420)
    at org.apache.kafka.streams.processor.internals.StreamThread.runOnce(StreamThread.java:890)
    at org.apache.kafka.streams.processor.internals.StreamThread.runLoop(StreamThread.java:805)
    at org.apache.kafka.streams.processor.internals.StreamThread.run(StreamThread.java:774)

Попытка отладки io.confluent.kafka.serializers.AbstractKafkaAvroSerializer.serializeImpl(AbstractKafkaAvroSerializer.java:82) показывает, что ошибка является частью getId этого кода

            schema = AvroSchemaUtils.getSchema(object);
            int id;
            if (this.autoRegisterSchema) {
                restClientErrorMsg = "Error registering Avro schema: ";
                id = this.schemaRegistry.register(subject, schema);
            } else {
                restClientErrorMsg = "Error retrieving Avro schema: ";
                id = this.schemaRegistry.getId(subject, schema);
            }

Я пытался получить типы из реестра схемы, используя предоставленную тему, и я могу видеть схему.

Кажется, проблема в том, что this.schemaRegistry имеет значение null, однако в моих свойствах я установил

props.put(AbstractKafkaAvroSerDeConfig.SCHEMA_REGISTRY_URL_CONFIG, "http://localhost:8081");

1 Ответ

0 голосов
/ 16 апреля 2019

Я исправил это, переместив определение типа в метод groupByKey, используя

.groupByKey(Grouped.with(Serdes.String(), new SpecificAvroSerde<>()))

Обновление:

, другой параметр не работал, так как значение должноиспользуйте конкретное значение avro serde, которое можно исправить следующим образом:

final Serde<InputList> valueSpecificAvroSerde = new SpecificAvroSerde<>();
final Map<String, String> serdeConfig = Collections.singletonMap("schema.registry.url", "http://localhost:8081");
valueSpecificAvroSerde.configure(serdeConfig, false);

Materialized.with(Serdes.String(), valueSpecificAvroSerde)
...