[CAMEL-18327] camel-kafka: Kafka consumer closes when it is paused

• Type: Task
• Status: Resolved
• Priority: Major
• Resolution: Fixed
• Affects Version/s: 3.17.0
• Fix Version/s: 3.18.3, 3.20.0
• Component/s: camel-kafka
• Estimated Complexity: Novice
• Assignee: Otavio Rodolfo Piske
• Reporter: rupam
• Created: 31/Jul/22 03:31
• Updated: 19/Oct/22 16:26
• Resolved: 12/Oct/22 16:20

Description

 

In the startPolling() method, if the consumer is suspending, then the unsubscribe method is called which permanently closes the consumer

 

    Duration pollDuration = Duration.ofMillis(pollTimeoutMs);
    while (isKafkaConsumerRunnable() && isConnected() && pollExceptionStrategy.canContinue()) {
        ConsumerRecords<Object, Object> allRecords = consumer.poll(pollDuration);
        if (consumerListener != null) {
            if (!consumerListener.afterConsume(consumer)) {
                continue;
            }
        }

        ProcessingResult result = recordProcessorFacade.processPolledRecords(allRecords);

        if (result.isBreakOnErrorHit()) {
            LOG.debug("We hit an error ... setting flags to force reconnect");
            // force re-connect
            setReconnect(true);
            setConnected(false);
        }

        updateTaskState();
    }

    if (!isConnected()) {
        LOG.debug("Not reconnecting, check whether to auto-commit or not ...");
        commitManager.commit();
    }

    safeUnsubscribe();
} catch (InterruptException e) {

 

 

Issue Links

• is a clone of: CAMEL-17947 camel-kafka: fix concurrent access in camel-health
• links to: GitHub Pull Request #8349
• links to: GitHub Pull Request #8351