Skip to content

Leave the consumer group explicitly on shutdown #2819

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Draft
delthas wants to merge 3 commits into from
Draft

Leave the consumer group explicitly on shutdown #2819

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
dedd655
Extract the drain predicate into a method
delthas Aug 28, 2026
162170d
Hoist the functional waitFor helper to module scope
delthas Sep 1, 2026
22f3a61
Leave the consumer group explicitly on shutdown
delthas Aug 28, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
160 changes: 123 additions & 37 deletions lib/BackbeatConsumer.js
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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');

/**
Expand Down Expand Up @@ -178,6 +180,7 @@ class BackbeatConsumer extends EventEmitter {

// a deferred un-assign gives up when this no longer matches
this._rebalanceId = 0;
this._shuttingDown = false;

this._messagesConsumed = 0;
// this variable represents how many kafka messages have been
Expand Down Expand Up @@ -755,6 +758,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.
Expand Down Expand Up @@ -792,7 +811,9 @@ class BackbeatConsumer extends EventEmitter {
if (err.code === kafka.CODES.ERRORS.ERR__ASSIGN_PARTITIONS) {
this._log.info('rdkafka.assign', { assignment });

this._setDrain(null);
if (!this._shuttingDown) {
this._setDrain(null);
}

try {
this._consumer.assign(assignment);
Expand All @@ -812,17 +833,32 @@ class BackbeatConsumer extends EventEmitter {
ledger: this._offsetLedger.getProcessingCount(this._topic),
});

const isSuperseded = () => rebalanceId !== this._rebalanceId;
// close() owns the departure: it drains, then un-assigns, which
// is what answers this revoke. Releasing here would cut that drain
// short and strand the offsets it exists to commit.
if (this._shuttingDown) {
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 => {
Expand Down Expand Up @@ -883,18 +919,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;
}
Expand All @@ -905,7 +930,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);
}
});
Expand Down Expand Up @@ -1287,39 +1312,100 @@ class BackbeatConsumer extends EventEmitter {
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([
Comment thread
delthas marked this conversation as resolved.
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());
}

/**
* 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) {
Comment thread
delthas marked this conversation as resolved.
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;
}
}

module.exports = BackbeatConsumer;
1 change: 1 addition & 0 deletions lib/constants.js
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -26,6 +26,7 @@ const constants = {
DRAINED: 'drained',
TIMEOUT: 'timeout',
SUPERSEDED: 'superseded',
SHUTDOWN: 'shutdown',
},
statusReady: 'READY',
statusUndefined: 'UNDEFINED',
Expand Down
Loading
Loading