Incompatibility between KafkaJS and Confluent "round robin" partition assigment

[apeloquin-agilysys]

Environment Information

• OS: Mac M3 Sonoma 14.6.1
• Node Version: 20.14.0
• NPM Version: 10.7.0
• confluent-kafka-javascript version: 0.5.2

Steps to Reproduce

Here's the scenario we're trying solve:

• Existing production application using KafkaJS
• Need a zero-downtime transition to a new version of same application using Confluent library
• To provide continuity and avoid message loss or repeat, the same consumer group IDs are used in both versions of the app
• Typical Kubernetes rollout, pods with new version are brought online, and then pods with older version are shutdown

The problem:

Consumer encountered error while consuming. Retrying. Error details: KafkaJSProtocolError: Broker: Inconsistent group protocol : Error: Broker: Inconsistent group protocol

Although both libraries default to "round robin", they are clearly different protocol implementations. Looking at the groups on the server I see:

Library	partitionAssignor
KafkaJS	RoundRobinAssigner
Confluent	roundrobin

I'm not sure if this will be considered a bug, I realize this may fall under the category of "as expected" or "unavoidable". If that's the case, then my ask would be to help us determine a zero-downtime rollout strategy for the transition. Today we can only bring up the new (Confluent) version of the app in an environment successfully if we shut down all instances of the old (KafkaJS) version first. That will not be a viable option for us in a production environment.

I've included (perhaps unnecessarily) a few tests the highlight the incompatibility.

import {KafkaJS as Confluent, RdKafka} from "@confluentinc/kafka-javascript";
import {Consumer, ConsumerConfig, Kafka, PartitionAssigner, PartitionAssigners} from "kafkajs";

const TOPIC = "test-confluent-topic";
const GROUP_ID = "test-confluent-group";

describe("Partition assigners", () => {
  let kafkaJSKafka: Kafka;
  let kafkaJSConsumer: Consumer;

  let confluentKafka: Confluent.Kafka;
  let confluentAdmin: Confluent.Admin;
  let confluentConsumer: Confluent.Consumer;

  before(async () => {
    kafkaJSKafka = new Kafka({brokers: ["localhost:9092"]});

    confluentKafka = new Confluent.Kafka({kafkaJS: {brokers: ["localhost:9092"]}});

    confluentAdmin = confluentKafka.admin();
    await confluentAdmin.connect();
    await confluentAdmin.createTopics({topics: [{topic: TOPIC}]});
    await confluentAdmin.disconnect();
  });

  beforeEach(async () => {});

  afterEach(async () => {
    await kafkaJSConsumer?.disconnect();
    await confluentConsumer?.disconnect();
  });

  it("KafkaJS and Confluent both with default (KafkaJS first)", async () => {
    await setupKafkaJSConsumer();
    await setupConfluentConsumer();
  });

  it("KafkaJS and Confluent both with default (Confluent first)", async () => {
    await setupConfluentConsumer();
    await setupKafkaJSConsumer();
  });

  it("KafkaJS and Confluent both explicitly set to round robin", async () => {
    await setupKafkaJSConsumer(PartitionAssigners.roundRobin);
    await setupConfluentConsumer(Confluent.PartitionAssigners.roundRobin);
  });

  async function setupKafkaJSConsumer(...assigners: PartitionAssigner[]) {
    let ready = false;
    const config: ConsumerConfig = {groupId: GROUP_ID};
    if (assigners.length) config.partitionAssigners = assigners;
    kafkaJSConsumer = kafkaJSKafka.consumer(config);
    kafkaJSConsumer.on(kafkaJSConsumer.events.GROUP_JOIN, (event: any) => {
      setImmediate(async () => (ready = true));
    });
    await kafkaJSConsumer.connect();
    await kafkaJSConsumer.subscribe({topic: TOPIC});
    await kafkaJSConsumer.run({
      eachMessage: async () => {}
    });
    await until(() => ready);
  }

  async function setupConfluentConsumer(...assigners: Confluent.PartitionAssigners[]) {
    let ready = false;
    const config: Confluent.ConsumerConstructorConfig = {
      kafkaJS: {groupId: GROUP_ID, fromBeginning: true},
      rebalance_cb: (err: any, assignment: any, consumer: any) => {
        if (err.code !== RdKafka.CODES.ERRORS.ERR__ASSIGN_PARTITIONS) return;
        if (!ready) ready = true;
      }
    };
    if (assigners.length) config.kafkaJS!.partitionAssigners = assigners;
    confluentConsumer = confluentKafka.consumer(config);
    await confluentConsumer.connect();
    await confluentConsumer.subscribe({topic: TOPIC});
    await confluentConsumer.run({
      eachMessage: async () => {}
    });
    await until(() => ready);
  }

  async function until(condition: () => boolean) {
    const timeout = 10000;
    const finish = Date.now() + timeout;
    while (Date.now() <= finish) {
      const result = condition();
      if (result) return;
      await new Promise(resolve => setTimeout(resolve, 50));
    }
    throw new Error(`Failed within ${timeout!}ms`);
  }
});

Resulting Confluent error:

{
  message: 'Consumer encountered error while consuming. Retrying. Error details: KafkaJSProtocolError: Broker: Inconsistent group protocol : Error: Broker: Inconsistent group protocol\n' +
    '    at Function.createLibrdkafkaError [as create] (/xxxxx/node_modules/@confluentinc/kafka-javascript/lib/error.js:456:10)\n' +
    '    at /xxxxx/node_modules/@confluentinc/kafka-javascript/lib/kafka-consumer.js:536:29\n' +
    '    at callbackTrampoline (node:internal/async_hooks:130:17)',
  name: 'rdkafka#consumer-2',
  fac: 'BINDING',
  timestamp: 1734045157428
}

[milindl]

I see potentially this workaround, in your KafkaJS code: tulios/kafkajs#463 (comment)

As a workaround, you can pass in a custom partitionAssigner to the Consumer constructor. You could copy our RoundRobinAssigner and just change the name to roundrobin. I believe that should be enough to allow your Java members to join the group - however, I cannot guarantee that our implementation works the same as the Java one, which is needed for it to actually work. You would need to verify this first.

So first, you'd need to create your custom assigner:

kafka.consumer({
    groupId: 'my-group',
    partitionAssigners: [
        newRoundRobin, // assigner name = 'roundrobin'
        roundRobin // assigner name = 'RoundRobinAssigner'
    ]
})

The consumers can be restarted one at a time if you make this change, and the unrestarted consumers will keep working without an issue since if given multiple assigners, the group leader picks the assigner that's common within all consumers in the group.

Once you're done with this, you can start replacing the consumers piecewise with Confluent consumers.

It's not even necessary for the newRoundRobin and the Confluent implementation to be exactly the same, just for the name to be the same. The implementation to be used is decided by whoever is chosen to be the consumer group leader by the groupcoordinator broker. For the best results and consistency of assignment, of course they should be the same, but for a short while during the migration, it should be okay.

I haven't yet tried this approach out myself, I'll do it in some days, and update the migration guide with the instructions.

[apeloquin-agilysys]

I have confirmed this is a viable workaround. Thanks @milindl!

const bridgePartitionAssigner: PartitionAssigner = (config: {
  cluster: Cluster;
  groupId: string;
  logger: Logger;
}): Assigner => {
  const assigner = PartitionAssigners.roundRobin({
    cluster: config.cluster,
    groupId: config.groupId,
    logger: config.logger
  });
  return {...assigner, name: "roundrobin"};
};

[dhruv-nudge]

@milindl I am also facing same issue, can you suggest anything we can do in the confluent-kafka-javascript package side to handle this?

kafka.consumer({
	kafkaJS: {
		groupId: group,
		logger: this.kafkaInstance.kafkaLogger,
		fromBeginning: true,
		partitionAssignors: ['RoundRobinAssigner' as KafkaJS.PartitionAssignors],
	}
)}

I tried this, but wasn't working

[KaiWedekind]

+1

[milindl]

Here's how to go from KafkaJS's round robin assigner to this libraries roundrobin assigner without any downtime for consuming from the consumer group in question (using @apeloquin-agilysys's assignor implementation as the bridge):

Suppose there are N consumers connected to a consumer group. Each of them is using KafkaJS in the beginning.

1. In the KafkaJS code, add the following change:

const bridgePartitionAssigner = (config) => {
    const assigner = Kafka.PartitionAssigners.roundRobin({
      cluster: config.cluster,
      groupId: config.groupId,
      logger: config.logger
    });
    return {...assigner, name: "roundrobin"};
  };

// .... And after adding above, make following addition to consumer config
// ...
        groupId: groupName,
        partitionAssigners: [Kafka.PartitionAssigners.roundRobin, bridgePartitionAssigner]
// ...

2. Restart consumers one-by-one. Now, there should be N consumers, all using KafkaJS, but they will all have 2 partition assigners available.
3. Start your first confluent-kafka-javascript consumer. It will connect to the consumer group successfully. Subsequently, any number of confluent-kafka-javascript consumers can be run without any issues in the same consumer group, and once enough confluent-kafka-javascript consumers are created, the KafkaJS consumers can be stopped.

This way allows changing consumers out one-by-one (so there's no global downtime required, the consumers which are running can continue consumption from the consumer group.)

[milindl]

Closing this as using an intermediate assignor solves the incompatibility without complete downtime for the consumer group.