将Kafka相同的分区分配给同一组中的使用者

Question

Kafka - Multiple Consumers From Same Group Assigned Same Partition

我刚刚开始学习Kafka和Nodejs。我已经写了一个消费者如下

// consumer.js
const kafka = require('kafka-node');
var client = new kafka.Client('localhost:2181');
var topics = [{
    topic: 'topic-4'
}];

var options = {
    groupId: 'kafka-node-group-2',
    autoCommit: true,
    fetchMaxWaitMs: 1000,
    fetchMaxBytes: 1024 * 1024,
    encoding: 'buffer'
};
var consumer = new kafka.HighLevelConsumer(client, topics, options);

// consumer.payloads has only one entry
console.log('Topic', consumer.payloads[0].topic);
console.log('Group', consumer.options.groupId);
console.log('Assigned Partition:', consumer.payloads[0].partition);

输出

Topic topic-4
Group kafka-node-group-2
Assigned Partition: 0

[topic-4有四个分区。

./desc_topic.sh topic-4
Topic:topic-4   PartitionCount:4    ReplicationFactor:1 Configs:
    Topic: topic-4  Partition: 0    Leader: 0   Replicas: 0 Isr: 0
    Topic: topic-4  Partition: 1    Leader: 2   Replicas: 2 Isr: 2
    Topic: topic-4  Partition: 2    Leader: 0   Replicas: 0 Isr: 0
    Topic: topic-4  Partition: 3    Leader: 2   Replicas: 2 Isr: 2

编辑

我已如下使用ConsumerGroup。

var options = {
    host: 'localhost:2181',  // zookeeper host omit if connecting directly to broker (see kafkaHost below)
    groupId: 'Group-1',
    sessionTimeout: 15000,
    // // An array of partition assignment protocols ordered by preference.
    // // 'roundrobin' or 'range' string for built ins (see below to pass in custom assignment protocol)
    protocol: ['roundrobin']
};
var consumer = new kafka.ConsumerGroup(options, ['topic-4']);

生产者发送的100条消息如下。这就是我知道分配的分区的方式（不是来自consumer对象）。

{
    topic: 'topic-4',
    value: '{"subject":"Message Id 30 "}',
    offset: 172,
    partition: 0,
    highWaterOffset: 173,
    key: null
}

当我运行两个这样的使用者实例（相同的主题和组）时，它们中只有一个接收来自partition-0的所有内容。这不是问题吗？

这是生产者代码。

const kafka = require('kafka-node');
const Client = kafka.Client;
var client = new Client('localhost:2181', 'my-client-id', {
  sessionTimeout: 300,
  spinDelay: 100,
  retries: 2
});

// For this demo we just log client errors to the console.
client.on('error', function(error) {
  console.error(error);
});

var producer = new kafka.HighLevelProducer(client);

producer.on('ready', function() {
    for (var i = 0; i <= 30; i++) {
        let id = 'Message Id ' + i + ' ';
        let msg = {
            'subject': id
        };
        var messageBuffer = Buffer.from(JSON.stringify(msg));

        // Create a new payload
        var payload = [{
            // topic: 'topic-', + (i%2+2),
            topic: 'topic-4',
            messages: messageBuffer,
            timestamp: Date.now(),
            attributes: 1 /* Use GZip compression for the payload */
        }];

        //Send payload to Kafka and log result/error
        producer.send(payload, function(error, result) {
            console.info('Sent payload to Kafka: ', payload);
            if (error) {
                console.error('Error', error);
            } else {
                var formattedResult = result[0];
                console.log('result: ', result)
            }
        });
    }
});

// For this demo we just log producer errors to the console.
producer.on('error', function(error) {
    console.error(error);
});

Answer 1

这是一个已知问题。我也遇到过。如果您使用的Kafka版本比该修复程序的发布版本要新，则可能值得重新检查并可能重新打开此问题。

https://issues.apache.org/jira/browse/KAFKA-6681