apache-kafka - Kafka Consumer Group Id 和消费者再平衡问题

Question

我在生产服务器中使用 Kafka 0.10.0 和 zookeeper 3.4.6。我有 20 个主题，每个主题有大约 50 个分区。我总共有 100 个消费者，每个消费者都订阅了不同的主题和分区。所有消费者都有相同的 groupId。那么，如果为特定主题添加或删除消费者，那么附加到不同主题的消费者也会进行重新平衡吗？

我的消费者代码是:

public static void main(String[] args) {
        String groupId = "prod"
        String topicRegex = args[0]
        String consumerTimeOut = "10000"
        int n_threads = 1
        if (args && args.size() > 1) {
            ConfigLoader.init(args[1])
        }
        else {
            ConfigLoader.init('development')
        }
        if(args && args.size() > 2 && args[2].isInteger()){
            n_threads = (args[2]).toInteger()
        }

        ExecutorService executor = Executors.newFixedThreadPool(n_threads)
        addShutdownHook(executor)
        String zooKeeper = ConfigLoader.conf.zookeeper.hostName
        List<Runnable> taskList = []
        for(int i = 0; i < n_threads; i++){
            KafkaConsumer example = new KafkaConsumer(zooKeeper, groupId, topicRegex, consumerTimeOut)
            taskList.add(example)
        }
        taskList.each{ task ->
            executor.submit(task)
        }
        executor.shutdown()
        executor.awaitTermination(Long.MAX_VALUE, TimeUnit.SECONDS)
    }

private static ConsumerConfig createConsumerConfig(String a_zookeeper, String a_groupId, String consumerTimeOut) {

        Properties props = new Properties()
        props.put("zookeeper.connect", a_zookeeper)
        props.put("group.id", a_groupId)
        props.put("zookeeper.session.timeout.ms", "10000")
        props.put("rebalance.backoff.ms","10000")
        props.put("zookeeper.sync.time.ms","200")
        props.put("rebalance.max.retries","10")
        props.put("enable.auto.commit", "false")
        props.put("consumer.timeout.ms", consumerTimeOut)
        props.put("auto.offset.reset", "smallest")
        return new ConsumerConfig(props)

    }

public void run(String topicRegex) {
        String threadName = Thread.currentThread().getName()
        logger.info("{} [{}] main Starting", TAG, threadName)
        Map<String, Integer> topicCountMap = new HashMap<String, Integer>()
        List<KafkaStream<byte[], byte[]>> streams = consumer.createMessageStreamsByFilter(new Whitelist(topicRegex),1)
        ConsumerConnector consumerConnector = consumer

        for (final KafkaStream stream : streams) {
            ConsumerIterator<byte[], byte[]> consumerIterator = stream.iterator()
            List<Object> batchTypeObjList = []
            String topic
            String topicObjectType
            String method
            String className
            String deserialzer
            Integer batchSize = 200
            while (true){
                boolean hasNext = false
                try {
                    hasNext = consumerIterator.hasNext()
                } catch (InterruptedException interruptedException) {
                    //if (exception instanceof InterruptedException) {
                    logger.error("{} [{}]Interrupted Exception: {}", TAG, threadName, interruptedException.getMessage())
                    throw interruptedException
                    //} else {
                } catch(ConsumerTimeoutException timeoutException){
                    logger.error("{} [{}] Timeout Exception: {}", TAG, threadName, timeoutException.getMessage())
                    topicListMap.each{ eachTopic, value ->
                        batchTypeObjList = topicListMap.get(eachTopic)
                        if(batchTypeObjList != null && !batchTypeObjList.isEmpty()) {
                            def dbObject = topicConfigMap.get(eachTopic)
                            logger.debug("{} [{}] Timeout Happened.. Indexing remaining objects in list for topic: {}", TAG, threadName, eachTopic)
                            className = dbObject.get(KafkaTopicConfigEntity.CLASS_NAME_KEY)
                            method = dbObject.get(KafkaTopicConfigEntity.METHOD_NAME_KEY)
                            int sleepTime = 0
                            if(dbObject.get(KafkaTopicConfigEntity.CONUSMER_SLEEP_IN_MS) != null)
                                sleepTime = dbObject.get(KafkaTopicConfigEntity.CONUSMER_SLEEP_IN_MS)?.toInteger()
                            executeMethod(className, method, batchTypeObjList)
                            batchTypeObjList.clear()
                            topicListMap.put(eachTopic,batchTypeObjList)
                            sleep(sleepTime)
                        }
                    }
                    consumer.commitOffsets()
                    continue
                } catch(Exception exception){
                    logger.error("{} [{}]Exception: {}", TAG, threadName, exception.getMessage())
                    throw exception
                }
                if(hasNext) {
                    def consumerObj = consumerIterator.next()
                    logger.debug("{} [{}] partition name: {}", TAG, threadName, consumerObj.partition())
                    topic = consumerObj.topic()
                    DBObject dbObject = topicConfigMap.get(topic)
                    logger.debug("{} [{}] topic name: {}", TAG, threadName, topic)
                    topicObjectType = dbObject.get(KafkaTopicConfigEntity.TOPIC_OBJECT_TYPE_KEY)
                    deserialzer = KafkaConfig.DEFAULT_DESERIALIZER
                    if(KafkaConfig.DESERIALIZER_MAP.containsKey(topicObjectType)){
                        deserialzer = KafkaConfig.DESERIALIZER_MAP.get(topicObjectType)
                    }
                    className = dbObject.get(KafkaTopicConfigEntity.CLASS_NAME_KEY)
                    method = dbObject.get(KafkaTopicConfigEntity.METHOD_NAME_KEY)
                    boolean isBatchJob = dbObject.get(KafkaTopicConfigEntity.IS_BATCH_JOB_KEY)
                    if(dbObject.get(KafkaTopicConfigEntity.BATCH_SIZE_KEY) != null)
                        batchSize = dbObject.get(KafkaTopicConfigEntity.BATCH_SIZE_KEY)
                    else
                        batchSize = 1
                    Object queueObj = (Class.forName(deserialzer)).deserialize(consumerObj.message())
                    int sleepTime = 0
                    if(dbObject.get(KafkaTopicConfigEntity.CONUSMER_SLEEP_IN_MS) != null)
                        sleepTime = dbObject.get(KafkaTopicConfigEntity.CONUSMER_SLEEP_IN_MS)?.toInteger()
                    if(isBatchJob == true){
                        batchTypeObjList = topicListMap.get(topic)
                        batchTypeObjList.add(queueObj)
                        if(batchTypeObjList.size() == batchSize) {
                            executeMethod(className, method, batchTypeObjList)
                            batchTypeObjList.clear()
                            sleep(sleepTime)
                        }
                        topicListMap.put(topic,batchTypeObjList)
                    } else {
                        executeMethod(className, method, queueObj)
                        sleep(sleepTime)
                    }
                    consumer.commitOffsets()
                }
            }
            logger.debug("{} [{}] Shutting Down Process ", TAG, threadName)
        }
    }

我们将不胜感激。

Answer 1

每当消费者离开或加入消费者组时，整个组都会进行重新平衡。由于该组跟踪其成员订阅的所有主题的所有分区，您的想法是正确的，这可能导致未订阅相关主题的消费者重新平衡。

请看下面的小测试来说明这一点，我有一个有两个主题 test1(2 个分区)和 test2(9 个分区)的代理，并且正在启动两个消费者，它们都具有相同的消费者组，每个只订阅一个的两个主题。如您所见，当 consumer2 加入组时，consumer1 将所有分区撤销并重新分配，因为整个组重新平衡。

Subscribing consumer1 to topic test1
Starting thread for consumer1
Polling consumer1
consumer1 got 0 partitions revoked!
consumer1 got 2 partitions assigned!
Polling consumer1
Polling consumer1
Polling consumer1
Subscribing consumer2 to topic test2
Starting thread for consumer2
Polling consumer2
Polling consumer1
consumer2 got 0 partitions revoked!
Polling consumer1
Polling consumer1
consumer1 got 2 partitions revoked!
consumer2 got 9 partitions assigned!
consumer1 got 2 partitions assigned!
Polling consumer2
Polling consumer1
Polling consumer2
Polling consumer1
Polling consumer2