Reactor Kafka中未创建订阅时出错

Question

消费者配置文件:在这里，我对键和值都使用了StringDeserializers。并且订阅是针对单个主题进行的。

@Bean("errorReceiver")
public ReceiverOptions<Object, String> errorConsumerConfig() {
    Map<String, Object> errorConsumerProps = new HashMap<>();
    errorConsumerProps.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, errorBootstrapServers);
    errorConsumerProps.put(ConsumerConfig.CLIENT_ID_CONFIG, "error-consumer");
    errorConsumerProps.put(ConsumerConfig.GROUP_ID_CONFIG, "error-consumer-1");
    errorConsumerProps.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
    errorConsumerProps.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
    ReceiverOptions<Object, String> errorReceiverOptions = ReceiverOptions.create(errorConsumerProps);
    errorReceiverOptions.subscription(Collections.singleton("order_topic"))
    .addAssignListener(partitions -> log.info("onPartitionsAssigned : {}", partitions))
    .addRevokeListener(partitions -> log.info("onPartitionsRevoked : {}", partitions));
    return errorReceiverOptions;
}
}

消费者代码:我在消费者代码中的日志将订阅的主题打印为空。AppUtility正在将数据转换为字符串。

@Autowired
@Qualifier("errorReceiver")
private ReceiverOptions<Object, String> errorReceiverOptions;

@EventListener(ApplicationStartedEvent.class)
public Disposable getErrorsTopic() {
    Flux<ReceiverRecord<Object, Object>> kafkaFlux = KafkaReceiver.create(errorReceiverOptions).receive();
    log.info("subs topics : {}", errorReceiverOptions.subscriptionTopics());
    return kafkaFlux.log()
            .doOnNext(AppUtility::toBinary)
            .doOnError(error -> log.error("error ocurred", error))
            .subscribe();
}

日志：

java.lang.IllegalStateException: No subscriptions have been created
at reactor.kafka.receiver.ReceiverOptions.subscriber(ReceiverOptions.java:385) ~[reactor-kafka-1.3.4.jar:1.3.4]
at reactor.kafka.receiver.internals.ConsumerEventLoop$SubscribeEvent.run(ConsumerEventLoop.java:187) ~[reactor-kafka-1.3.4.jar:1.3.4]
at reactor.core.scheduler.SchedulerTask.call(SchedulerTask.java:68) [reactor-core-3.4.6.jar:3.4.6]
at reactor.core.scheduler.SchedulerTask.call(SchedulerTask.java:28) [reactor-core-3.4.6.jar:3.4.6]
at java.util.concurrent.FutureTask.run(FutureTask.java:266) [na:1.8.0_261]
at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.access$201(ScheduledThreadPoolExecutor.java:180) [na:1.8.0_261]
at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:293) [na:1.8.0_261]
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149) [na:1.8.0_261]
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624) [na:1.8.0_261]
at java.lang.Thread.run(Thread.java:748) [na:1.8.0_261]

pom导入：

<dependency>
    <groupId>io.projectreactor.kafka</groupId>
    <artifactId>reactor-kafka</artifactId>
</dependency>

有没有人遇到过这样的问题？我无法解决这个问题。

Answer 1

我忘记了反应式编程中不可变性的主要概念。通过将选项分配给另一个options对象(配置代码中倒数第四行)解决了这个问题。

public ReceiverOptions<Object, String> errorConsumerConfig() {
Map<String, Object> errorConsumerProps = new HashMap<>();
errorConsumerProps.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, errorBootstrapServers);
errorConsumerProps.put(ConsumerConfig.CLIENT_ID_CONFIG, "error-consumer");
errorConsumerProps.put(ConsumerConfig.GROUP_ID_CONFIG, "error-consumer-1");
errorConsumerProps.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
errorConsumerProps.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
ReceiverOptions<Object, String> errorReceiverOptions = ReceiverOptions.create(errorConsumerProps);
ReceiverOptions<Object, String> options = errorReceiverOptions.subscription(Collections.singleton("order_topic")) // setting the subscription doesn't work unless assigned to an object, reason being immutability
.addAssignListener(partitions -> log.debug("onPartitionsAssigned : {}", partitions))
.addRevokeListener(partitions -> log.debug("onPartitionsRevoked : {}", partitions));
return options; }

编辑:尝试在监听器代码中设置订阅主题。

Answer 2

嗨，请看一下下面的例子，根据例子你遗漏了一些类似下面的代码

        return kafkaFlux.subscribe(record -> {
        ReceiverOffset offset = record.receiverOffset();
        System.out.printf("Received message: topic-partition=%s offset=%d timestamp=%s key=%d value=%s\n",
                offset.topicPartition(),
                offset.offset(),
                dateFormat.format(new Date(record.timestamp())),
                record.key(),
                record.value());
        offset.acknowledge();
        latch.countDown();
    });

https://github.com/reactor/reactor-kafka/blob/main/reactor-kafka-samples/src/main/java/reactor/kafka/samples/SampleConsumer.java