Кафка: Как получить предыдущие разделы для потребителя после ребалансировки - PullRequest
Новая
       16

Кафка: Как получить предыдущие разделы для потребителя после ребалансировки

0 голосов
/ 21 июня 2019

У меня есть тема Kafka gbtopic с 12 разделами и группой потребителей gbconsumers с тремя пользователями C1, C2, C3, подписавшимися на эту тему. Когда все потребители запущены, каждому потребителю выделяется 4 раздела. Тогда я остановил потребителя С3. Теперь разделы перебалансированы с 6 разделами на C1 и C2. Теперь я начал потреблять C3, и разделы перебалансированы и распределены по C1, C2, C3. Я ожидаю вернуть тот же набор разделов с C1, C2, C3, который был назначен до остановки C3.

Я использую StickyAssignor для partition.assignment.strategy. Я выполняю операции в следующем порядке: 

Start C1
Start C2
Start C3
Stop C3
Restart C3
Ниже приводится моя клиентская программа: 
public class ConsumerApp {
    public static void main(String[] args) {
        Scanner scanIn = new Scanner(System.in);
        System.out.print("Enter Consumer Name :: ");
        String consumerId = scanIn.nextLine().trim();
        scanIn.close();

        Properties props = new Properties();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.put(ConsumerConfig.GROUP_ID_CONFIG, "gbconsumers");
        props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "true");
        props.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, "1000");
        props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
        props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, "30000");
        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");
        props.put(ConsumerConfig.PARTITION_ASSIGNMENT_STRATEGY_CONFIG, StickyAssignor.class.getName());
        props.put(ConsumerConfig.CLIENT_ID_CONFIG, consumerId);

        KafkaConsumer<String, String> kafkaConsumer = new KafkaConsumer<>(props);
        ConsumerRebalanceListener listener = new ConsumerRebalanceListener() {
            @Override
            public void onPartitionsRevoked(Collection<TopicPartition> collection) {
                System.out.println("-------------------------------------------------------------------------------------");
                System.out.println(consumerId + " :: onPartitionsRevoked :: " + collection);
            }

            @Override
            public void onPartitionsAssigned(Collection<TopicPartition> collection) {
                System.out.println(consumerId + " :: onPartitionsAssigned :: " + collection + "\n");
            }
        };
        kafkaConsumer.subscribe(Collections.singletonList("gbtopic"), listener);
        while (true) {
            ConsumerRecords<String, String> records = kafkaConsumer.poll((100));
            for (ConsumerRecord<String, String> record : records) {
                performSomeOperation(record);
            }
        }
    }

    private static void performSomeOperation(ConsumerRecord<String, String> record) {
        // operations
    }
}

Вывод, когда все три потребителя работают, до остановки C3 


    C1 :: onPartitionsAssigned :: [gbtopic-10, gbtopic-11, gbtopic-8, gbtopic-9]
    C2 :: onPartitionsAssigned :: [gbtopic-2, gbtopic-3, gbtopic-4, gbtopic-5]
    C3 :: onPartitionsAssigned :: [gbtopic-0, gbtopic-1, gbtopic-6, gbtopic-7]

Вывод после остановки и запуска C3: 


    C1 :: onPartitionsAssigned :: [gbtopic-0, gbtopic-10, gbtopic-11, gbtopic-6]
    C2 :: onPartitionsAssigned :: [gbtopic-4, gbtopic-5, gbtopic-1, gbtopic-7]
    C3 :: onPartitionsAssigned :: [gbtopic-2, gbtopic-3, gbtopic-8, gbtopic-9]

Я ожидаю, что разделы, назначенные каждому потребителю после остановки / перезапуска C3, будут такими же, какими они были до остановки / перезапуска. Я скучаю по какой-либо собственности или есть другой способ добиться этого?

...