使用Spring Kafka ReplyingKafka模板与Kafka Streams应用程序

Question

我想有一个客户端应用程序与请求/响应语义学调用另一个应用程序，这是一个Kafka流应用程序。

我的客户端应用程序基于此示例（基本上没有变化）。我需要从客户端接收消息的应用程序是Kafka Streams应用程序。但是包含相关id的消息头丢失。

Kafka Streams应用程序是一个简单的拓扑结构，用于测试此。。。

    @Bean
    public KafkaStreams stream(KafkaStreamsConfiguration kafkaStreamsConfiguration) {
        final StreamsBuilder builder = new StreamsBuilder();
        builder.<String, String>stream(REQUEST_TOPIC_NAME)
                .groupByKey()
                .count()
                .toStream()
                .mapValues((ValueMapper<Long, String>)String::valueOf)
                .to(REPLY_TOPIC_NAME);

        return new KafkaStreams(builder.build(), kafkaStreamsConfiguration.asProperties());
    }

对于这个POC，我保持它的简单性，让客户机和服务器“同意”主题名称（kRequests和kReplies）。所以在这一点上，我只想得到要识别并返回的相关id。

我现在看到的是

2019-10-01 10:55:38.792  WARN 76830 --- [TaskScheduler-1] o.s.k.r.ReplyingKafkaTemplate            : Reply timed out for: ProducerRecord(topic=kRequests, partition=null, headers=RecordHeaders(headers = [RecordHeader(key = kafka_replyTopic, value = [107, 82, 101, 112, 108, 105, 101, 115]), RecordHeader(key = kafka_correlationId, value = [101, -4, -35, 41, -127, -66, 69, 37, -117, -127, -95, -92, 38, 79, 73, 127])], isReadOnly = true), key=null, value=foo21074, timestamp=null) with correlationId: [135564972083657938538225367552235620735]
2019-10-01 10:55:38.792 ERROR 76830 --- [TaskScheduler-1] org.KRequestingApplication  : Reply timed out

org.springframework.kafka.KafkaException: Reply timed out
    at org.springframework.kafka.requestreply.ReplyingKafkaTemplate.lambda$scheduleTimeout$0(ReplyingKafkaTemplate.java:257) ~[spring-kafka-2.2.8.RELEASE.jar:2.2.8.RELEASE]
    at org.springframework.scheduling.support.DelegatingErrorHandlingRunnable.run(DelegatingErrorHandlingRunnable.java:54) ~[spring-context-5.1.9.RELEASE.jar:5.1.9.RELEASE]
    at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511) ~[na:1.8.0_211]
    at java.util.concurrent.FutureTask.run(FutureTask.java:266) ~[na:1.8.0_211]
    at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.access$201(ScheduledThreadPoolExecutor.java:180) ~[na:1.8.0_211]
    at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:293) ~[na:1.8.0_211]
    at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149) ~[na:1.8.0_211]
    at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624) ~[na:1.8.0_211]
    at java.lang.Thread.run(Thread.java:748) ~[na:1.8.0_211]

在超时时间内，回复主题上没有具有匹配相关性id的消息。似乎至少使用Kafka Streams DSL无法支持回复KafkaTemplate。

Answer 1

你的场景没有意义；您的KStream正在对多个输入进行分组；请求/回复为1请求1回复。

这个很好用。。。

@SpringBootApplication
@EnableKafkaStreams
public class So58193901Application {

    private static final String REQUEST_TOPIC_NAME = "requests";

    private static final String REPLY_TOPIC_NAME = "replies";

    public static void main(String[] args) {
        SpringApplication.run(So58193901Application.class, args);
    }

    @Bean
    public KStream<byte[], byte[]> stream(StreamsBuilder builder) {
        KStream<byte[], byte[]> stream = builder.stream(REQUEST_TOPIC_NAME);
        stream
                .mapValues(val -> new String(val).toUpperCase().getBytes())
                .to(REPLY_TOPIC_NAME);
        return stream;
    }

    @Bean
    public NewTopic topic1() {
        return TopicBuilder.name(REQUEST_TOPIC_NAME).partitions(1).replicas(1).build();
    }

    @Bean
    public NewTopic topic2() {
        return TopicBuilder.name(REPLY_TOPIC_NAME).partitions(1).replicas(1).build();
    }

    @Bean
    public ReplyingKafkaTemplate<String, String, String> template(ProducerFactory<String, String> pf,
            ConcurrentKafkaListenerContainerFactory<String, String> factory) {

        ConcurrentMessageListenerContainer<String, String> replyContainer = factory.createContainer(REPLY_TOPIC_NAME);
        return new ReplyingKafkaTemplate<>(pf, replyContainer);
    }

    @Bean
    public ApplicationRunner runner(ReplyingKafkaTemplate<String, String, String> template) {
        return args -> {
            System.out.println(template.sendAndReceive(new ProducerRecord<>(REQUEST_TOPIC_NAME, "bar", "foo"))
                    .get(10, TimeUnit.SECONDS).value());
        };
    }

}