1

我有这个问题：

当我使用处理器 API 从主题中消费时，当在处理器内部使用方法时context().forward(K, V)，Kafka Streams 会引发空指针异常。

这是它的堆栈跟踪：

Exception in thread "StreamThread-1" java.lang.NullPointerException
at org.apache.kafka.streams.processor.internals.StreamTask.forward(StreamTask.java:336)
at org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forward(ProcessorContextImpl.java:187)
at org.apache.kafka.streams.processor.ProcessorContext$forward.call(Unknown Source)
at org.codehaus.groovy.runtime.callsite.CallSiteArray.defaultCall(CallSiteArray.java:48)
at org.codehaus.groovy.runtime.callsite.AbstractCallSite.call(AbstractCallSite.java:113)
at org.codehaus.groovy.runtime.callsite.AbstractCallSite.call(AbstractCallSite.java:133)
at com.bnsf.ltf.processor.ConversionProcessor.process(ConversionProcessor.groovy:23)
at com.bnsf.ltf.processor.ConversionProcessor.process(ConversionProcessor.groovy)
at org.apache.kafka.streams.processor.internals.ProcessorNode.process(ProcessorNode.java:68)
at org.apache.kafka.streams.processor.internals.StreamTask.forward(StreamTask.java:338)
at org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forward(ProcessorContextImpl.java:187)
at org.apache.kafka.streams.processor.internals.SourceNode.process(SourceNode.java:64)
at org.apache.kafka.streams.processor.internals.StreamTask.process(StreamTask.java:174)
at org.apache.kafka.streams.processor.internals.StreamThread.runLoop(StreamThread.java:320)
at org.apache.kafka.streams.processor.internals.StreamThread.run(StreamThread.java:218)

我的 Gradle 依赖项如下所示：

compile('org.codehaus.groovy:groovy-all')
compile('org.apache.kafka:kafka-streams:0.10.0.0')

更新：我尝试使用 0.10.0.1 版本，但仍然抛出相同的错误。

这是我正在构建的拓扑的代码......

 topologyBuilder.addSource('inboundTopic', stringDeserializer, stringDeserializer, conversionConfiguration.inTopic)
    .addProcessor('conversionProcess', new ProcessorSupplier() {
        @Override
        Processor get() {
            return conversionProcessor
        }
    }, 'inboundTopic')
    .addSink('outputTopic', conversionConfiguration.outTopic, stringSerializer, stringSerializer, 'conversionProcess')

    stream = new KafkaStreams(topologyBuilder, streamConfig)
    stream.start()

我的处理器看起来像这样：

@Override
void process(String key, String message) {
    // Call to a service and the return of the service is set on the
    // converted local variable named converted
    context().forward(key, converted)
    context().commit()
}
4

1 回答 1

1

直接提供您的处理器。

.addProcessor('conversionProcess', () -> new MyProcessor(), 'inboundTopic')

MyProcessor反过来，应该继承自AbstractProcessor.

于 2017-08-16T08:12:33.870 回答