Kafka镜像制造者未复制消费者组偏移量

以下是一个使用Java代码解决“Kafka镜像制造者未复制消费者组偏移量”的示例：

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRebalanceListener;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.errors.WakeupException;

import java.util.Collection;
import java.util.Collections;
import java.util.Properties;

public class KafkaConsumerExample {

    private KafkaConsumer<String, String> consumer;

    public static void main(String[] args) {
        KafkaConsumerExample example = new KafkaConsumerExample();
        example.run();
    }

    public KafkaConsumerExample() {
        Properties props = new Properties();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.put(ConsumerConfig.GROUP_ID_CONFIG, "my-consumer-group");
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");

        consumer = new KafkaConsumer<>(props);
    }

    public void run() {
        consumer.subscribe(Collections.singletonList("my-topic"), new ConsumerRebalanceListener() {
            @Override
            public void onPartitionsRevoked(Collection<TopicPartition> partitions) {
                // 在重新分配分区之前，提交偏移量
                consumer.commitSync();
            }

            @Override
            public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
                // 在重新分配分区之后，从最近提交的偏移量处开始消费
                consumer.seekToBeginning(partitions);
            }
        });

        try {
            while (true) {
                consumer.poll(100);
                // 处理消费的消息
            }
        } catch (WakeupException e) {
            // 忽略WakeUpException异常
        } finally {
            consumer.close();
        }
    }
}

在这个示例中，我们创建了一个Kafka消费者，并订阅了一个主题。在消费者的构造函数中，我们设置了必要的属性，如引导服务器地址、消费者组ID和键值的反序列化器。然后，我们使用subscribe方法订阅主题，并传入一个ConsumerRebalanceListener实例。

ConsumerRebalanceListener接口定义了两个方法：onPartitionsRevoked和onPartitionsAssigned。在onPartitionsRevoked方法中，我们在重新分配分区之前提交偏移量，以确保在重新分配之后不会丢失偏移量。在onPartitionsAssigned方法中，我们从最近提交的偏移量处开始消费。

在主循环中，我们使用poll方法来拉取和处理消息。当我们想要关闭消费者时，我们可以调用consumer.wakeup()来触发一个WakeupException异常，并在catch块中忽略它。最后，在finally块中关闭消费者。

通过使用上述示例代码，您可以解决“Kafka镜像制造者未复制消费者组偏移量”的问题，并确保消费者组的偏移量在重新分配分区时不会丢失。