Use cooperative rebalance

This should minimize disruptions on rebalance. Issue: BB-441
scality · Oct 6, 2023 · f00734a · f00734a
1 parent 58ef55b
commit f00734a
Showing 1 changed file with 4 additions and 3 deletions.
diff --git a/lib/BackbeatConsumer.js b/lib/BackbeatConsumer.js
@@ -182,6 +182,7 @@ class BackbeatConsumer extends EventEmitter {
             'allow.auto.create.topics': true,
             'statistics.interval.ms': 1000,
             'rebalance_cb': this._onRebalance.bind(this),
+            'partition.assignment.strategy': 'cooperative-sticky',
         };
         const topicParams = {};
         if (this._fromOffset !== undefined) {
@@ -496,7 +497,7 @@ class BackbeatConsumer extends EventEmitter {
             this._log.debug('rdkafka.assign', { err, assignment });
 
             try {
-                this._consumer.assign(assignment);
+                this._consumer.incrementalAssign(assignment);
             } catch (e) {
                 // Ignore exceptions if we are not connected
                 const logger = this._consumer.isConnected() ? this._log.error : this._log.debug;
@@ -510,7 +511,7 @@ class BackbeatConsumer extends EventEmitter {
             });
 
             if (!this._processingQueue || this._processingQueue.length() + this._processingQueue.running() === 0) {
-                this._consumer.unassign();
+                this._consumer.incrementalUnassign(assignment);
                 return;
             }
 
@@ -521,7 +522,7 @@ class BackbeatConsumer extends EventEmitter {
                 this._log.debug('processing queue drained, un-assigning');
 
                 try {
-                    this._consumer.unassign();
+                    this._consumer.incrementalUnassign(assignment);
                 } catch (e) {
                     // Ignore exceptions if we are not connected
                     const logger = this._consumer.isConnected() ? this._log.error : this._log.debug;