Kafka 流：状态存储在左连接期间未初始化

Question

我正在尝试加入两个 Kafka 主题。一个是 KStream，另一个是 KTable。左连接抱怨处理器的状态存储不存在。我确实看过 kafka GitHub 和其他地方的许多代码示例，其中 StateStore 未由 KStream 客户端代码明确创建。请告知以下代码中缺少的内容。

应用程序流与用户 table 一起发出记录，应用程序和用户一起发出记录。应用程序的所有者是用户。

版本：1.1.0

谢谢

  public void process() {
    Properties config = new Properties();
    config.put(StreamsConfig.APPLICATION_ID_CONFIG, APPLICATION_ID);
    config.put(StreamsConfig.CLIENT_ID_CONFIG, CLIENT_ID);
    config.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, Common.KAFKA_SOCKET);
    config.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass());
    config.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, CustomSerdes.applicationSerde);
    config.put(StreamsConfig.DEFAULT_TIMESTAMP_EXTRACTOR_CLASS_CONFIG, MyEventTimeExtractor.class);
    config.put(StreamsConfig.STATE_DIR_CONFIG, "/tmp/kafka-streams");

    // User properties: userid, username
    KTable<String, User> users = new StreamsBuilder().table(TOPIC_USERS,
        Consumed.with(Serdes.String(), CustomSerdes.serdeFor(User.class)));

    StreamsBuilder builder = new StreamsBuilder();
    // Application properties: id, name
    KStream<String, Application> stream = builder.stream(TOPIC_APPLICATIONS);

    stream.
        map((appId, app) -> KeyValue.pair(app.getOwnerId(), app.getAppId()))
        .leftJoin(users, (app, user) -> "a:" + app + " u:" + user.getUserName())
        .to(OUTPUT_TOPIC);

    KafkaStreams streams = new KafkaStreams(builder.build(), config);
    StreamsManager.startAndHandleShutdown(streams);
  }

错误：

Exception in thread "main" org.apache.kafka.streams.errors.TopologyException: Invalid topology: StateStore topic-users-STATE-STORE-0000000000 is not added yet.
    at org.apache.kafka.streams.processor.internals.InternalTopologyBuilder.connectProcessorAndStateStore(InternalTopologyBuilder.java:716)
    at org.apache.kafka.streams.processor.internals.InternalTopologyBuilder.connectProcessorAndStateStores(InternalTopologyBuilder.java:615)
    at org.apache.kafka.streams.kstream.internals.KStreamImpl.doStreamTableJoin(KStreamImpl.java:797)
    at org.apache.kafka.streams.kstream.internals.KStreamImpl.leftJoin(KStreamImpl.java:817)
    at org.apache.kafka.streams.kstream.internals.KStreamImpl.leftJoin(KStreamImpl.java:805)
    at com.test.streams.users.AppWithUserConsumerMain.process(AppWithUserConsumerMain.java:50)

Answer 1

为了能够使用联接，联接的两个部分（在您的情况下 KStream 和 KTable）应该从同一个 StreamsBuilder 创建，因此它们将属于相同的拓扑结构。

在您的情况下，您创建了两个 StreamsBuilder，因此，KStream 和 KTable 不属于同一拓扑。

Kafka 流：状态存储在左连接期间未初始化

Kafka streams: State store is not initialised during left join

apache-kafka

apache-kafka-streams