Я реализую KStream-GlobalKTable-Join с использованием Spring-Cloud-Stream, и я столкнулся с проблемой, что операция объединения не дает никаких совпадений, но она определенно должна.Код выглядит следующим образом:
@Component
@EnableBinding(CustomProcessor.class)
public class MyProcessor {
private static final Log LOGGER =
LogFactory.getLog(MyProcessor.class);
@Autowired
private InteractiveQueryService interactiveQueryService;
ReadOnlyKeyValueStore<Object, Object> keyValueStore;
@StreamListener
@SendTo(CustomProcessor.OUTPUT)
public KStream<EventKey, EventEnriched> process(
@Input(CustomProcessor.INPUT) KStream<EventKey, EventEnriched> inputStream,
@Input(CustomProcessor.LOOKUP) GlobalKTable<LookupKey, LookupData> lookupStore
) {
keyValueStore = interactiveQueryService.getQueryableStore("lookupStore", QueryableStoreTypes.keyValueStore());
LOGGER.info("Lookup: " + keyValueStore.get(new LookupKey("google.de")));
return inputStream.leftJoin(
lookupStore,
(inputKey, inputValue) -> {
return new LookupKey(inputValue.getDomain().replace("www.", ""));
},
this::enrichData
);
}
public EventEnriched enrichData(EventEnriched input, LookupData lookupRecord) {
...
}
}
Здесь CustomProcessor :
public interface CustomProcessor extends KafkaStreamsProcessor {
String INPUT = "input";
String OUTPUT = "output";
String LOOKUP = "lookupTable";
@Input(CustomProcessor.LOOKUP)
GlobalKTable<LookupKey, ?> lookupTable();
}
Без вызова строки в MyProcessor
keyValueStore.get(...)
код работает нормально, но GlobalKTable кажется нулевым.Но если я вызову
LOGGER.info("Lookup: " + keyValueStore.get(new LookupKey("google.de")));
для проверки GlobalKTable, при запуске приложения произойдет сбой:
Error starting ApplicationContext. To display the conditions report re-run your application with 'debug' enabled.
2019-06-26T09:04:00.000 [ERROR] [main-858] [org.springframework.boot.SpringApplication] [reportFailure:858] Application run failed
org.springframework.beans.factory.BeanInitializationException: Cannot setup StreamListener for public org.apache.kafka.streams.kstream.KStream MyProcessor.process(org.apache.kafka.streams.kstream.KStream,org.apache.kafka.streams.kstream.GlobalKTable); nested exception is java.lang.reflect.InvocationTargetException
at org.springframework.cloud.stream.binder.kafka.streams.KafkaStreamsStreamListenerSetupMethodOrchestrator.orchestrateStreamListenerSetupMethod(KafkaStreamsStreamListenerSetupMethodOrchestrator.java:214)
at org.springframework.cloud.stream.binding.StreamListenerAnnotationBeanPostProcessor.doPostProcess(StreamListenerAnnotationBeanPostProcessor.java:226)
at org.springframework.cloud.stream.binding.StreamListenerAnnotationBeanPostProcessor.lambda$postProcessAfterInitialization$0(StreamListenerAnnotationBeanPostProcessor.java:196)
at java.base/java.lang.Iterable.forEach(Iterable.java:75)
at org.springframework.cloud.stream.binding.StreamListenerAnnotationBeanPostProcessor.injectAndPostProcessDependencies(StreamListenerAnnotationBeanPostProcessor.java:330)
at org.springframework.cloud.stream.binding.StreamListenerAnnotationBeanPostProcessor.afterSingletonsInstantiated(StreamListenerAnnotationBeanPostProcessor.java:113)
at org.springframework.beans.factory.support.DefaultListableBeanFactory.preInstantiateSingletons(DefaultListableBeanFactory.java:866)
at org.springframework.context.support.AbstractApplicationContext.finishBeanFactoryInitialization(AbstractApplicationContext.java:877)
at org.springframework.context.support.AbstractApplicationContext.refresh(AbstractApplicationContext.java:549)
at org.springframework.boot.web.servlet.context.ServletWebServerApplicationContext.refresh(ServletWebServerApplicationContext.java:142)
at org.springframework.boot.SpringApplication.refresh(SpringApplication.java:775)
at org.springframework.boot.SpringApplication.refreshContext(SpringApplication.java:397)
at org.springframework.boot.SpringApplication.run(SpringApplication.java:316)
at org.springframework.boot.SpringApplication.run(SpringApplication.java:1260)
at org.springframework.boot.SpringApplication.run(SpringApplication.java:1248)
at Transformer.main(Transformer.java:31)
Caused by: java.lang.reflect.InvocationTargetException: null
at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
at java.base/jdk.internal.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
at java.base/java.lang.reflect.Method.invoke(Method.java:566)
at org.springframework.cloud.stream.binder.kafka.streams.KafkaStreamsStreamListenerSetupMethodOrchestrator.orchestrateStreamListenerSetupMethod(KafkaStreamsStreamListenerSetupMethodOrchestrator.java:179)
... 15 common frames omitted
Caused by: java.lang.NullPointerException: null
at MyProcessor.process(MyProcessor.java:62)
... 20 common frames omitted
Process finished with exit code 1
Кто-нибудь видит проблему в коде?Как я могу проверить содержимое GlobaKTable?
С уважением, Мартин