Great question! Implementing priority-based Kafka consumption with preemption is a super useful pattern for scenarios where critical messages need to take precedence over non-urgent ones. Let’s walk through how to build this step by step.

The goal here is to enforce two key behaviors:

• Priority First: Consumers always try to read from high-priority topics first.
• Preemption: As soon as new messages arrive in high-priority topics, consumers immediately pause low-priority consumption and switch back.

Kafka doesn’t have built-in support for topic prioritization, so we’ll implement this logic at the consumer application layer.

1. Define Topic Priority Order #

First, formalize your priority hierarchy. For example:

• High-priority: order-urgent
• Low-priority: order-bulk

Store this as an ordered list in your consumer code to enforce the priority sequence.

2. Priority Consumption & Auto-Switch Logic #

The consumer will follow this workflow:

1. Start by subscribing to the highest-priority topic.
2. Poll for messages:
   • If messages are available: process them, and keep polling the same topic.
   • If no messages are found: unsubscribe from the high-priority topic, switch to the next lower-priority topic, and start polling there.

3. Preemption Mechanism #

To handle immediate switching back to high-priority topics when new messages arrive:

• While consuming from a low-priority topic, periodically check if the high-priority topic has unconsumed messages.
• If unconsumed messages are detected, immediately unsubscribe from the low-priority topic, switch back to the high-priority one, and resume consumption.

How to Check for Unconsumed Messages #

You can use the Kafka consumer API to compare the current committed offset with the end offset (latest available message offset) for each partition in the high-priority topic. If the end offset is greater than the committed offset, there are new messages waiting.

Here’s a working example that implements the above logic:

import org.apache.kafka.clients.consumer.*;
import org.apache.kafka.common.TopicPartition;
import java.time.Duration;
import java.util.*;

public class PriorityKafkaConsumer {
    // Define your priority topic list (highest first)
    private static final List<String> PRIORITY_TOPICS = Arrays.asList("order-urgent", "order-bulk");
    private static final Duration POLL_TIMEOUT = Duration.ofMillis(100);
    // Adjust this interval based on your preemption latency requirements
    private static final Duration PREEMPT_CHECK_INTERVAL = Duration.ofSeconds(1);

    public static void main(String[] args) {
        Properties consumerProps = new Properties();
        consumerProps.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        consumerProps.put(ConsumerConfig.GROUP_ID_CONFIG, "priority-order-consumers");
        consumerProps.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
        consumerProps.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
        consumerProps.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "latest");
        consumerProps.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "true");
        consumerProps.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, "1000");

        try (KafkaConsumer<String, String> consumer = new KafkaConsumer<>(consumerProps)) {
            String currentTopic = PRIORITY_TOPICS.get(0);
            consumer.subscribe(Collections.singletonList(currentTopic));
            long lastPreemptCheckTime = System.currentTimeMillis();

            while (true) {
                ConsumerRecords<String, String> records = consumer.poll(POLL_TIMEOUT);

                if (!records.isEmpty()) {
                    // Process messages from current topic
                    processConsumedRecords(records);
                    // Reset check time since we're actively consuming high-priority messages
                    lastPreemptCheckTime = System.currentTimeMillis();
                } else {
                    if (currentTopic.equals(PRIORITY_TOPICS.get(0))) {
                        // Switch to low-priority topic when high-priority is empty
                        consumer.unsubscribe();
                        currentTopic = PRIORITY_TOPICS.get(1);
                        consumer.subscribe(Collections.singletonList(currentTopic));
                        System.out.println("Switched to low-priority topic: " + currentTopic);
                    } else {
                        // Periodically check if high-priority has new messages
                        long currentTime = System.currentTimeMillis();
                        if (currentTime - lastPreemptCheckTime >= PREEMPT_CHECK_INTERVAL.toMillis()) {
                            if (hasUnconsumedMessages(consumer, PRIORITY_TOPICS.get(0))) {
                                // Preempt: switch back to high-priority topic
                                consumer.unsubscribe();
                                currentTopic = PRIORITY_TOPICS.get(0);
                                consumer.subscribe(Collections.singletonList(currentTopic));
                                System.out.println("Preempted back to high-priority topic: " + currentTopic);
                            }
                            lastPreemptCheckTime = currentTime;
                        }
                    }
                }
            }
        }
    }

    private static void processConsumedRecords(ConsumerRecords<String, String> records) {
        for (ConsumerRecord<String, String> record : records) {
            System.out.printf("Consumed from %s | Offset: %d | Key: %s | Value: %s%n",
                    record.topic(), record.offset(), record.key(), record.value());
            // Add your business logic here
        }
    }

    private static boolean hasUnconsumedMessages(KafkaConsumer<String, String> consumer, String topic) {
        // Get all partitions for the high-priority topic
        List<TopicPartition> partitions = consumer.partitionsFor(topic).stream()
                .map(p -> new TopicPartition(topic, p.partition()))
                .toList();
        
        // Temporarily assign partitions to check offsets
        consumer.assign(partitions);
        Map<TopicPartition, OffsetAndMetadata> committedOffsets = consumer.committed(partitions);
        Map<TopicPartition, Long> endOffsets = consumer.endOffsets(partitions);

        // Check if any partition has unconsumed messages
        for (TopicPartition tp : partitions) {
            long committedOffset = committedOffsets.getOrDefault(tp, new OffsetAndMetadata(0)).offset();
            long latestOffset = endOffsets.get(tp);
            if (latestOffset > committedOffset) {
                // Reset consumer back to current subscribed topic before returning
                consumer.subscribe(Collections.singletonList(consumer.subscription().iterator().next()));
                return true;
            }
        }

        // Reset consumer back to current subscribed topic
        consumer.subscribe(Collections.singletonList(consumer.subscription().iterator().next()));
        return false;
    }
}

• Check Interval Tuning: The PREEMPT_CHECK_INTERVAL balances preemption latency and resource usage. Shorter intervals mean faster preemption but more offset checks; longer intervals reduce overhead but delay switching.
• Offset Management: Ensure auto-commit is enabled (or implement manual commit) to avoid duplicate messages when switching topics.
• Multi-Consumer Groups: If you have multiple consumers in the same group, this logic will work per-consumer since each consumer handles its own subscription switching.
• Performance Optimization: For large clusters, consider using the Kafka AdminClient to fetch topic offsets instead of the consumer API, as it avoids temporary partition assignment overhead.

内容的提问来源于stack exchange，提问作者Krati Jain