[CAMEL-18588] Kafka consumer on any exception should repoll records after the committed offset

• Type: Bug
• Status: Resolved
• Priority: Minor
• Resolution: Fixed
• Affects Version/s: 3.18.0
• Fix Version/s: 3.18.3, 3.20.0
• Component/s: camel-kafka
• Estimated Complexity: Novice
• Assignee: Otavio Rodolfo Piske
• Reporter: rupam
• Created: 06/Oct/22 03:27
• Updated: 03/Oct/23 09:01
• Resolved: 12/Oct/22 16:20

Description

In Camel-Kafka , if max poll records is set to more than 1 and breakOnFirstError is set to TRUE, then on any exception thrown on the first message in the batch will cause the KafkaConsumer to start repolling from zero offset because partitionLastOffset is -1 for the first message

  private boolean processException(
            Exchange exchange, TopicPartition partition, long partitionLastOffset,
            ExceptionHandler exceptionHandler) {

        // processing failed due to an unhandled exception, what should we do
        if (configuration.isBreakOnFirstError()) {
            // we are failing and we should break out
            if (LOG.isWarnEnabled()) {
                LOG.warn("Error during processing {} from topic: {}", exchange, partition.topic(), exchange.getException());
                LOG.warn("Will seek consumer to offset {} and start polling again.", partitionLastOffset);
            }

            // force commit, so we resume on next poll where we failed
            commitManager.forceCommit(partition, partitionLastOffset);

            // continue to next partition
            return true;
        } else {
            // will handle/log the exception and then continue to next
            exceptionHandler.handleException("Error during processing", exchange, exchange.getException());
        }

        return false;
    }

Issue Links

• is a clone of: CAMEL-18587 Kafka Consumer closes on any exception if breakOnFirstError is set to TRUE
• relates to: CAMEL-18760 camel-kafka - Issue using ThrottlingExceptionRoutePolicy with Kafka consumer