Leave the consumer group explicitly on shutdown

lib/BackbeatConsumer.js

@@ -28,6 +28,8 @@ const tracing = require('./tracing');
 const { startLinkedSpanFromKafkaEntry } = require('arsenal/build/lib/tracing').kafka;
 
 const CLIENT_ID = 'BackbeatConsumer';
+// the group has already been left by the time we disconnect
+const DISCONNECT_TIMEOUT_MS = 5000;
 const { withTopicPrefix } = require('./util/topic');
 
 /**
@@ -178,6 +180,8 @@ class BackbeatConsumer extends EventEmitter {
 
         // a deferred un-assign gives up when this no longer matches
         this._rebalanceId = 0;
+        this._shuttingDown = false;
+        this._closeCallbacks = null;
 
         this._messagesConsumed = 0;
         // this variable represents how many kafka messages have been
@@ -435,6 +439,14 @@ class BackbeatConsumer extends EventEmitter {
             this._tryConsumedTimeout = null;
         }
 
+        // the shutdown drains what is already in flight, so fetching more
+        // only delays the departure and strands the extra work. This also
+        // ends the self-rescheduling loop, which would otherwise keep
+        // consuming against a closed client for the life of the process.
+        if (this._shuttingDown) {
+            return undefined;
+        }
+
         // use non-flowing mode of consumption to add some flow
         // control: explicit consumption of messages is required,
         // needs explicit polling to get new messages
@@ -755,6 +767,22 @@ class BackbeatConsumer extends EventEmitter {
         }
     }
 
+    /**
+     * Both the processing queue and the offset ledger must be drained before
+     * releasing partitions. The queue covers in-flight worker invocations
+     * (committable: true path); the ledger covers entries still waiting for
+     * their deferred onEntryCommittable to land (committable: false path —
+     * typically a Kafka status producer's delivery callback). Releasing the
+     * partition before the ledger drains would cause offsetsStore to hit
+     * ERR__STATE on the deferred callbacks (BB-758).
+     *
+     * @returns {boolean} true if there is no work left in flight
+     */
+    _isFullyDrained() {
+        return this._processingQueue.idle() &&
+            this._offsetLedger.getProcessingCount(this._topic) === 0;
+    }
+
     /**
      * Run a shutdown/rebalance step that must not abort the sequence it
      * belongs to, logging rather than throwing.
@@ -792,6 +820,27 @@ class BackbeatConsumer extends EventEmitter {
         if (err.code === kafka.CODES.ERRORS.ERR__ASSIGN_PARTITIONS) {
             this._log.info('rdkafka.assign', { assignment });
 
+            // close() has already handed the partitions back. Accepting a
+            // fresh grant now leaves the client holding an assignment the
+            // disconnect has to revoke all over again, which wedges it and
+            // costs us the LeaveGroup the shutdown exists to send.
+            if (this._shuttingDown) {
+                // assign() is refused once disconnect() has started -- the
+                // binding gates it on isConnected(), which is already false --
+                // while unassign() is explicitly permitted while closing, and
+                // librdkafka coerces an assign into a full unassign during
+                // termination anyway. Leaving the callback unanswered is what
+                // parks the client in the rebalance.
+                this._bestEffort('assign.declined', () => {
+                    try {
+                        this._consumer.assign([]);
+                    } catch (e) { // eslint-disable-line no-unused-vars
+                        this._consumer.unassign();
+                    }
+                });
+                return;
+            }
+
             this._setDrain(null);
 
             try {
@@ -812,17 +861,38 @@ class BackbeatConsumer extends EventEmitter {
                 ledger: this._offsetLedger.getProcessingCount(this._topic),
             });
 
-            const isSuperseded = () => rebalanceId !== this._rebalanceId;
+            // librdkafka requires every rebalance callback to be answered,
+            // and leaving this one to close() is not enough: a revoke raised
+            // after close() has already un-assigned has nothing left to answer
+            // it, so the client stays in the rebalance and the disconnect
+            // wedges on it. Answering here does not cut the drain short --
+            // close() still waits for the in-flight work, the partitions are
+            // simply given back sooner.
+            if (this._shuttingDown) {
+                this._bestEffort('unassign.shutdown',
+                    () => this._consumer.unassign());
+                KafkaBacklogMetrics.onRebalance(
+                    this._topic, this._groupId, unassignStatus.SHUTDOWN);
+                return;
+            }
+
+            const isSuperseded = () =>
+                rebalanceId !== this._rebalanceId || this._shuttingDown;
             const skipSuperseded = status => {
-                this._log.info('skipping superseded un-assign', {
+                // close() answers the revoke itself once it has drained, so
+                // a shutdown is a distinct reason from a later rebalance
+                const reason = this._shuttingDown ?
+                    unassignStatus.SHUTDOWN : unassignStatus.SUPERSEDED;
+                this._log.info('skipping deferred un-assign', {
                     status,
+                    reason,
                     rebalanceId,
                     currentRebalanceId: this._rebalanceId,
                     topic: this._topic,
                     groupId: this._groupId,
                 });
                 KafkaBacklogMetrics.onRebalance(
-                    this._topic, this._groupId, unassignStatus.SUPERSEDED);
+                    this._topic, this._groupId, reason);
             };
 
             const unassign = jsutil.once(status => {
@@ -883,18 +953,7 @@ class BackbeatConsumer extends EventEmitter {
                 }
             });
 
-            // Both the processing queue and the offset ledger must be drained
-            // before unassigning. The queue covers in-flight worker invocations
-            // (committable: true path); the ledger covers entries still waiting
-            // for their deferred onEntryCommittable to land (committable: false
-            // path — typically a Kafka status producer's delivery callback).
-            // Releasing the partition before the ledger drains would cause
-            // offsetsStore to hit ERR__STATE on the deferred callbacks (BB-758).
-            const isFullyDrained = () =>
-                this._processingQueue.idle() &&
-                this._offsetLedger.getProcessingCount(this._topic) === 0;
-
-            if (isFullyDrained()) {
+            if (this._isFullyDrained()) {
                 unassign(unassignStatus.IDLE);
                 return;
             }
@@ -905,7 +964,7 @@ class BackbeatConsumer extends EventEmitter {
             // checkFullyDrained re-checks both conditions and only
             // triggers unassign once both hold.
             this._setDrain(() => {
-                if (isFullyDrained()) {
+                if (this._isFullyDrained()) {
                     unassign(unassignStatus.DRAINED);
                 }
             });
@@ -1283,42 +1342,113 @@ class BackbeatConsumer extends EventEmitter {
      * @return {undefined}
      */
     close(cb) {
+        if (this._closeCallbacks) {
+            this._closeCallbacks.push(cb);
+            return undefined;
+        }
+        this._closeCallbacks = [cb];
+
         if (this._publishOffsetsCronTimer) {
             clearInterval(this._publishOffsetsCronTimer);
             this._publishOffsetsCronTimer = null;
         }
-        if (this._publishOffsetsCronActive) {
-            return setTimeout(() => this.close(cb), 1000);
-        }
+        // its watchdog would otherwise disconnect us mid-departure
+        clearTimeout(this._drainProcessQueueTimeout);
+        this._drainProcessQueueTimeout = null;
+        this._shuttingDown = true;
         this._circuitBreaker.stop();
+
         return async.waterfall([
             next => {
-                if (this._consumer?.isConnected()) {
-                    const subscription = this._getSubscription();
-                    if (subscription !== null) {
-                        this._consumer.unsubscribe();
-                        // Wait for partition unassign to complete before
-                        // disconnecting, the rebalance callback will handle
-                        // waiting for current jobs to complete as well as commit
-                        // the latest offsets
-                        this.once('unassign', () => next());
-                        return;
-                    }
+                // draining buys a commit before the partitions go, which is
+                // pointless once the client is gone
+                if (!this._consumer?.isConnected()) {
+                    return process.nextTick(next);
                 }
-                process.nextTick(next);
+                return this._drainBeforeShutdown(next);
             },
             next => {
                 if (this._zookeeper) {
                     this._zookeeper.close();
                 }
-                if (this._consumer?.isConnected()) {
-                    this._consumer.disconnect();
-                    this._consumer.once('disconnected', () => next());
-                } else {
-                    process.nextTick(next);
+                if (!this._consumer?.isConnected()) {
+                    return process.nextTick(next);
                 }
+                // commit first: un-assigning resets the stored offsets.
+                // unsubscribe next, which leaves the group protocol waiting on
+                // an un-assign, and that un-assign is what sends the
+                // LeaveGroup — before disconnect(), and without depending on
+                // a revoke callback reaching us mid-close.
+                this._bestEffort('commit', () => this._consumer.commit());
+                this._bestEffort('unsubscribe', () => this._consumer.unsubscribe());
+                this._bestEffort('unassign', () => this._consumer.unassign());
+
+                const disconnected = jsutil.once(next);
+                const timer = setTimeout(() => {
+                    this._log.warn('consumer did not finish disconnecting, ' +
+                        'exiting anyway', {
+                        timeoutMs: DISCONNECT_TIMEOUT_MS,
+                        topic: this._topic,
+                        groupId: this._groupId,
+                    });
+                    disconnected();
+                }, DISCONNECT_TIMEOUT_MS);
+                this._consumer.once('disconnected', () => {
+                    clearTimeout(timer);
+                    disconnected();
+                });
+                this._consumer.disconnect();
+                return undefined;
             },
-        ], () => cb());
+        ], () => {
+            const callbacks = this._closeCallbacks;
+            this._closeCallbacks = null;
+            callbacks.forEach(done => done());
+        });
+    }
+
+    /**
+     * Wait for in-flight work so its offsets are committed before the
+     * partitions go, bounded like the revoke path so a wedged task cannot
+     * hold the departure any longer than it already did.
+     *
+     * @param {function} cb - callback
+     * @returns {undefined}
+     */
+    _drainBeforeShutdown(cb) {
+        const done = jsutil.once(cb);
+        if (this._isFullyDrained()) {
+            return process.nextTick(done);
+        }
+        this._log.info('waiting for in-flight work before leaving the group', {
+            queueLen: this._processingQueue.length(),
+            running: this._processingQueue.running(),
+            ledger: this._offsetLedger.getProcessingCount(this._topic),
+            topic: this._topic,
+            groupId: this._groupId,
+        });
+        // same bound the revoke path uses, so a wedged task delays the
+        // departure no longer than it did before
+        const timer = setTimeout(() => {
+            this._log.warn('giving up on in-flight work, leaving the group', {
+                queueLen: this._processingQueue.length(),
+                running: this._processingQueue.running(),
+                ledger: this._offsetLedger.getProcessingCount(this._topic),
+                topic: this._topic,
+                groupId: this._groupId,
+            });
+            this._setDrain(null);
+            done();
+        }, this._maxPollIntervalMs - 1000);
+
+        this._setDrain(() => {
+            if (this._isFullyDrained()) {
+                clearTimeout(timer);
+                this._setDrain(null);
+                done();
+            }
+        });
+        return undefined;
     }
 }
 

lib/constants.js

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