join - KStream-GlobalKTable-Join using Spring-Cloud-Stream - 如何检查 GlobalKTable 的内容？

Question

我正在使用 Spring-Cloud-Stream 实现 KStream-GlobalKTable-Join 并且我面临的问题是，连接操作没有得到任何匹配，但它绝对应该。代码如下所示：

@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 的内容？

最好的问候马丁

标签： joinapache-kafkaspring-cloud-stream

Answer 1

现在我越来越接近问题了。我试图查询lookupStore。如果我使用

final ReadOnlyKeyValueStore<LookupKey, LookupData> lookupStore =
            interactiveQueryService.getQueryableStore("myStore", QueryableStoreTypes.<LookupKey, LookupData>keyValueStore())

然后

lookupStore.get(key)

从不返回值。但是如果我像这样创建一个 HashMap：

final KeyValueIterator<LookupKey, LookupData> lookups = lookupStore.all();

Map<LookupKey, LookupData> lookupMap = new HashMap<>();
while (lookups.hasNext()) {
    KeyValue<LookupKey, LookupData> nextLookup = lookups.next();
    lookupMap.put(nextLookup.key, nextLookup.value);
}
lookups.close();

hashMap 包含正确的数据，并向每个键返回正确的值。但是由于某种原因，无法加入 GlobalKTable 本身。它永远不会得到任何匹配。