Abort the rebalance when a grant supersedes a parked revoke

lib/BackbeatConsumer.js

@@ -206,6 +206,10 @@ class BackbeatConsumer extends EventEmitter {
         this._disconnecting = false;
         // True between receiving a revoke and answering it (draining)
         this._waitingDrain = false;
+        // Set when a grant lands on a revoke we have not answered yet:
+        // the group has moved on without us, so we stop consuming and
+        // leave once the pending drain has committed
+        this._abortingRebalance = false;
         // Bounds close()'s wait for the drain
         this._closeDrainTimeout = null;
 
@@ -445,10 +449,10 @@ class BackbeatConsumer extends EventEmitter {
             this._tryConsumedTimeout = null;
         }
 
-        // stop fetching once closing: new entries would refill the
-        // pipeline close() is draining, and this ends the
+        // stop fetching once closing or aborting: new entries would
+        // refill the pipeline being drained, and this ends the
         // self-rescheduling consume loop that would poll a closed client
-        if (this._closing) {
+        if (this._closing || this._abortingRebalance) {
             return undefined;
         }
 
@@ -474,6 +478,20 @@ class BackbeatConsumer extends EventEmitter {
         this._nConsumePendingRequests += nNewConsumeRequests;
         return this._consumer.consume(nNewConsumeRequests, (err, entries) => {
             this._nConsumePendingRequests -= nNewConsumeRequests;
+            if (this._abortingRebalance) {
+                // A request outstanding when the abort started can still
+                // deliver, and the grant we accepted reset the fetch
+                // position to the last commit, so these are the entries
+                // being drained coming back around. Drop them: they are
+                // uncommitted, and whoever takes the partitions over
+                // reads them again from that same commit.
+                this._log.info('dropping entries consumed while aborting', {
+                    topic: this._topic,
+                    groupId: this._groupId,
+                    count: entries ? entries.length : 0,
+                });
+                return;
+            }
             if (!err) {
                 entries.forEach(entry => {
                     const { topic, partition, offset, key, timestamp } = entry;
@@ -772,6 +790,76 @@ class BackbeatConsumer extends EventEmitter {
         }
     }
 
+    /**
+     * Handle a grant that lands on a revoke we have not answered yet.
+     *
+     * librdkafka answers a subscribed-topic metadata change with an
+     * immediate rejoin, without waiting for the un-assign the revoke
+     * asked for. The generation granted here has already moved past the
+     * one the draining entries belong to, so carrying on would mean
+     * consuming those partitions again from their last commit while
+     * still finishing the previous generation's work: the same entries
+     * processed twice, concurrently for consumers that do not order by
+     * key. Treat it like the drain timeout instead. Stop consuming, let
+     * the pending drain commit what it has, then leave the group so the
+     * partitions are taken over and this consumer is restarted.
+     *
+     * The grant is accepted rather than declined because the draining
+     * entries can only store their offsets while the partitions are
+     * held.
+     *
+     * @param {TopicPartition[]} assignment - partitions being granted
+     * @returns {undefined}
+     */
+    _abortSupersededRebalance(assignment) {
+        // Keep answering grants while leaving, or the client stays parked
+        this._bestEffort('assign', () => this._consumer.assign(assignment));
+        if (this._abortingRebalance) {
+            return;
+        }
+        this._log.error('rdkafka.assign while a revoke is still draining: '
+            + 'the group moved on without us, aborting', {
+            topic: this._topic,
+            groupId: this._groupId,
+            assignment,
+            queueLen: this._processingQueue.length(),
+            running: this._processingQueue.running(),
+            ledger: this._offsetLedger.getProcessingCount(this._topic),
+        });
+        KafkaBacklogMetrics.onRebalance(this._topic, this._groupId,
+            unassignStatus.SUPERSEDED);
+        this._abortingRebalance = true;
+        // The pending drain commits and un-assigns as it always does;
+        // once it has, there is nothing left to lose by leaving.
+        this.once('unassign', () => this._leaveAfterAbort().catch(err => {
+            this._log.error('failed to leave after an aborted rebalance', {
+                topic: this._topic,
+                groupId: this._groupId,
+                error: err.message,
+            });
+        }));
+    }
+
+    /**
+     * Leave the group and stop being ready after an aborted rebalance,
+     * so the liveness probe (or supervisord on S3C) restarts us.
+     * @returns {Promise<undefined>} resolves once the client is closed
+     */
+    async _leaveAfterAbort() {
+        await this._close();
+        this._log.fatal('consumer left the queue after an aborted rebalance, '
+            + 'restart needed', {
+            topic: this._topic,
+            groupId: this._groupId,
+        });
+        if (process.env.CRASH_ON_REBALANCE_TIMEOUT === 'true') {
+            // On S3C nothing reacts to the healthcheck that fails once
+            // disconnected, so exit and let supervisord restart the
+            // program; grace period so the fatal line flushes first.
+            setTimeout(() => process.exit(1), 1000);
+        }
+    }
+
     /**
      * @param {kafka.KafkaError} err Rebalance event
      * @param {TopicPartition[]} assignment List of (un)assigned partitions
@@ -792,6 +880,10 @@ class BackbeatConsumer extends EventEmitter {
                 }
                 return;
             }
+            if (this._waitingDrain) {
+                this._abortSupersededRebalance(assignment);
+                return;
+            }
             this._log.info('rdkafka.assign', { assignment });
 
             try {

lib/constants.js

@@ -26,6 +26,7 @@ const constants = {
         DRAINED: 'drained',
         TIMEOUT: 'timeout',
         SHUTDOWN: 'shutdown',
+        SUPERSEDED: 'superseded',
     },
     statusReady: 'READY',
     statusUndefined: 'UNDEFINED',