From 5f857505a92a5f50603a106e21d9a03598b0f0f3 Mon Sep 17 00:00:00 2001 From: Anurag Mittal <1321012+anurag4DSB@users.noreply.github.com> Date: Fri, 14 Aug 2026 09:57:39 +0200 Subject: [PATCH 1/3] S3C-11127: add shared delivery key helper for bucket notification The delivery topic is keyed by destination, optionally spread over several keys, so that records of one object always land on the same partition. The populator and the cutover replay both build the key, hence the shared helper. --- extensions/notification/utils/deliveryKey.js | 25 ++++++++++++++++++++ 1 file changed, 25 insertions(+) create mode 100644 extensions/notification/utils/deliveryKey.js diff --git a/extensions/notification/utils/deliveryKey.js b/extensions/notification/utils/deliveryKey.js new file mode 100644 index 000000000..4ea9fe6e4 --- /dev/null +++ b/extensions/notification/utils/deliveryKey.js @@ -0,0 +1,25 @@ +const crypto = require('crypto'); + +/** + * Builds the delivery-topic record key for an addressed notification. + * The key is stable across processes and reruns: same destination and + * object always map to the same key, hence the same partition. + * + * @param {Object} destination - destination config entry + * @param {String} bucket - bucket name + * @param {String} objectKey - object key + * @return {String} record key + */ +function buildDeliveryKey(destination, bucket, objectKey) { + const m = destination.spreadFactor || 1; + if (m <= 1) { + return destination.resource; + } + const h = crypto.createHash('md5') + .update(`${bucket}/${objectKey}`) + .digest() + .readUInt32BE(0); + return `${destination.resource}|${h % m}`; +} + +module.exports = { buildDeliveryKey }; From f873f61f1fda4bbcdb7828df467c856494c312c5 Mon Sep 17 00:00:00 2001 From: Anurag Mittal <1321012+anurag4DSB@users.noreply.github.com> Date: Fri, 14 Aug 2026 09:57:47 +0200 Subject: [PATCH 2/3] S3C-11127: drain old notification topics into the delivery topic At cutover the old per-destination queue processors leave records behind in the internal topics. DeliveryTopicDrainer replays them as addressed records so the new delivery worker pool starts with nothing lost: it reads each old group's committed offsets, consumes from there up to the head captured at start, rematches every record against the bucket configuration to recover its configurationId, and produces it keyed like the populator would. It never writes a consumer group offset, so a failed run can be rerun and a rollback to the old pipeline still resumes where it stopped. Duplicates are the accepted cost of that: any error aborts the drain rather than stepping over a record. --- .../deliveryWorker/DeliveryTopicDrainer.js | 834 ++++++++++++++++++ .../notification/DeliveryTopicDrainer.spec.js | 638 ++++++++++++++ 2 files changed, 1472 insertions(+) create mode 100644 extensions/notification/deliveryWorker/DeliveryTopicDrainer.js create mode 100644 tests/unit/notification/DeliveryTopicDrainer.spec.js diff --git a/extensions/notification/deliveryWorker/DeliveryTopicDrainer.js b/extensions/notification/deliveryWorker/DeliveryTopicDrainer.js new file mode 100644 index 000000000..4aa6c1490 --- /dev/null +++ b/extensions/notification/deliveryWorker/DeliveryTopicDrainer.js @@ -0,0 +1,834 @@ +'use strict'; + +const async = require('async'); +const { KafkaConsumer } = require('node-rdkafka'); +const Logger = require('werelogs').Logger; +const { errors, jsutil } = require('arsenal'); + +const BackbeatProducer = require('../../../lib/BackbeatProducer'); +const { withTopicPrefix } = require('../../../lib/util/topic'); +const NotificationConfigManager = require('../NotificationConfigManager'); +const configUtil = require('../utils/config'); +const { buildDeliveryKey } = require('../utils/deliveryKey'); + +const CONNECT_TIMEOUT_MS = 30000; +const METADATA_TIMEOUT_MS = 10000; +const WATERMARK_TIMEOUT_MS = 10000; +const COMMITTED_TIMEOUT_MS = 10000; +const CONSUME_BATCH_SIZE = 100; +const EMPTY_BATCH_SLEEP_MS = 500; +const MAX_EMPTY_BATCHES = 120; +const MATCH_CONCURRENCY = 10; + +/** + * @class DeliveryTopicDrainer + * + * @classdesc Drains the internal notification topics of the old + * per-destination pipeline into the delivery topic consumed by the + * delivery worker pool, as addressed records (old payload plus + * destinationId and configurationId). + * + * It is meant to run once during cutover, with the queue populator and + * every queue processor stopped, so that the topics it reads are not + * moving. It reads the committed offsets of the old consumer groups but + * never writes them: a rerun starts from the same place and produces + * duplicates, and a rollback to the old pipeline resumes exactly where + * it stopped. Duplicates are acceptable on the delivery topic, gaps are + * not, so any error (an unreadable bucket configuration, a failed + * produce) aborts the drain instead of stepping over the record. + */ +class DeliveryTopicDrainer { + + /** + * @constructor + * @param {Object} params - params object + * @param {Object} params.kafkaConfig - kafka configuration object + * @param {String} params.kafkaConfig.hosts - list of kafka brokers + * @param {Object} [params.mongoConfig] - mongodb connection config, + * when absent the zookeeper config backend is used + * @param {Object} [params.zkConfig] - zookeeper configuration object + * @param {Object} params.notifConfig - notification configuration + * object, the same "extensions.notification" block the populator + * and the queue processors read + * @param {String} [params.replayId] - suffix of the throwaway + * consumer group, defaults to the process id + * @param {Object} [params.consumer] - node-rdkafka consumer, for tests + * @param {Object} [params.producer] - BackbeatProducer, for tests + * @param {Object} [params.bnConfigManager] - config manager, for tests + * @param {Number} [params.batchSize] - records per consume call + * @param {Number} [params.emptyBatchSleepMs] - pause before rechecking + * partition positions after an empty batch + * @param {Number} [params.maxEmptyBatches] - how many consecutive + * fruitless batches to accept before giving up on a partition + * @param {Number} [params.matchConcurrency] - how many records of a + * batch to match against bucket configurations in parallel + * @param {Object} [params.logger] - werelogs logger + */ + constructor(params) { + const { + kafkaConfig, mongoConfig, zkConfig, notifConfig, replayId, + consumer, producer, bnConfigManager, batchSize, + emptyBatchSleepMs, maxEmptyBatches, matchConcurrency, logger, + } = params; + this.kafkaConfig = kafkaConfig; + this.mongoConfig = mongoConfig; + this.zkConfig = zkConfig; + this.notifConfig = notifConfig; + this.replayId = replayId || String(process.pid); + this.groupId = `bn-replay-${this.replayId}`; + + this.bnConfigManager = bnConfigManager || null; + this._consumer = consumer || null; + this._producer = producer || null; + // injected clients belong to the caller, we neither connect nor + // close them + this._ownsConsumer = !consumer; + this._ownsProducer = !producer; + + this._batchSize = batchSize || CONSUME_BATCH_SIZE; + this._emptyBatchSleepMs = emptyBatchSleepMs === undefined ? + EMPTY_BATCH_SLEEP_MS : emptyBatchSleepMs; + this._maxEmptyBatches = maxEmptyBatches || MAX_EMPTY_BATCHES; + this._matchConcurrency = matchConcurrency || MATCH_CONCURRENCY; + + this.totals = { drained: 0, produced: 0, skipped: 0 }; + this.logger = logger || + new Logger('Backbeat:Notification:DeliveryTopicDrainer'); + } + + /** + * Resolves where a partition has to be drained from, out of the old + * consumer group offset and the partition watermarks. + * + * A group that never committed on the partition, or that committed + * below the low watermark because retention already dropped those + * records, restarts at the low watermark: that is the earliest thing + * we can still deliver. + * + * @param {Object} bounds - partition bounds + * @param {Number} [bounds.committedOffset] - offset committed by the + * old consumer group, negative or missing when there is none + * @param {Number} bounds.lowOffset - low watermark + * @param {Number} bounds.highOffset - high watermark, the offset the + * next record would get + * @return {Object} object with the resolved startOffset, a skip flag + * telling there is nothing to drain, and a reason to log + */ + static resolveStartOffset(bounds) { + const { committedOffset, lowOffset, highOffset } = bounds; + let startOffset = committedOffset; + let reason = 'resuming at old consumer group offset'; + if (typeof startOffset !== 'number' || + !Number.isFinite(startOffset) || startOffset < 0) { + startOffset = lowOffset; + reason = 'old consumer group has no committed offset'; + } else if (startOffset < lowOffset) { + startOffset = lowOffset; + reason = 'old consumer group offset is below the low watermark, ' + + 'records were already dropped by retention'; + } + if (startOffset >= highOffset) { + return { startOffset, skip: true, reason: 'nothing left to drain' }; + } + return { startOffset, skip: false, reason }; + } + + /** + * Initializes the config manager, the delivery topic producer and the + * consumer used to read the old topics + * + * @param {Function} done - callback + * @return {undefined} + */ + start(done) { + async.series([ + next => this._setupNotificationConfigManager(next), + next => this._setupProducer(next), + next => this._setupConsumer(next), + ], err => { + if (err) { + this.logger.error('error starting delivery topic drainer', { + method: 'DeliveryTopicDrainer.start', + error: err.message, + }); + return done(err); + } + return done(); + }); + } + + /** + * Initializes the NotificationConfigManager + * + * @param {Function} done - callback + * @return {undefined} + */ + _setupNotificationConfigManager(done) { + if (this.bnConfigManager) { + return process.nextTick(done); + } + try { + this.bnConfigManager = new NotificationConfigManager({ + mongoConfig: this.mongoConfig, + bucketMetastore: this.notifConfig.bucketMetastore, + maxCachedConfigs: this.notifConfig.maxCachedConfigs, + zkConfig: this.zkConfig, + zkPath: this.notifConfig.zookeeperPath, + zkConcurrency: this.notifConfig.zookeeperOpConcurrency, + logger: this.logger, + }); + return this.bnConfigManager.setup(done); + } catch (err) { + return done(err); + } + } + + /** + * Creates the single producer shared by every destination. It is + * bound to the delivery topic and applies the topic prefix itself. + * + * @param {Function} done - callback + * @return {undefined} + */ + _setupProducer(done) { + if (this._producer) { + return process.nextTick(done); + } + const topic = this._getDeliveryTopic(); + if (!topic) { + return process.nextTick(() => done(errors.InternalError + .customizeDescription( + 'missing extensions.notification.deliveryPool.topic'))); + } + const doneOnce = jsutil.once(done); + this._producer = new BackbeatProducer({ + kafka: { hosts: this.kafkaConfig.hosts }, + topic, + compressionType: this.kafkaConfig.compressionType, + requiredAcks: this.kafkaConfig.requiredAcks, + }); + this._producer.on('error', err => { + this.logger.error('error with delivery topic producer', { + method: 'DeliveryTopicDrainer._setupProducer', + error: err.message, + }); + doneOnce(err); + }); + this._producer.once('ready', () => doneOnce()); + return undefined; + } + + /** + * Creates the consumer that reads the old topics. It uses a throwaway + * group and has every form of offset writing disabled: the old groups + * must be left exactly as the queue processors left them. + * + * @param {Function} done - callback + * @return {undefined} + */ + _setupConsumer(done) { + if (this._consumer) { + return process.nextTick(done); + } + this._consumer = new KafkaConsumer({ + 'metadata.broker.list': this.kafkaConfig.hosts, + 'group.id': this.groupId, + 'enable.auto.commit': false, + 'enable.auto.offset.store': false, + 'allow.auto.create.topics': false, + 'metadata.max.age.ms': 5000, + }, {}); + this._consumer.on('event.error', err => + this.logger.error('rdkafka.error', { err })); + return this._consumer.connect({ timeout: CONNECT_TIMEOUT_MS }, err => { + if (err) { + this.logger.error('error connecting replay consumer', { + method: 'DeliveryTopicDrainer._setupConsumer', + groupId: this.groupId, + error: err.message, + }); + return done(err); + } + return done(); + }); + } + + _getDeliveryTopic() { + return this.notifConfig.deliveryPool && + this.notifConfig.deliveryPool.topic; + } + + /** + * Drains every configured destination, one after the other + * + * @param {Function} done - callback: done(err, totals) + * @return {undefined} + */ + run(done) { + const destinations = this.notifConfig.destinations || []; + this.logger.info('starting notification delivery replay', { + destinations: destinations.map(d => d.resource), + deliveryTopic: this._getDeliveryTopic(), + replayGroupId: this.groupId, + }); + return async.eachSeries(destinations, (destination, next) => + this.drainDestination(destination, next), err => { + if (err) { + this.logger.error('notification delivery replay failed', { + method: 'DeliveryTopicDrainer.run', + error: err.message, + totals: this.totals, + }); + return done(err); + } + this.logger.info('notification delivery replay complete', + this.totals); + return done(null, this.totals); + }); + } + + /** + * Drains the old topic of a single destination + * + * @param {Object} destination - destination config entry + * @param {Function} done - callback + * @return {undefined} + */ + drainDestination(destination, done) { + const groupIdPrefix = this.notifConfig.queueProcessor && + this.notifConfig.queueProcessor.groupId; + if (!groupIdPrefix) { + return done(errors.InternalError.customizeDescription( + 'missing extensions.notification.queueProcessor.groupId')); + } + const internalTopic = destination.internalTopic || + this.notifConfig.topic; + if (!internalTopic) { + return done(errors.InternalError.customizeDescription( + `no internal topic configured for ${destination.resource}`)); + } + // raw rdkafka consumption does not apply the topic prefix, unlike + // BackbeatConsumer, so it is applied here + const oldTopic = withTopicPrefix(internalTopic); + const oldGroup = `${groupIdPrefix}-${destination.resource}`; + return async.waterfall([ + next => this._getPartitions(oldTopic, next), + (partitions, next) => + this._getDrainPlan(oldTopic, oldGroup, partitions, next), + (plan, next) => + this._drainPartitions(destination, oldTopic, oldGroup, plan, + next), + ], done); + } + + /** + * Lists the partitions of an old topic + * + * @param {String} oldTopic - prefixed topic name + * @param {Function} done - callback: done(err, partitions) + * @return {undefined} + */ + _getPartitions(oldTopic, done) { + return this._consumer.getMetadata({ + topic: oldTopic, + timeout: METADATA_TIMEOUT_MS, + }, (err, metadata) => { + if (err) { + this.logger.error('error getting metadata for old topic', { + method: 'DeliveryTopicDrainer._getPartitions', + oldTopic, + errorCode: err, + }); + return done(errors.InternalError); + } + const topicMd = metadata.topics.find(t => t.name === oldTopic); + if (!topicMd || topicMd.partitions.length === 0) { + this.logger.info('old topic has no partitions, nothing to ' + + 'drain', { oldTopic }); + return done(null, []); + } + return done(null, topicMd.partitions.map(p => p.id)); + }); + } + + /** + * Builds the per partition drain plan: where to start, and where the + * head was when the drain started. The head is captured once so that + * the drain has a fixed target and terminates. + * + * @param {String} oldTopic - prefixed topic name + * @param {String} oldGroup - old consumer group id + * @param {Number[]} partitions - partition ids + * @param {Function} done - callback: done(err, plan) + * @return {undefined} + */ + _getDrainPlan(oldTopic, oldGroup, partitions, done) { + if (partitions.length === 0) { + return process.nextTick(() => done(null, [])); + } + return async.waterfall([ + next => this._readCommittedOffsets(oldTopic, oldGroup, partitions, + next), + (committedOffsets, next) => async.mapSeries(partitions, + (partition, partitionDone) => this._planPartition(oldTopic, + partition, committedOffsets[partition], partitionDone), + next), + ], (err, plan) => { + if (err) { + return done(err); + } + plan.forEach(entry => this._logPlanEntry(oldTopic, oldGroup, entry)); + return done(null, plan.filter(entry => !entry.skip)); + }); + } + + /** + * Captures the watermarks of one partition once, and resolves where + * the drain starts from + * + * @param {String} oldTopic - prefixed topic name + * @param {Number} partition - partition id + * @param {Number} [committedOffset] - old consumer group offset + * @param {Function} done - callback: done(err, planEntry) + * @return {undefined} + */ + _planPartition(oldTopic, partition, committedOffset, done) { + return this._consumer.queryWatermarkOffsets(oldTopic, partition, + WATERMARK_TIMEOUT_MS, (err, offsets) => { + if (err) { + this.logger.error('error getting watermark offsets', { + method: 'DeliveryTopicDrainer._planPartition', + oldTopic, + partition, + errorCode: err, + }); + return done(errors.InternalError); + } + const { lowOffset, highOffset } = offsets; + const resolved = DeliveryTopicDrainer.resolveStartOffset({ + committedOffset, + lowOffset, + highOffset, + }); + return done(null, Object.assign({ + partition, + committedOffset, + lowOffset, + headOffset: highOffset, + }, resolved)); + }); + } + + _logPlanEntry(oldTopic, oldGroup, entry) { + const info = { + oldTopic, + oldGroup, + partition: entry.partition, + committedOffset: entry.committedOffset, + lowOffset: entry.lowOffset, + headOffset: entry.headOffset, + startOffset: entry.startOffset, + reason: entry.reason, + }; + if (typeof entry.committedOffset === 'number' && + entry.committedOffset >= 0 && + entry.committedOffset < entry.lowOffset) { + this.logger.warn('records were lost before the replay', info); + } else { + this.logger.info('resolved replay start offset', info); + } + } + + /** + * Reads the offsets committed by the old consumer group. + * + * In node-rdkafka 2.18 committed() is scoped to the group.id of the + * client it is called on, so a client configured with the old group id + * is the only way to see those offsets. That client only ever calls + * committed(): it never subscribes, assigns or commits, so the old + * group keeps its offsets and a rollback stays possible. + * + * @param {String} oldTopic - prefixed topic name + * @param {String} oldGroup - old consumer group id + * @param {Number[]} partitions - partition ids + * @param {Function} done - callback: done(err, offsetsByPartition) + * @return {undefined} + */ + _readCommittedOffsets(oldTopic, oldGroup, partitions, done) { + const toppars = partitions.map(partition => ({ + topic: oldTopic, + partition, + })); + return this._withOffsetReader(oldGroup, (reader, next) => + reader.committed(toppars, COMMITTED_TIMEOUT_MS, next), + (err, committedToppars) => { + if (err) { + this.logger.error('error reading committed offsets of ' + + 'the old consumer group', { + method: 'DeliveryTopicDrainer._readCommittedOffsets', + oldTopic, + oldGroup, + error: err.message, + }); + return done(errors.InternalError.customizeDescription( + err.message)); + } + const offsets = {}; + (committedToppars || []).forEach(tp => { + offsets[tp.partition] = tp.offset; + }); + return done(null, offsets); + }); + } + + /** + * Runs fn against a consumer bound to the given group id. When a + * consumer was injected it is reused as is, otherwise a short lived + * one is connected and disconnected around the call. + * + * @param {String} groupId - consumer group id to read offsets of + * @param {Function} fn - fn(reader, cb) + * @param {Function} done - callback: done(err, result) + * @return {undefined} + */ + _withOffsetReader(groupId, fn, done) { + if (!this._ownsConsumer) { + return fn(this._consumer, done); + } + const reader = new KafkaConsumer({ + 'metadata.broker.list': this.kafkaConfig.hosts, + 'group.id': groupId, + 'enable.auto.commit': false, + 'enable.auto.offset.store': false, + }, {}); + return reader.connect({ timeout: CONNECT_TIMEOUT_MS }, connectErr => { + if (connectErr) { + return done(connectErr); + } + return fn(reader, (err, result) => + reader.disconnect(() => done(err, result))); + }); + } + + /** + * Assigns the partitions to drain at their start offsets and consumes + * them until each one reaches the head captured in the plan + * + * @param {Object} destination - destination config entry + * @param {String} oldTopic - prefixed topic name + * @param {String} oldGroup - old consumer group id + * @param {Object[]} plan - per partition drain plan + * @param {Function} done - callback + * @return {undefined} + */ + _drainPartitions(destination, oldTopic, oldGroup, plan, done) { + if (plan.length === 0) { + this.logger.info('nothing to drain for destination', { + destination: destination.resource, + oldTopic, + oldGroup, + }); + return process.nextTick(done); + } + const state = new Map(); + plan.forEach(p => state.set(p.partition, { + partition: p.partition, + startOffset: p.startOffset, + headOffset: p.headOffset, + lastOffset: -1, + position: -1, + drained: 0, + produced: 0, + skipped: 0, + })); + this._consumer.assign(plan.map(p => ({ + topic: oldTopic, + partition: p.partition, + offset: p.startOffset, + }))); + let emptyBatches = 0; + const loop = () => { + const pending = this._pendingPartitions(state); + if (pending.length === 0) { + return this._finishDestination(destination, oldTopic, oldGroup, + state, done); + } + if (emptyBatches >= this._maxEmptyBatches) { + this.logger.error('gave up waiting for partitions to reach ' + + 'their head offset', { + method: 'DeliveryTopicDrainer._drainPartitions', + oldTopic, + oldGroup, + pending: pending.map(s => ({ + partition: s.partition, + lastOffset: s.lastOffset, + headOffset: s.headOffset, + })), + }); + return done(errors.InternalError.customizeDescription( + `replay stalled on topic ${oldTopic}`)); + } + return this._drainBatch(destination, oldTopic, state, + (err, consumed) => { + if (err) { + return done(err); + } + emptyBatches = consumed > 0 ? 0 : emptyBatches + 1; + return setImmediate(loop); + }); + }; + return loop(); + } + + _pendingPartitions(state) { + return [...state.values()].filter(s => { + if (s.lastOffset >= s.headOffset - 1) { + return false; + } + // position() is the offset the consumer would read next, it + // moves past records that consume() never returns + return !(s.position >= 0 && s.position >= s.headOffset); + }); + } + + _finishDestination(destination, oldTopic, oldGroup, state, done) { + // destinations sharing one internal topic are drained from + // different offsets, so drop the assignment and its fetch queue + // before the next one assigns the same partitions + this._consumer.unassign(); + state.forEach(s => { + this.logger.info('drained partition', { + oldTopic, + oldGroup, + partition: s.partition, + startOffset: s.startOffset, + headOffset: s.headOffset, + drained: s.drained, + produced: s.produced, + skipped: s.skipped, + }); + }); + this.logger.info('drained destination', { + destination: destination.resource, + oldTopic, + oldGroup, + partitions: state.size, + }); + return done(); + } + + /** + * Consumes one batch and produces every matching record of it to the + * delivery topic. The next batch is only consumed once the delivery + * reports of this one are in, which bounds memory and keeps the + * counters honest. + * + * @param {Object} destination - destination config entry + * @param {String} oldTopic - prefixed topic name + * @param {Map} state - per partition drain state + * @param {Function} done - callback: done(err, consumedCount) + * @return {undefined} + */ + _drainBatch(destination, oldTopic, state, done) { + return this._consumer.consume(this._batchSize, (err, records) => { + if (err) { + this.logger.error('error consuming from old topic', { + method: 'DeliveryTopicDrainer._drainBatch', + oldTopic, + errorCode: err, + }); + return done(errors.InternalError); + } + if (!records || records.length === 0) { + return this._refreshPositions(oldTopic, state, done); + } + return this._processBatch(destination, records, state, batchErr => { + if (batchErr) { + return done(batchErr); + } + return done(null, records.length); + }); + }); + } + + /** + * Waits a little then rereads the consumer positions, so that a + * partition whose remaining offsets yield no record still gets to the + * head instead of looping forever + * + * @param {String} oldTopic - prefixed topic name + * @param {Map} state - per partition drain state + * @param {Function} done - callback: done(err, consumedCount) + * @return {undefined} + */ + _refreshPositions(oldTopic, state, done) { + return setTimeout(() => { + let positions; + try { + positions = this._consumer.position() || []; + } catch (err) { + this.logger.error('error reading consumer positions', { + method: 'DeliveryTopicDrainer._refreshPositions', + oldTopic, + error: err.message, + }); + return done(errors.InternalError.customizeDescription( + err.message)); + } + positions.filter(p => p.topic === oldTopic).forEach(p => { + const partitionState = state.get(p.partition); + if (partitionState && typeof p.offset === 'number' && + p.offset >= 0) { + partitionState.position = p.offset; + } + }); + return done(null, 0); + }, this._emptyBatchSleepMs); + } + + _processBatch(destination, records, state, done) { + return async.mapLimit(records, this._matchConcurrency, + (record, next) => this._matchRecord(destination, record, next), + (err, messages) => { + if (err) { + return done(err); + } + const toProduce = []; + records.forEach((record, i) => { + const partitionState = state.get(record.partition); + if (partitionState) { + partitionState.drained++; + if (record.offset > partitionState.lastOffset) { + partitionState.lastOffset = record.offset; + } + } + this.totals.drained++; + if (messages[i]) { + toProduce.push(messages[i]); + this.totals.produced++; + if (partitionState) { + partitionState.produced++; + } + } else { + this.totals.skipped++; + if (partitionState) { + partitionState.skipped++; + } + } + }); + if (toProduce.length === 0) { + return process.nextTick(done); + } + return this._producer.send(toProduce, sendErr => { + if (sendErr) { + this.logger.error('error producing to delivery topic', { + method: 'DeliveryTopicDrainer._processBatch', + destination: destination.resource, + error: sendErr.message, + }); + return done(sendErr); + } + return done(); + }); + }); + } + + /** + * Rematches one old record against the bucket notification + * configuration, the way the queue processor of this destination + * would have, and turns it into an addressed delivery record. + * + * Old records carry no configurationId, it is only known once the + * matching rule is found again here. + * + * @param {Object} destination - destination config entry + * @param {Object} record - kafka record from the old topic + * @param {Function} done - callback: done(err, message), message being + * null when the record has nothing to deliver for this destination + * @return {undefined} + */ + _matchRecord(destination, record, done) { + let entry; + try { + entry = JSON.parse(record.value); + } catch (err) { + this.logger.error('error parsing JSON entry, skipping record', { + method: 'DeliveryTopicDrainer._matchRecord', + partition: record.partition, + offset: record.offset, + error: err.message, + }); + return process.nextTick(() => done(null, null)); + } + const { bucket, key } = entry; + return this.bnConfigManager.getConfig(bucket, (err, bnConfig) => { + if (err) { + // a record we cannot match is a record we cannot decide + // about, and skipping it would be a gap, so give up and + // let the operator rerun the replay + this.logger.error('error getting notification configuration', { + method: 'DeliveryTopicDrainer._matchRecord', + bucket, + key, + error: err.message, + }); + return done(err); + } + if (!bnConfig || Object.keys(bnConfig).length === 0 || + !bnConfig.notificationConfiguration) { + return done(null, null); + } + const queueConfig = + bnConfig.notificationConfiguration.queueConfig.filter( + c => c.queueArn.split(':').pop() === destination.resource); + if (!queueConfig.length) { + return done(null, null); + } + const destConfig = { + bucket, + notificationConfiguration: { queueConfig }, + }; + const { isValid, matchingConfig } = + configUtil.validateEntry(destConfig, entry); + if (!isValid) { + return done(null, null); + } + entry.destinationId = destination.resource; + entry.configurationId = matchingConfig.id; + return done(null, { + // the populator's publish() path url-encodes record keys, + // producing here goes straight through BackbeatProducer so + // the same encoding has to be applied, otherwise records of + // one object would land on two different partitions + key: encodeURIComponent( + buildDeliveryKey(destination, bucket, key)), + message: JSON.stringify(entry), + }); + }); + } + + /** + * Disconnects the clients this drainer created + * + * @param {Function} done - callback + * @return {undefined} + */ + stop(done) { + return async.series([ + next => { + if (!this._consumer || !this._ownsConsumer) { + return next(); + } + return this._consumer.disconnect(() => next()); + }, + next => { + if (!this._producer || !this._ownsProducer) { + return next(); + } + return this._producer.close(() => next()); + }, + ], () => done()); + } +} + +module.exports = DeliveryTopicDrainer; diff --git a/tests/unit/notification/DeliveryTopicDrainer.spec.js b/tests/unit/notification/DeliveryTopicDrainer.spec.js new file mode 100644 index 000000000..6525f098b --- /dev/null +++ b/tests/unit/notification/DeliveryTopicDrainer.spec.js @@ -0,0 +1,638 @@ +const assert = require('assert'); +const sinon = require('sinon'); + +const DeliveryTopicDrainer = require( + '../../../extensions/notification/deliveryWorker/DeliveryTopicDrainer'); +const { buildDeliveryKey } = require( + '../../../extensions/notification/utils/deliveryKey'); + +const OLD_TOPIC = 'bucket-notification'; +const DESTINATION = { resource: 'destId', type: 'kafka', host: 'external-host' }; + +const notifConfig = { + topic: OLD_TOPIC, + queueProcessor: { + groupId: 'backbeat-bucket-notification-group', + }, + deliveryPool: { + topic: 'bucket-notification-delivery', + }, + destinations: [DESTINATION], +}; + +const bucketConfig = { + bucket: 'mybucket', + notificationConfiguration: { + queueConfig: [ + { + id: 'config-1', + queueArn: 'arn:scality:bucketnotif:::destId', + events: ['s3:ObjectCreated:*'], + }, + ], + }, +}; + +function makeEntry(overrides) { + return Object.assign({ + bucket: 'mybucket', + key: 'obj1', + eventType: 's3:ObjectCreated:Put', + value: '{}', + }, overrides); +} + +function makeRecord(topic, partition, offset, entry) { + return { + topic, + partition, + offset, + key: Buffer.from(`${entry.bucket}/${entry.key}`), + value: Buffer.from(JSON.stringify(entry)), + }; +} + +/** + * Stubs the bits of a node-rdkafka consumer the drainer uses. Offset + * writing methods are spies so that tests can assert the drainer never + * touches the offsets of the old consumer groups. + * + * @param {Object} params - stub params + * @param {Object} params.watermarks - partition id to { lowOffset, highOffset } + * @param {Object} params.committed - partition id to committed offset + * @param {Array[]} params.batches - batches returned by consume(), in order + * @param {Object[]} [params.positions] - what position() returns + * @return {Object} consumer stub + */ +function makeConsumer(params) { + const { watermarks, committed, batches, positions } = params; + const consumer = { + consumeCalls: 0, + unassignCalls: 0, + assigned: null, + committedArgs: null, + onConsume: null, + commit: sinon.spy(), + commitSync: sinon.spy(), + commitMessage: sinon.spy(), + subscribe: sinon.spy(), + getMetadata: (opts, cb) => process.nextTick(() => cb(null, { + topics: [{ + name: opts.topic, + partitions: Object.keys(watermarks) + .map(id => ({ id: Number(id) })), + }], + })), + committed: (toppars, timeout, cb) => { + consumer.committedArgs = { toppars, timeout }; + return process.nextTick(() => cb(null, toppars.map(tp => ({ + topic: tp.topic, + partition: tp.partition, + offset: committed[tp.partition], + })))); + }, + queryWatermarkOffsets: (topic, partition, timeout, cb) => + process.nextTick(() => cb(null, watermarks[partition])), + assign: assignments => { + consumer.assigned = assignments; + }, + unassign: () => { + consumer.unassignCalls++; + }, + consume: (count, cb) => { + consumer.consumeCalls++; + if (consumer.onConsume) { + consumer.onConsume(); + } + const batch = batches.length > 0 ? batches.shift() : []; + return process.nextTick(() => cb(null, batch)); + }, + position: () => positions || [], + }; + return consumer; +} + +function makeProducer(sent, onSend) { + return { + send: (entries, cb) => { + sent.push(...entries); + if (onSend) { + return onSend(entries, cb); + } + return process.nextTick(cb); + }, + }; +} + +function makeDrainer(params) { + return new DeliveryTopicDrainer(Object.assign({ + kafkaConfig: { hosts: 'localhost:9092' }, + notifConfig, + emptyBatchSleepMs: 0, + batchSize: 10, + }, params)); +} + +describe('notification DeliveryTopicDrainer', () => { + describe('resolveStartOffset', () => { + it('should start at the low watermark when the old group never ' + + 'committed', () => { + // -1001 is what librdkafka reports for an unset offset + const res = DeliveryTopicDrainer.resolveStartOffset({ + committedOffset: -1001, + lowOffset: 4, + highOffset: 20, + }); + assert.strictEqual(res.startOffset, 4); + assert.strictEqual(res.skip, false); + }); + + it('should start at the low watermark when there is no committed ' + + 'offset at all', () => { + const res = DeliveryTopicDrainer.resolveStartOffset({ + committedOffset: undefined, + lowOffset: 0, + highOffset: 3, + }); + assert.strictEqual(res.startOffset, 0); + assert.strictEqual(res.skip, false); + }); + + it('should start at the low watermark when retention dropped the ' + + 'committed offset', () => { + const res = DeliveryTopicDrainer.resolveStartOffset({ + committedOffset: 3, + lowOffset: 10, + highOffset: 20, + }); + assert.strictEqual(res.startOffset, 10); + assert.strictEqual(res.skip, false); + }); + + it('should resume at the committed offset', () => { + const res = DeliveryTopicDrainer.resolveStartOffset({ + committedOffset: 7, + lowOffset: 0, + highOffset: 12, + }); + assert.strictEqual(res.startOffset, 7); + assert.strictEqual(res.skip, false); + }); + + it('should skip a partition the old group fully consumed', () => { + const res = DeliveryTopicDrainer.resolveStartOffset({ + committedOffset: 5, + lowOffset: 0, + highOffset: 5, + }); + assert.strictEqual(res.skip, true); + }); + + it('should skip an empty partition', () => { + const res = DeliveryTopicDrainer.resolveStartOffset({ + committedOffset: -1001, + lowOffset: 0, + highOffset: 0, + }); + assert.strictEqual(res.skip, true); + }); + }); + + describe('drainDestination', () => { + it('should read the committed offsets of the old destination group ' + + 'without ever writing them', done => { + const consumer = makeConsumer({ + watermarks: { 0: { lowOffset: 0, highOffset: 1 } }, + committed: { 0: 0 }, + batches: [[makeRecord(OLD_TOPIC, 0, 0, makeEntry())]], + }); + const drainer = makeDrainer({ + consumer, + producer: makeProducer([]), + bnConfigManager: { getConfig: (b, cb) => cb(null, bucketConfig) }, + }); + const readOffsets = sinon.spy(drainer, '_readCommittedOffsets'); + drainer.drainDestination(DESTINATION, err => { + assert.ifError(err); + assert(readOffsets.calledOnce); + assert.strictEqual(readOffsets.args[0][0], OLD_TOPIC); + assert.strictEqual(readOffsets.args[0][1], + 'backbeat-bucket-notification-group-destId'); + assert.deepStrictEqual(consumer.committedArgs.toppars, + [{ topic: OLD_TOPIC, partition: 0 }]); + assert(consumer.commit.notCalled); + assert(consumer.commitSync.notCalled); + assert(consumer.commitMessage.notCalled); + assert(consumer.subscribe.notCalled); + done(); + }); + }); + + it('should assign the partitions left to drain at their start ' + + 'offsets', done => { + const consumer = makeConsumer({ + watermarks: { + 0: { lowOffset: 0, highOffset: 3 }, + 1: { lowOffset: 8, highOffset: 20 }, + 2: { lowOffset: 0, highOffset: 5 }, + }, + // partition 1 committed below the low watermark, partition 2 + // was fully consumed by the old queue processor + committed: { 0: 1, 1: 2, 2: 5 }, + batches: [[ + makeRecord(OLD_TOPIC, 0, 2, makeEntry()), + makeRecord(OLD_TOPIC, 1, 19, makeEntry()), + ]], + }); + const drainer = makeDrainer({ + consumer, + producer: makeProducer([]), + bnConfigManager: { getConfig: (b, cb) => cb(null, bucketConfig) }, + }); + drainer.drainDestination(DESTINATION, err => { + assert.ifError(err); + assert.deepStrictEqual(consumer.assigned, [ + { topic: OLD_TOPIC, partition: 0, offset: 1 }, + { topic: OLD_TOPIC, partition: 1, offset: 8 }, + ]); + done(); + }); + }); + + it('should address matching records and key them with the encoded ' + + 'delivery key', done => { + const sent = []; + const consumer = makeConsumer({ + watermarks: { 0: { lowOffset: 0, highOffset: 1 } }, + committed: { 0: 0 }, + batches: [[makeRecord(OLD_TOPIC, 0, 0, makeEntry())]], + }); + const drainer = makeDrainer({ + consumer, + producer: makeProducer(sent), + bnConfigManager: { getConfig: (b, cb) => cb(null, bucketConfig) }, + }); + drainer.drainDestination(DESTINATION, err => { + assert.ifError(err); + assert.strictEqual(sent.length, 1); + assert.strictEqual(sent[0].key, encodeURIComponent( + buildDeliveryKey(DESTINATION, 'mybucket', 'obj1'))); + assert.strictEqual(sent[0].key, 'destId'); + const message = JSON.parse(sent[0].message); + assert.strictEqual(message.destinationId, 'destId'); + assert.strictEqual(message.configurationId, 'config-1'); + // the old payload is carried over untouched + assert.strictEqual(message.bucket, 'mybucket'); + assert.strictEqual(message.key, 'obj1'); + assert.strictEqual(message.eventType, 's3:ObjectCreated:Put'); + assert.deepStrictEqual(drainer.totals, + { drained: 1, produced: 1, skipped: 0 }); + done(); + }); + }); + + it('should url-encode a spread delivery key', done => { + const spreadDestination = Object.assign({ spreadFactor: 4 }, + DESTINATION); + const sent = []; + const consumer = makeConsumer({ + watermarks: { 0: { lowOffset: 0, highOffset: 1 } }, + committed: { 0: 0 }, + batches: [[makeRecord(OLD_TOPIC, 0, 0, makeEntry())]], + }); + const drainer = makeDrainer({ + consumer, + producer: makeProducer(sent), + bnConfigManager: { getConfig: (b, cb) => cb(null, bucketConfig) }, + }); + drainer.drainDestination(spreadDestination, err => { + assert.ifError(err); + const rawKey = buildDeliveryKey(spreadDestination, 'mybucket', + 'obj1'); + assert(rawKey.includes('|')); + assert.strictEqual(sent[0].key, encodeURIComponent(rawKey)); + assert(sent[0].key.includes('%7C')); + done(); + }); + }); + + it('should skip records that no longer match a configuration', done => { + const sent = []; + const entries = [ + // no configuration at all for this bucket + makeEntry({ bucket: 'noconfig' }), + // configuration targets another destination + makeEntry({ bucket: 'otherdest' }), + // event type is not subscribed to + makeEntry({ eventType: 's3:ObjectRemoved:Delete' }), + // deletion placeholder, no event type + makeEntry({ eventType: undefined }), + makeEntry({ key: 'obj5' }), + ]; + const configs = { + mybucket: bucketConfig, + noconfig: undefined, + otherdest: { + bucket: 'otherdest', + notificationConfiguration: { + queueConfig: [{ + id: 'other', + queueArn: 'arn:scality:bucketnotif:::otherDestId', + events: ['s3:ObjectCreated:*'], + }], + }, + }, + }; + const consumer = makeConsumer({ + watermarks: { 0: { lowOffset: 0, highOffset: 5 } }, + committed: { 0: 0 }, + batches: [entries.map((entry, i) => + makeRecord(OLD_TOPIC, 0, i, entry))], + }); + const drainer = makeDrainer({ + consumer, + producer: makeProducer(sent), + bnConfigManager: { + getConfig: (bucket, cb) => cb(null, configs[bucket]), + }, + }); + drainer.drainDestination(DESTINATION, err => { + assert.ifError(err); + assert.strictEqual(sent.length, 1); + assert.strictEqual(JSON.parse(sent[0].message).key, 'obj5'); + assert.deepStrictEqual(drainer.totals, + { drained: 5, produced: 1, skipped: 4 }); + done(); + }); + }); + + it('should skip a record whose payload is not valid JSON', done => { + const sent = []; + const record = makeRecord(OLD_TOPIC, 0, 0, makeEntry()); + record.value = Buffer.from('{not json'); + const consumer = makeConsumer({ + watermarks: { 0: { lowOffset: 0, highOffset: 1 } }, + committed: { 0: 0 }, + batches: [[record]], + }); + const drainer = makeDrainer({ + consumer, + producer: makeProducer(sent), + bnConfigManager: { getConfig: (b, cb) => cb(null, bucketConfig) }, + }); + drainer.drainDestination(DESTINATION, err => { + assert.ifError(err); + assert.strictEqual(sent.length, 0); + assert.strictEqual(drainer.totals.skipped, 1); + done(); + }); + }); + + it('should consume the next batch only once the previous one is ' + + 'acked', done => { + const events = []; + const consumer = makeConsumer({ + watermarks: { 0: { lowOffset: 0, highOffset: 3 } }, + committed: { 0: 0 }, + batches: [ + [ + makeRecord(OLD_TOPIC, 0, 0, makeEntry()), + makeRecord(OLD_TOPIC, 0, 1, makeEntry()), + ], + [makeRecord(OLD_TOPIC, 0, 2, makeEntry())], + ], + }); + consumer.onConsume = () => events.push('consume'); + const producer = makeProducer([], (entries, cb) => { + events.push('send'); + // a delivery report is not immediate, the drainer has to + // wait for it before pulling more records + return setTimeout(() => { + events.push('ack'); + cb(); + }, 10); + }); + const drainer = makeDrainer({ + consumer, + producer, + bnConfigManager: { getConfig: (b, cb) => cb(null, bucketConfig) }, + }); + drainer.drainDestination(DESTINATION, err => { + assert.ifError(err); + assert.deepStrictEqual(events, [ + 'consume', 'send', 'ack', + 'consume', 'send', 'ack', + ]); + done(); + }); + }); + + it('should stop at the head offset captured when the drain started', + done => { + const sent = []; + const consumer = makeConsumer({ + watermarks: { 0: { lowOffset: 0, highOffset: 2 } }, + committed: { 0: 0 }, + batches: [ + [ + makeRecord(OLD_TOPIC, 0, 0, makeEntry()), + makeRecord(OLD_TOPIC, 0, 1, makeEntry()), + ], + // would be returned if the drainer kept consuming + [makeRecord(OLD_TOPIC, 0, 2, makeEntry())], + ], + }); + const drainer = makeDrainer({ + consumer, + producer: makeProducer(sent), + bnConfigManager: { getConfig: (b, cb) => cb(null, bucketConfig) }, + }); + drainer.drainDestination(DESTINATION, err => { + assert.ifError(err); + assert.strictEqual(consumer.consumeCalls, 1); + assert.strictEqual(sent.length, 2); + done(); + }); + }); + + it('should finish a partition whose remaining offsets yield no ' + + 'record', done => { + const consumer = makeConsumer({ + watermarks: { 0: { lowOffset: 0, highOffset: 4 } }, + committed: { 0: 2 }, + batches: [], + positions: [{ topic: OLD_TOPIC, partition: 0, offset: 4 }], + }); + const drainer = makeDrainer({ + consumer, + producer: makeProducer([]), + bnConfigManager: { getConfig: (b, cb) => cb(null, bucketConfig) }, + }); + drainer.drainDestination(DESTINATION, err => { + assert.ifError(err); + assert.strictEqual(drainer.totals.drained, 0); + done(); + }); + }); + + it('should give up when a partition never reaches its head offset', + done => { + const consumer = makeConsumer({ + watermarks: { 0: { lowOffset: 0, highOffset: 4 } }, + committed: { 0: 0 }, + batches: [], + positions: [{ topic: OLD_TOPIC, partition: 0, offset: 1 }], + }); + const drainer = makeDrainer({ + consumer, + producer: makeProducer([]), + bnConfigManager: { getConfig: (b, cb) => cb(null, bucketConfig) }, + maxEmptyBatches: 3, + }); + drainer.drainDestination(DESTINATION, err => { + assert(err); + assert.strictEqual(consumer.consumeCalls, 3); + done(); + }); + }); + + it('should abort when a bucket configuration cannot be read', done => { + const consumer = makeConsumer({ + watermarks: { 0: { lowOffset: 0, highOffset: 1 } }, + committed: { 0: 0 }, + batches: [[makeRecord(OLD_TOPIC, 0, 0, makeEntry())]], + }); + const drainer = makeDrainer({ + consumer, + producer: makeProducer([]), + bnConfigManager: { + getConfig: (b, cb) => cb(new Error('mongo is down')), + }, + }); + drainer.drainDestination(DESTINATION, err => { + assert(err); + assert.strictEqual(err.message, 'mongo is down'); + done(); + }); + }); + + it('should abort when producing to the delivery topic fails', done => { + const consumer = makeConsumer({ + watermarks: { 0: { lowOffset: 0, highOffset: 1 } }, + committed: { 0: 0 }, + batches: [[makeRecord(OLD_TOPIC, 0, 0, makeEntry())]], + }); + const producer = makeProducer([], (entries, cb) => + process.nextTick(() => cb(new Error('delivery report error')))); + const drainer = makeDrainer({ + consumer, + producer, + bnConfigManager: { getConfig: (b, cb) => cb(null, bucketConfig) }, + }); + drainer.drainDestination(DESTINATION, err => { + assert(err); + assert.strictEqual(err.message, 'delivery report error'); + done(); + }); + }); + + it('should read the old topic through its prefixed name', done => { + process.env.KAFKA_TOPIC_PREFIX = 'pfx-'; + const prefixedTopic = `pfx-${OLD_TOPIC}`; + const consumer = makeConsumer({ + watermarks: { 0: { lowOffset: 0, highOffset: 1 } }, + committed: { 0: 0 }, + batches: [[makeRecord(prefixedTopic, 0, 0, makeEntry())]], + }); + const drainer = makeDrainer({ + consumer, + producer: makeProducer([]), + bnConfigManager: { getConfig: (b, cb) => cb(null, bucketConfig) }, + }); + drainer.drainDestination(DESTINATION, err => { + delete process.env.KAFKA_TOPIC_PREFIX; + assert.ifError(err); + assert.strictEqual(consumer.committedArgs.toppars[0].topic, + prefixedTopic); + assert.strictEqual(consumer.assigned[0].topic, prefixedTopic); + done(); + }); + }); + + it('should use the destination internal topic when it has one', + done => { + const destination = Object.assign({ internalTopic: 'dest-topic' }, + DESTINATION); + const consumer = makeConsumer({ + watermarks: { 0: { lowOffset: 0, highOffset: 1 } }, + committed: { 0: 0 }, + batches: [[makeRecord('dest-topic', 0, 0, makeEntry())]], + }); + const drainer = makeDrainer({ + consumer, + producer: makeProducer([]), + bnConfigManager: { getConfig: (b, cb) => cb(null, bucketConfig) }, + }); + drainer.drainDestination(destination, err => { + assert.ifError(err); + assert.strictEqual(consumer.assigned[0].topic, 'dest-topic'); + done(); + }); + }); + + it('should not assign anything when the old topic does not exist', + done => { + const consumer = makeConsumer({ + watermarks: {}, + committed: {}, + batches: [], + }); + const drainer = makeDrainer({ + consumer, + producer: makeProducer([]), + bnConfigManager: { getConfig: (b, cb) => cb(null, bucketConfig) }, + }); + drainer.drainDestination(DESTINATION, err => { + assert.ifError(err); + assert.strictEqual(consumer.assigned, null); + assert.strictEqual(consumer.consumeCalls, 0); + done(); + }); + }); + }); + + describe('run', () => { + it('should drain every destination and report the totals', done => { + const destinations = [ + { resource: 'destId', type: 'kafka', host: 'h' }, + { resource: 'otherDestId', type: 'kafka', host: 'h', + internalTopic: 'other-topic' }, + ]; + const sent = []; + const consumer = makeConsumer({ + watermarks: { 0: { lowOffset: 0, highOffset: 1 } }, + committed: { 0: 0 }, + batches: [ + [makeRecord(OLD_TOPIC, 0, 0, makeEntry())], + [makeRecord('other-topic', 0, 0, makeEntry())], + ], + }); + const drainer = makeDrainer({ + consumer, + producer: makeProducer(sent), + bnConfigManager: { getConfig: (b, cb) => cb(null, bucketConfig) }, + notifConfig: Object.assign({}, notifConfig, { destinations }), + }); + drainer.run((err, totals) => { + assert.ifError(err); + // the second destination is not in the bucket configuration + assert.deepStrictEqual(totals, + { drained: 2, produced: 1, skipped: 1 }); + assert.strictEqual(sent.length, 1); + // each destination drops its assignment when it is done + assert.strictEqual(consumer.unassignCalls, 2); + done(); + }); + }); + }); +}); From 2f0cf619523cace304459d5c57b5e36dc4cbc116 Mon Sep 17 00:00:00 2001 From: Anurag Mittal <1321012+anurag4DSB@users.noreply.github.com> Date: Fri, 14 Aug 2026 09:57:53 +0200 Subject: [PATCH 3/3] S3C-11127: add notification delivery replay entrypoint Thin wrapper around DeliveryTopicDrainer, run once during cutover with the populator and the queue processors stopped. Exits 0 on a full drain, non-zero otherwise, so the cutover can gate on it. --- bin/notificationDeliveryReplay.js | 65 +++++++++++++++++++++++++++++++ 1 file changed, 65 insertions(+) create mode 100644 bin/notificationDeliveryReplay.js diff --git a/bin/notificationDeliveryReplay.js b/bin/notificationDeliveryReplay.js new file mode 100644 index 000000000..6e73f6178 --- /dev/null +++ b/bin/notificationDeliveryReplay.js @@ -0,0 +1,65 @@ +'use strict'; + +/** + * Cutover replay: drains whatever the old per-destination queue + * processors left undelivered in the internal notification topics into + * the delivery topic, as addressed records. + * + * Run it once, with the queue populator and every notification queue + * processor stopped, before starting the delivery worker pool. It exits + * 0 once every destination is drained, non-zero otherwise. It writes no + * consumer group offset, so a failed run can simply be rerun, and going + * back to the old pipeline stays possible. + * + * Usage: node bin/notificationDeliveryReplay.js [replayId] + * The optional replayId only names the throwaway consumer group. + */ + +const async = require('async'); +const werelogs = require('werelogs'); + +const config = require('../lib/Config'); +const kafkaConfig = config.kafka; +const notifConfig = config.extensions.notification; +const mongoConfig = config.queuePopulator.mongo; +const zkConfig = config.zookeeper; +const DeliveryTopicDrainer = + require('../extensions/notification/deliveryWorker/DeliveryTopicDrainer'); + +const log = new werelogs.Logger('Backbeat:Notification:DeliveryReplay'); + +werelogs.configure({ + level: config.log.logLevel, + dump: config.log.dumpLevel, +}); + +const replayId = process.argv[2] || String(Date.now()); + +const drainer = new DeliveryTopicDrainer({ + kafkaConfig, + mongoConfig, + zkConfig, + notifConfig, + replayId, + logger: log, +}); + +async.waterfall([ + next => drainer.start(err => next(err)), + next => drainer.run(next), +], (err, totals) => { + if (err) { + log.error('notification delivery replay failed', { + method: 'notificationDeliveryReplay', + error: err.message, + }); + return drainer.stop(() => process.exit(1)); + } + log.info('notification delivery replay done', totals); + return drainer.stop(() => process.exit(0)); +}); + +process.on('SIGTERM', () => { + log.info('received SIGTERM, exiting'); + drainer.stop(() => process.exit(1)); +});