Kafka consumer unable to connect to Kafka after Kafka restart

[Avinashshetty44]

Kafka Receiver is unable to consume messages from a topic after Kafka is restarted.

public void subscribeAndUpdatePipelineStatues(String topic) {
    ReceiverOptions<String, byte[]> receiverConfig = getReceiverConfig(topic);
    pipelineStatusKafkaReceiver = KafkaReceiver.create(
        receiverConfig.subscription(Collections.singleton(topic)));
    pipelineStatusKafkaReceiver.receive()
        .retryWhen(Retry.backoff(300, Duration.of(10L, ChronoUnit.SECONDS)))
        .doOnError(throwable -> LOGGER.error("Error from retry"))
        .subscribe(
            receiverRecord -> Mono.just("Kafka: Listening Pipeline status on {}")
                .doOnEach(logOnNext(s -> LOGGER.info(s, topic)))
                .flatMap(s -> updatePipelineStatus(receiverRecord))
                .doOnSuccess(receiveRecord -> `receiverRecord.receiverOffset().acknowledge())`
                .subscribe());

Scenario:
Initially, the Kafka receiver worked as expected but once the Kafka broker restarts and comes back, the same receiver is unable to receive messages and there's no error seen in the logs except warn

We followed the approach mentioned here in the documentation: https://projectreactor.io/docs/kafka/release/reference/#_error_handling_2
Kafka reactor version: 1.3.10
Please let us know the solution

[garyrussell]

If it still occurs with 1.3.17; please provide complete instructions to reproduce.