Перебрать объект KStream и распечатать его значение - PullRequest
Новая
       89

Перебрать объект KStream и распечатать его значение

0 голосов
/ 02 октября 2019

Я пытаюсь следовать одному из примеров кодов для Kafka Stream, чтобы убедиться, что мой потребитель может правильно использовать тему, я хочу повторить и распечатать событие / значение, которое он читает. Из документации здесь я попытался использовать foreach, а следующий код - это мой. 

        @Component
        public static class PageCountSink {

        private final Log log = LogFactory.getLog(getClass());
        @StreamListener
        public void process(@Input((AnalyticsBinding.CONSUMER_IN)) KStream<String, PageViewEvent> event) {

            event
                    .foreach((key,value) -> log.info("Test"));

        }

И я получаю эту ошибку: - 

2019-10-02 16:59:17.866 ERROR 30553 --- [-StreamThread-2] o.a.k.s.p.i.AssignedStreamsTasks         : stream-thread [consumer-2-453a13f7-6551-4d6a-b692-6d35c9a57195-StreamThread-2] Failed to process stream task 0_0 due to the following error:

java.lang.ClassCastException: class java.lang.String cannot be cast to class com.example.analytics.PageViewEvent (java.lang.String is in module java.base of loader 'bootstrap'; com.example.analytics.PageViewEvent is in unnamed module of loader 'app')
    at org.apache.kafka.streams.kstream.internals.KStreamPeek$KStreamPeekProcessor.process(KStreamPeek.java:42) ~[kafka-streams-2.0.1.jar:na]
    at org.apache.kafka.streams.processor.internals.ProcessorNode$1.run(ProcessorNode.java:50) ~[kafka-streams-2.0.1.jar:na]
    at org.apache.kafka.streams.processor.internals.ProcessorNode.runAndMeasureLatency(ProcessorNode.java:244) ~[kafka-streams-2.0.1.jar:na]
    at org.apache.kafka.streams.processor.internals.ProcessorNode.process(ProcessorNode.java:133) ~[kafka-streams-2.0.1.jar:na]
    at org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forward(ProcessorContextImpl.java:143) ~[kafka-streams-2.0.1.jar:na]
    at org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forward(ProcessorContextImpl.java:126) ~[kafka-streams-2.0.1.jar:na]
    at org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forward(ProcessorContextImpl.java:90) ~[kafka-streams-2.0.1.jar:na]
    at org.apache.kafka.streams.kstream.internals.KStreamMapValues$KStreamMapProcessor.process(KStreamMapValues.java:41) ~[kafka-streams-2.0.1.jar:na]
    at org.apache.kafka.streams.processor.internals.ProcessorNode$1.run(ProcessorNode.java:50) ~[kafka-streams-2.0.1.jar:na]
    at org.apache.kafka.streams.processor.internals.ProcessorNode.runAndMeasureLatency(ProcessorNode.java:244) ~[kafka-streams-2.0.1.jar:na]
    at org.apache.kafka.streams.processor.internals.ProcessorNode.process(ProcessorNode.java:133) ~[kafka-streams-2.0.1.jar:na]
    at org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forward(ProcessorContextImpl.java:143) ~[kafka-streams-2.0.1.jar:na]
    at org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forward(ProcessorContextImpl.java:126) ~[kafka-streams-2.0.1.jar:na]
    at org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forward(ProcessorContextImpl.java:90) ~[kafka-streams-2.0.1.jar:na]
    at org.apache.kafka.streams.processor.internals.SourceNode.process(SourceNode.java:87) ~[kafka-streams-2.0.1.jar:na]
    at org.apache.kafka.streams.processor.internals.StreamTask.process(StreamTask.java:302) ~[kafka-streams-2.0.1.jar:na]
    at org.apache.kafka.streams.processor.internals.AssignedStreamsTasks.process(AssignedStreamsTasks.java:94) ~[kafka-streams-2.0.1.jar:na]
    at org.apache.kafka.streams.processor.internals.TaskManager.process(TaskManager.java:409) ~[kafka-streams-2.0.1.jar:na]
    at org.apache.kafka.streams.processor.internals.StreamThread.processAndMaybeCommit(StreamThread.java:964) ~[kafka-streams-2.0.1.jar:na]
    at org.apache.kafka.streams.processor.internals.StreamThread.runOnce(StreamThread.java:832) ~[kafka-streams-2.0.1.jar:na]
    at org.apache.kafka.streams.processor.internals.StreamThread.runLoop(StreamThread.java:767) ~[kafka-streams-2.0.1.jar:na]
    at org.apache.kafka.streams.processor.internals.StreamThread.run(StreamThread.java:736) ~[kafka-streams-2.0.1.jar:na]

2019-10-02 16:59:17.879 ERROR 30553 --- [-StreamThread-2] o.a.k.s.p.internals.StreamThread         : stream-thread [consumer-2-453a13f7-6551-4d6a-b692-6d35c9a57195-StreamThread-2] Encountered the following error during processing:

java.lang.ClassCastException: class java.lang.String cannot be cast to class com.example.analytics.PageViewEvent (java.lang.String is in module java.base of loader 'bootstrap'; com.example.analytics.PageViewEvent is in unnamed module of loader 'app')
    at org.apache.kafka.streams.kstream.internals.KStreamPeek$KStreamPeekProcessor.process(KStreamPeek.java:42) ~[kafka-streams-2.0.1.jar:na]
    at org.apache.kafka.streams.processor.internals.ProcessorNode$1.run(ProcessorNode.java:50) ~[kafka-streams-2.0.1.jar:na]
    at org.apache.kafka.streams.processor.internals.ProcessorNode.runAndMeasureLatency(ProcessorNode.java:244) ~[kafka-streams-2.0.1.jar:na]
    at org.apache.kafka.streams.processor.internals.ProcessorNode.process(ProcessorNode.java:133) ~[kafka-streams-2.0.1.jar:na]
    at org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forward(ProcessorContextImpl.java:143) ~[kafka-streams-2.0.1.jar:na]
    at org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forward(ProcessorContextImpl.java:126) ~[kafka-streams-2.0.1.jar:na]
    at org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forward(ProcessorContextImpl.java:90) ~[kafka-streams-2.0.1.jar:na]
    at org.apache.kafka.streams.kstream.internals.KStreamMapValues$KStreamMapProcessor.process(KStreamMapValues.java:41) ~[kafka-streams-2.0.1.jar:na]
    at org.apache.kafka.streams.processor.internals.ProcessorNode$1.run(ProcessorNode.java:50) ~[kafka-streams-2.0.1.jar:na]
    at org.apache.kafka.streams.processor.internals.ProcessorNode.runAndMeasureLatency(ProcessorNode.java:244) ~[kafka-streams-2.0.1.jar:na]
    at org.apache.kafka.streams.processor.internals.ProcessorNode.process(ProcessorNode.java:133) ~[kafka-streams-2.0.1.jar:na]
    at org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forward(ProcessorContextImpl.java:143) ~[kafka-streams-2.0.1.jar:na]
    at org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forward(ProcessorContextImpl.java:126) ~[kafka-streams-2.0.1.jar:na]
    at org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forward(ProcessorContextImpl.java:90) ~[kafka-streams-2.0.1.jar:na]
    at org.apache.kafka.streams.processor.internals.SourceNode.process(SourceNode.java:87) ~[kafka-streams-2.0.1.jar:na]
    at org.apache.kafka.streams.processor.internals.StreamTask.process(StreamTask.java:302) ~[kafka-streams-2.0.1.jar:na]
    at org.apache.kafka.streams.processor.internals.AssignedStreamsTasks.process(AssignedStreamsTasks.java:94) ~[kafka-streams-2.0.1.jar:na]
    at org.apache.kafka.streams.processor.internals.TaskManager.process(TaskManager.java:409) ~[kafka-streams-2.0.1.jar:na]
    at org.apache.kafka.streams.processor.internals.StreamThread.processAndMaybeCommit(StreamThread.java:964) ~[kafka-streams-2.0.1.jar:na]
    at org.apache.kafka.streams.processor.internals.StreamThread.runOnce(StreamThread.java:832) ~[kafka-streams-2.0.1.jar:na]
    at org.apache.kafka.streams.processor.internals.StreamThread.runLoop(StreamThread.java:767) ~[kafka-streams-2.0.1.jar:na]
    at org.apache.kafka.streams.processor.internals.StreamThread.run(StreamThread.java:736) ~[kafka-streams-2.0.1.jar:na]

Ниже приведены соответствующие фрагменты для этого потребителя 


    interface AnalyticsBinding {

        String CONSUMER_IN = "pcin" ;

        @Input(CONSUMER_IN)
        KStream<String, PageViewEvent> consumerIn();

    }

    @Data
    @AllArgsConstructor
    @NoArgsConstructor
    class PageViewEvent {
        private String userId, page;
        private int randNum;
        private long duration;
    }

Буду признателен, если кто-нибудь может указать мнечего мне не хватаетСпасибо

1 Ответ

1 голос
/ 02 октября 2019

Похоже, он не знает, как десериализовать в PageViewEvent. Возможно, вы настроили его на использование StringSerializer по умолчанию. В этом случае вам необходимо определить Serdes для вашего типа PageViewEvent.

...