Pause and Resume

Purpose

This subtopic addresses the need to temporarily halt and subsequently resume Kafka message consumption within Apache Camel Kafka consumers. Pausing is essential for managing transient processing issues, applying backpressure, or reacting to consumer-side errors without fully stopping the consumer. Resuming restores message flow once the conditions that triggered the pause have been resolved.

Unlike simply stopping and restarting the consumer, pause and resume enable fine-grained control over consumption lifecycle, allowing for smoother error recovery and operational flexibility. This feature is deeply integrated with error handling and circuit breaker mechanisms to enhance consumer resilience.

Functionality

Core Workflow

• Pausing: When a critical processing error occurs (e.g., a message fails processing), the consumer can pause fetching new records from Kafka partitions. This prevents further messages from being consumed and processed until the issue is resolved.

• Resuming: After the root cause of the pause is addressed, the consumer resumes fetching messages, continuing from the last committed offset or seek position.

Key Components and Behavior

• KafkaConsumer: Implements the pause and resume lifecycle by controlling one or more KafkaFetchRecords tasks (worker threads that poll Kafka).

• KafkaFetchRecords Tasks: Each task manages its own Kafka consumer instance and can be individually paused or resumed.

• Pause/Resume Methods in KafkaConsumer:

  • doSuspend() calls pause() on all fetch tasks, effectively halting message polling.

  • doResume() calls resume() on all fetch tasks, restarting message polling.

• Consumer Listener (KafkaConsumerListener): Acts as an observer of consumer processing results, integrating pause and resume logic tightly with error handling:

  • If message processing fails (afterProcess() returns failure), it pauses the consumer and optionally seeks to beginning or end of topic partitions based on a configured SeekPolicy.

  • It monitors when the consumer can be resumed by evaluating a predicate condition (afterConsumeEval), resuming consumption only when safe.

Error and Seek Policy Handling

• Upon detecting failure in message processing:

  • The consumer is paused to prevent fetching more messages.

  • Depending on the configured SeekPolicy, the Kafka consumer seeks to the beginning or end of the assigned partitions to reposition the offset for retry or skipping messages.

• The KafkaConsumerListener continuously evaluates whether the consumer should be resumed by checking a predicate, ensuring that resumption only happens when appropriate (e.g., after manual intervention or recovery).

Integration with Circuit Breaker and Health Checks

• Pausing is often triggered by circuit breaker events to avoid processing during system instability.

• Health checks can report whether consumers are paused, aiding in monitoring and alerting.

Integration

Pause and resume complement the broader Kafka message consumption functionality by enhancing control over the consumer lifecycle without full shutdown. They work alongside:

• Fetch Task Management: Pause/resume operations control the lifecycle of fetch tasks that pull messages from Kafka.

• Consumer Error Handling: Integrate tightly with error strategies to pause on failures and resume when safe.

• Health Monitoring: Provide visibility into consumer paused state for operational insight.

• Resume Strategies: Pause/resume cooperate with resume strategies that manage offsets and rebalance listeners to ensure consumers restart from correct positions.

This subtopic introduces dynamic control over message flow in the consumer, extending beyond simple start/stop semantics covered in the parent topic and other subtopics.

Diagram

sequenceDiagram
    participant KafkaConsumer
    participant FetchTask as KafkaFetchRecords Task
    participant KafkaBroker
    participant ConsumerListener as KafkaConsumerListener

    KafkaConsumer->>FetchTask: start polling
    FetchTask->>KafkaBroker: poll messages
    KafkaBroker-->>FetchTask: ConsumerRecords
    FetchTask->>KafkaConsumer: deliver records
    KafkaConsumer->>ConsumerListener: process records
    alt processing success
        ConsumerListener-->>FetchTask: continue polling
    else processing failure
        ConsumerListener->>FetchTask: pause consumption
        FetchTask-->>KafkaBroker: pause assigned partitions
        ConsumerListener->>FetchTask: seek offsets (beginning/end) if needed
        loop wait for recovery
            ConsumerListener->>ConsumerListener: evaluate resume predicate
            alt resume condition met
                ConsumerListener->>FetchTask: resume consumption
                FetchTask-->>KafkaBroker: resume assigned partitions
                break
            else not ready
                ConsumerListener-->>ConsumerListener: remain paused
            end
        end
    end

Code Snippets Highlight

• Pause and Resume in KafkaConsumer:

@Override
protected void doSuspend() throws Exception {
    for (KafkaFetchRecords task : tasks) {
        LOG.info("Pausing Kafka record fetcher task running client ID {}", task.healthState().getClientId());
        task.pause();
    }
    super.doSuspend();
}

@Override
protected void doResume() throws Exception {
    for (KafkaFetchRecords task : tasks) {
        LOG.info("Resuming Kafka record fetcher task running client ID {}", task.healthState().getClientId());
        task.resume();
    }
    super.doResume();
}

• Pause and Seek Logic in KafkaConsumerListener:

@Override
public boolean afterProcess(ProcessingResult result) {
    if (result.isFailed()) {
        LOG.warn("Pausing consumer due to error on the last processing");
        consumer.pause(consumer.assignment());
        paused = true;

        if (seekPolicy == SeekPolicy.BEGINNING) {
            consumer.seekToBeginning(consumer.assignment());
        } else if (seekPolicy == SeekPolicy.END) {
            consumer.seekToEnd(consumer.assignment());
        }
        return false;
    }
    return true;
}

@Override
public boolean afterConsume(Object ignored) {
    if (paused) {
        if (afterConsumeEval.test(null)) {
            LOG.warn("State changed, therefore resuming the consumer");
            consumer.resume(consumer.assignment());
            return true;
        }
        LOG.warn("The consumer is not yet resumable");
        return false;
    }
    return true;
}

This pause and resume mechanism provides a resilient, responsive approach to managing Kafka consumption flow, tightly integrated with error handling and offset management to support stable and fault-tolerant streaming applications.