Apache Kafka 流到流

Question

使用 Apache Kafka 我想将从 topic1 收到的记录传递到 topic2 流。但仔细阅读文档后，我无法将事件发送到 topic2 流。这是我的代码

 @Test
    public void topicToTopic() throws InterruptedException {
        String broker = embeddedKafkaBroker.getBrokersAsString();
        String topic1 = "Consumer-topic-1";
        String topic2 = "Consumer-topic-2";

        Properties config = getSourceConfig(broker);
        Serde<String> stringSerde = Serdes.String();

        StreamsBuilder builder = new StreamsBuilder();
        builder.stream(topic1, Consumed.with(stringSerde, stringSerde))
                .peek((k, v) -> System.out.println("Topic 1 Observed event:" + v))
                .mapValues((ValueMapper<String, String>) String::toUpperCase)
                .to(topic2, Produced.with(stringSerde, stringSerde));
        Topology topology = builder.build();
        KafkaStreams streams = new KafkaStreams(topology, config);
        streams.start();
        Thread.sleep(2000);

        publishMessages(broker, topic1);

        StreamsBuilder builder1 = new StreamsBuilder();
        builder1.stream(topic2, Consumed.with(stringSerde, stringSerde))
                .foreach((key, value) -> System.out.println("key: " + key + " -> " + value));

        Topology topology1 = builder1.build();
        KafkaStreams streams1 = new KafkaStreams(topology1, config);
        streams1.start();
        Thread.sleep(2000);

        streams.close();
        streams1.close();
    }

知道有什么问题吗？

问候

Answer 1

我认为您需要在发布消息之前启动消费者流