From 5ea4c9590b600cb698a31725e9799148c7aff1c3 Mon Sep 17 00:00:00 2001 From: Anurag Mittal <1321012+anurag4DSB@users.noreply.github.com> Date: Fri, 14 Aug 2026 09:53:02 +0200 Subject: [PATCH 1/3] S3C-11127: add delivery key helper for addressed notifications The key maps a destination and an object to a stable record key, so the same object always lands on the same partition of the delivery topic. A destination with a spread factor above 1 is split over that many keys to let several delivery workers share it. --- extensions/notification/utils/deliveryKey.js | 25 +++++++++ tests/unit/notification/utils/deliveryKey.js | 55 ++++++++++++++++++++ 2 files changed, 80 insertions(+) create mode 100644 extensions/notification/utils/deliveryKey.js create mode 100644 tests/unit/notification/utils/deliveryKey.js diff --git a/extensions/notification/utils/deliveryKey.js b/extensions/notification/utils/deliveryKey.js new file mode 100644 index 0000000000..4ea9fe6e4b --- /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 }; diff --git a/tests/unit/notification/utils/deliveryKey.js b/tests/unit/notification/utils/deliveryKey.js new file mode 100644 index 0000000000..2437dd7041 --- /dev/null +++ b/tests/unit/notification/utils/deliveryKey.js @@ -0,0 +1,55 @@ +const assert = require('assert'); + +const { buildDeliveryKey } + = require('../../../../extensions/notification/utils/deliveryKey'); + +describe('deliveryKey ::', () => { + it('should use the destination resource when no spread factor is set', () => { + const key = buildDeliveryKey({ resource: 'destination1' }, + 'example-bucket', 'example-key'); + assert.strictEqual(key, 'destination1'); + }); + + it('should use the destination resource when the spread factor is 1', () => { + const key = buildDeliveryKey({ resource: 'destination1', spreadFactor: 1 }, + 'example-bucket', 'example-key'); + assert.strictEqual(key, 'destination1'); + }); + + it('should keep the key stable for the same bucket and object', () => { + const destination = { resource: 'destination1', spreadFactor: 4 }; + const key = buildDeliveryKey(destination, 'example-bucket', 'example-key'); + for (let i = 0; i < 10; i++) { + assert.strictEqual( + buildDeliveryKey(destination, 'example-bucket', 'example-key'), key); + } + }); + + it('should keep the spread index within the spread factor bounds', () => { + const destination = { resource: 'destination1', spreadFactor: 4 }; + for (let i = 0; i < 100; i++) { + const key = buildDeliveryKey(destination, 'example-bucket', `example-key-${i}`); + const [resource, index] = key.split('|'); + assert.strictEqual(resource, 'destination1'); + assert(Number.isInteger(Number(index))); + assert(Number(index) >= 0 && Number(index) < 4); + } + }); + + it('should spread the objects of a destination over several keys', () => { + const destination = { resource: 'destination1', spreadFactor: 4 }; + const keys = new Set(); + for (let i = 0; i < 100; i++) { + keys.add(buildDeliveryKey(destination, 'example-bucket', `example-key-${i}`)); + } + assert(keys.size > 1); + }); + + it('should give two destinations different keys for the same object', () => { + const first = buildDeliveryKey({ resource: 'destination1', spreadFactor: 4 }, + 'example-bucket', 'example-key'); + const second = buildDeliveryKey({ resource: 'destination2', spreadFactor: 4 }, + 'example-bucket', 'example-key'); + assert.notStrictEqual(first, second); + }); +}); From 94b3748bed794363a750267f8d14defc5b40d14e Mon Sep 17 00:00:00 2001 From: Anurag Mittal <1321012+anurag4DSB@users.noreply.github.com> Date: Fri, 14 Aug 2026 09:53:09 +0200 Subject: [PATCH 2/3] S3C-11127: validate the notification delivery pool configuration Adds the deliveryPool block, disabled by default, and the per-destination spreadFactor. topic and groupId are only required once the pool is enabled, so existing configurations keep validating unchanged. --- .../NotificationConfigValidator.js | 26 +++++ tests/config.notification.json | 6 + .../NotificationConfigValidator.js | 105 ++++++++++++++++++ 3 files changed, 137 insertions(+) diff --git a/extensions/notification/NotificationConfigValidator.js b/extensions/notification/NotificationConfigValidator.js index f6854522e6..67d2939ba9 100644 --- a/extensions/notification/NotificationConfigValidator.js +++ b/extensions/notification/NotificationConfigValidator.js @@ -77,6 +77,9 @@ const destinationSchema = joi.object({ then: joi.forbidden(), otherwise: joi.string().default('none'), }), + // number of record keys the destination is spread over: raise it to let + // more than one delivery worker handle the destination in parallel + spreadFactor: joi.number().integer().min(1).default(1), }); const joiSchema = joi.object({ @@ -89,6 +92,29 @@ const joiSchema = joi.object({ concurrency: joi.number().greater(0).default(1000), maxQueued: joi.number().greater(0).default(MAX_QUEUED_DEFAULT), }), + // single consumer group delivering to every destination, addressed by + // the record itself instead of by one topic per destination. + // deliveryTimeoutMs must stay above the producer request timeout (5000) + // and below kafka.maxPollIntervalMs minus a margin, otherwise a slow + // destination holds the partition past the poll deadline and the + // consumer is evicted. + deliveryPool: joi.object({ + enabled: joi.boolean().default(false), + topic: joi.string().when('enabled', { + is: joi.boolean().valid(true).required(), + then: joi.required(), + }), + groupId: joi.string().when('enabled', { + is: joi.boolean().valid(true).required(), + then: joi.required(), + }), + deliveryTimeoutMs: joi.number().min(6000).max(240000).default(30000), + producerIdleMs: joi.number().greater(0).default(300000), + maxProducers: joi.number().greater(0).default(50), + concurrency: joi.number().greater(0).default(1000), + maxQueued: joi.number().greater(0).default(MAX_QUEUED_DEFAULT), + probeServer: probeServerJoi.optional(), + }).optional(), destinations: joi.array().items(destinationSchema).default([]), // TODO: BB-625 reset to being required after supporting probeserver in S3C // for bucket notification proceses diff --git a/tests/config.notification.json b/tests/config.notification.json index 5fc51ab1db..36a99d7f32 100644 --- a/tests/config.notification.json +++ b/tests/config.notification.json @@ -35,6 +35,11 @@ "groupId": "backbeat-bucket-notification-group", "concurrency": 10 }, + "deliveryPool": { + "enabled": false, + "topic": "backbeat-bucket-notification-delivery", + "groupId": "backbeat-bucket-notification-delivery-group" + }, "destinations": [ { "resource": "destination1", @@ -43,6 +48,7 @@ "port": 9092, "topic": "destination-topic-1", "internalTopic": "internal-notification-topic-destination1", + "spreadFactor": 1, "auth": {} }, { diff --git a/tests/unit/notification/NotificationConfigValidator.js b/tests/unit/notification/NotificationConfigValidator.js index b65e5a64e1..2829d06343 100644 --- a/tests/unit/notification/NotificationConfigValidator.js +++ b/tests/unit/notification/NotificationConfigValidator.js @@ -476,3 +476,108 @@ describe('NotificationConfigValidator ::', () => { }) ); }); + +describe('NotificationConfigValidator delivery pool ::', () => { + const destinationConfig = { + resource: 'resource', + type: 'kafka', + host: 'host', + port: 8000, + topic: 'topic', + }; + + it('should default the destination spread factor to 1', () => { + const config = notificationConfigValidator(null, { + ...defaultExtConfig, + destinations: [destinationConfig], + }); + assert.strictEqual(config.destinations[0].spreadFactor, 1); + }); + + it('should reject a spread factor below 1', () => { + assert.throws(() => notificationConfigValidator(null, { + ...defaultExtConfig, + destinations: [{ ...destinationConfig, spreadFactor: 0 }], + })); + }); + + it('should reject a non integer spread factor', () => { + assert.throws(() => notificationConfigValidator(null, { + ...defaultExtConfig, + destinations: [{ ...destinationConfig, spreadFactor: 1.5 }], + })); + }); + + it('should leave the delivery pool unset when it is not configured', () => { + const config = notificationConfigValidator(null, defaultExtConfig); + assert.strictEqual(config.deliveryPool, undefined); + }); + + it('should apply the delivery pool defaults', () => { + const config = notificationConfigValidator(null, { + ...defaultExtConfig, + deliveryPool: {}, + }); + assert.strictEqual(config.deliveryPool.enabled, false); + assert.strictEqual(config.deliveryPool.deliveryTimeoutMs, 30000); + assert.strictEqual(config.deliveryPool.producerIdleMs, 300000); + assert.strictEqual(config.deliveryPool.maxProducers, 50); + assert.strictEqual(config.deliveryPool.concurrency, 1000); + assert.strictEqual(config.deliveryPool.maxQueued, 1000); + }); + + it('should accept an enabled delivery pool with a topic and a group id', () => { + assert.doesNotThrow(() => notificationConfigValidator(null, { + ...defaultExtConfig, + deliveryPool: { + enabled: true, + topic: 'delivery-topic', + groupId: 'delivery-group', + }, + })); + }); + + it('should require a topic when the delivery pool is enabled', () => { + assert.throws(() => notificationConfigValidator(null, { + ...defaultExtConfig, + deliveryPool: { + enabled: true, + groupId: 'delivery-group', + }, + })); + }); + + it('should require a group id when the delivery pool is enabled', () => { + assert.throws(() => notificationConfigValidator(null, { + ...defaultExtConfig, + deliveryPool: { + enabled: true, + topic: 'delivery-topic', + }, + })); + }); + + it('should reject a delivery timeout below the producer request timeout', () => { + assert.throws(() => notificationConfigValidator(null, { + ...defaultExtConfig, + deliveryPool: { + enabled: true, + topic: 'delivery-topic', + groupId: 'delivery-group', + deliveryTimeoutMs: 5000, + }, + })); + }); + + it('should reject a delivery timeout above the poll interval margin', () => { + assert.throws(() => notificationConfigValidator(null, { + ...defaultExtConfig, + deliveryPool: { + enabled: true, + topic: 'delivery-topic', + groupId: 'delivery-group', + deliveryTimeoutMs: 240001, + }, + })); + }); +}); From 4e415b4f5051366575e13f8a9a7f85017c47c4b2 Mon Sep 17 00:00:00 2001 From: Anurag Mittal <1321012+anurag4DSB@users.noreply.github.com> Date: Fri, 14 Aug 2026 09:53:15 +0200 Subject: [PATCH 3/3] S3C-11127: publish addressed entries when the delivery pool is enabled Splits the destination fan-out in two: the existing path, moved as is, and a new one that publishes one record per matching destination on the shared delivery topic, carrying destinationId and configurationId in the payload instead of encoding the destination in the topic. With the pool disabled the populator behaves exactly as before. --- .../NotificationQueuePopulator.js | 185 +++++++++++++----- .../NotificationQueuePopulator.js | 166 ++++++++++++++++ 2 files changed, 299 insertions(+), 52 deletions(-) diff --git a/extensions/notification/NotificationQueuePopulator.js b/extensions/notification/NotificationQueuePopulator.js index 30b084d5fb..4fe3f8d6e0 100644 --- a/extensions/notification/NotificationQueuePopulator.js +++ b/extensions/notification/NotificationQueuePopulator.js @@ -7,6 +7,7 @@ const VID_SEPERATOR = require('arsenal').versioning.VersioningConstants.VersionI const configUtil = require('./utils/config'); const safeJsonParse = require('../../lib/util/safeJsonParse'); const messageUtil = require('./utils/message'); +const { buildDeliveryKey } = require('./utils/deliveryKey'); const notifConstants = require('./constants'); const QueuePopulatorExtension = require('../../lib/queuePopulator/QueuePopulatorExtension'); @@ -204,6 +205,132 @@ class NotificationQueuePopulator extends QueuePopulatorExtension { return (value && value[notifConstants.eventMessageProperty.dateTime]) || null; } + /** + * Publish the entry on the internal topic of every destination it + * matches, at most once per internal topic + * + * @param {String} bucket - bucket + * @param {Object} config - bucket notification configuration + * @param {Object} ent - notification entry + * @param {Object} value - log entry object + * @return {undefined} + */ + _publishLegacyEntries(bucket, config, ent, value) { + const { versionId, eventType } = ent; + const pushedToTopic = new Map(); + // validate and push kafka message foreach destination topic + this.notificationConfig.destinations.forEach(destination => { + const topic = destination.internalTopic || + this.notificationConfig.topic; + // avoid pushing a message multiple times to the + // same internal topic + if (pushedToTopic[topic]) { + return undefined; + } + // get destination specific notification config + const queueConfig = config.notificationConfiguration.queueConfig.filter( + c => c.queueArn.split(':').pop() === destination.resource + ); + if (!queueConfig.length) { + // skip, if there is no config for the current + // destination resource + return undefined; + } + // pass only destination resource specific config to + // validate entry + const destConfig = { + bucket, + notificationConfiguration: { + queueConfig, + }, + }; + const { isValid, matchingConfig } = configUtil.validateEntry(destConfig, ent); + if (isValid) { + const message + = messageUtil.addLogAttributes(value, ent); + this.log.info('publishing message', { + method: 'NotificationQueuePopulator._processObjectEntry', + bucket, + key: message.key, + versionId, + eventType, + eventTime: message.dateTime, + matchingConfig, + }); + this.publish(topic, + // keeping all messages for same object + // in the same partition to keep the order. + // here we use the object name and not the + // "_id" which also includes the versionId + `${bucket}/${message.key}`, + JSON.stringify(message)); + // keep track of internal topics we have pushed to + pushedToTopic[topic] = true; + } + return undefined; + }); + } + + /** + * Publish one addressed entry per matching destination on the shared + * delivery topic. The destination is carried in the message instead of + * the topic, so the delivery pool knows where to send it without + * needing a topic per destination. + * + * @param {String} bucket - bucket + * @param {Object} config - bucket notification configuration + * @param {Object} ent - notification entry + * @param {Object} value - log entry object + * @return {undefined} + */ + _publishAddressedEntries(bucket, config, ent, value) { + const { versionId, eventType } = ent; + const { deliveryPool } = this.notificationConfig; + this.notificationConfig.destinations.forEach(destination => { + // get destination specific notification config + const queueConfig = config.notificationConfiguration.queueConfig.filter( + c => c.queueArn.split(':').pop() === destination.resource + ); + if (!queueConfig.length) { + // skip, if there is no config for the current + // destination resource + return undefined; + } + // pass only destination resource specific config to + // validate entry + const destConfig = { + bucket, + notificationConfiguration: { + queueConfig, + }, + }; + const { isValid, matchingConfig } = configUtil.validateEntry(destConfig, ent); + if (isValid) { + const message + = messageUtil.addLogAttributes(value, ent); + message.destinationId = destination.resource; + message.configurationId = matchingConfig.id; + this.log.info('publishing addressed message', { + method: 'NotificationQueuePopulator._publishAddressedEntries', + bucket, + key: message.key, + versionId, + eventType, + eventTime: message.dateTime, + destinationId: message.destinationId, + matchingConfig, + }); + this.publish(deliveryPool.topic, + // the key keeps all messages for the same destination + // and object in the same partition, so that a single + // worker handles them in order + buildDeliveryKey(destination, bucket, message.key), + JSON.stringify(message)); + } + return undefined; + }); + } + /** * Process object entry from the log * @@ -250,58 +377,12 @@ class NotificationQueuePopulator extends QueuePopulatorExtension { key, eventType, }); - const pushedToTopic = new Map(); - // validate and push kafka message foreach destination topic - this.notificationConfig.destinations.forEach(destination => { - const topic = destination.internalTopic || - this.notificationConfig.topic; - // avoid pushing a message multiple times to the - // same internal topic - if (pushedToTopic[topic]) { - return undefined; - } - // get destination specific notification config - const queueConfig = config.notificationConfiguration.queueConfig.filter( - c => c.queueArn.split(':').pop() === destination.resource - ); - if (!queueConfig.length) { - // skip, if there is no config for the current - // destination resource - return undefined; - } - // pass only destination resource specific config to - // validate entry - const destConfig = { - bucket, - notificationConfiguration: { - queueConfig, - }, - }; - const { isValid, matchingConfig } = configUtil.validateEntry(destConfig, ent); - if (isValid) { - const message - = messageUtil.addLogAttributes(value, ent); - this.log.info('publishing message', { - method: 'NotificationQueuePopulator._processObjectEntry', - bucket, - key: message.key, - versionId, - eventType, - eventTime: message.dateTime, - matchingConfig, - }); - this.publish(topic, - // keeping all messages for same object - // in the same partition to keep the order. - // here we use the object name and not the - // "_id" which also includes the versionId - `${bucket}/${message.key}`, - JSON.stringify(message)); - // keep track of internal topics we have pushed to - pushedToTopic[topic] = true; - } - return undefined; - }); + const { deliveryPool } = this.notificationConfig; + if (deliveryPool && deliveryPool.enabled) { + this._publishAddressedEntries(bucket, config, ent, value); + } else { + this._publishLegacyEntries(bucket, config, ent, value); + } } // skip if there is no bucket notification configuration return undefined; diff --git a/tests/unit/notification/NotificationQueuePopulator.js b/tests/unit/notification/NotificationQueuePopulator.js index b38a16d678..95766630b4 100644 --- a/tests/unit/notification/NotificationQueuePopulator.js +++ b/tests/unit/notification/NotificationQueuePopulator.js @@ -741,3 +741,169 @@ describe('NotificationQueuePopulator with multiple rules ::', () => { }); }); }); + +describe('NotificationQueuePopulator with delivery pool ::', () => { + const deliveryTopic = notificationConfig.deliveryPool.topic; + const addressedConfig = { + bucket: 'example-bucket', + notificationConfiguration: { + queueConfig: [ + { + events: ['s3:ObjectCreated:Put'], + queueArn: 'arn:scality:bucketnotif:::destination1', + id: 'config-1', + filterRules: [], + }, + { + events: ['s3:ObjectCreated:Put'], + queueArn: 'arn:scality:bucketnotif:::destination2', + id: 'config-2', + filterRules: [], + }, + ], + }, + }; + const objectEntry = { + 'originOp': 's3:ObjectCreated:Put', + 'dataStoreName': 'metastore', + 'content-length': '100', + 'last-modified': '0000', + 'md-model-version': '1', + }; + let bnConfigManager; + let notificationQueuePopulator; + + beforeEach(() => { + bnConfigManager = new NotificationConfigManager({ + mongoConfig, + bucketMetastore: '__metastore', + maxCachedConfigs: 1000, + logger, + }); + sinon.stub(bnConfigManager, 'getConfig').returns(addressedConfig); + notificationQueuePopulator = new NotificationQueuePopulator({ + config: { + ...notificationConfig, + deliveryPool: { + ...notificationConfig.deliveryPool, + enabled: true, + }, + }, + bnConfigManager, + logger, + }); + notificationQueuePopulator._metricsStore = { + notifEvent: () => null, + }; + }); + + it('should publish one record per matching destination on the delivery topic', async () => { + const publishStub = sinon.stub(notificationQueuePopulator, 'publish'); + await notificationQueuePopulator._processObjectEntry( + 'example-bucket', + 'example-key', + objectEntry); + assert(publishStub.calledTwice); + assert.strictEqual(publishStub.getCall(0).args.at(0), deliveryTopic); + assert.strictEqual(publishStub.getCall(1).args.at(0), deliveryTopic); + }); + + it('should use the destination resource as key when the spread factor is 1', async () => { + const publishStub = sinon.stub(notificationQueuePopulator, 'publish'); + await notificationQueuePopulator._processObjectEntry( + 'example-bucket', + 'example-key', + objectEntry); + assert.strictEqual(publishStub.getCall(0).args.at(1), 'destination1'); + assert.strictEqual(publishStub.getCall(1).args.at(1), 'destination2'); + }); + + it('should address each record with its destination and configuration', async () => { + const publishStub = sinon.stub(notificationQueuePopulator, 'publish'); + await notificationQueuePopulator._processObjectEntry( + 'example-bucket', + 'example-key', + objectEntry); + const first = JSON.parse(publishStub.getCall(0).args.at(2)); + const second = JSON.parse(publishStub.getCall(1).args.at(2)); + assert.strictEqual(first.destinationId, 'destination1'); + assert.strictEqual(first.configurationId, 'config-1'); + assert.strictEqual(first.bucket, 'example-bucket'); + assert.strictEqual(first.key, 'example-key'); + assert.strictEqual(first.eventType, 's3:ObjectCreated:Put'); + assert.strictEqual(second.destinationId, 'destination2'); + assert.strictEqual(second.configurationId, 'config-2'); + }); + + it('should publish one record per destination even when destinations ' + + 'share an internal topic', async () => { + notificationQueuePopulator.notificationConfig = { + ...notificationQueuePopulator.notificationConfig, + destinations: notificationConfig.destinations.map(destination => ({ + ...destination, + internalTopic: 'custom-topic', + })), + }; + const publishStub = sinon.stub(notificationQueuePopulator, 'publish'); + await notificationQueuePopulator._processObjectEntry( + 'example-bucket', + 'example-key', + objectEntry); + assert(publishStub.calledTwice); + assert.strictEqual(publishStub.getCall(0).args.at(0), deliveryTopic); + assert.strictEqual(publishStub.getCall(1).args.at(0), deliveryTopic); + const destinationIds = [0, 1].map(call => + JSON.parse(publishStub.getCall(call).args.at(2)).destinationId); + assert.deepStrictEqual(destinationIds, ['destination1', 'destination2']); + }); + + it('should spread a destination over its spread factor and keep the key ' + + 'stable for the same object', async () => { + notificationQueuePopulator.notificationConfig = { + ...notificationQueuePopulator.notificationConfig, + destinations: notificationConfig.destinations.map(destination => ({ + ...destination, + spreadFactor: 4, + })), + }; + const publishStub = sinon.stub(notificationQueuePopulator, 'publish'); + await notificationQueuePopulator._processObjectEntry( + 'example-bucket', + 'example-key', + objectEntry); + await notificationQueuePopulator._processObjectEntry( + 'example-bucket', + 'example-key', + objectEntry); + assert.strictEqual(publishStub.callCount, 4); + const key = publishStub.getCall(0).args.at(1); + const [resource, index] = key.split('|'); + assert.strictEqual(resource, 'destination1'); + assert(Number(index) >= 0 && Number(index) < 4); + // the same object is always addressed to the same key, so the same + // partition, so the same delivery worker + assert.strictEqual(publishStub.getCall(2).args.at(1), key); + }); + + it('should not publish anything when no destination matches', async () => { + bnConfigManager.getConfig.returns({ + bucket: 'example-bucket', + notificationConfiguration: { + queueConfig: [ + { + events: ['s3:ObjectRemoved:Delete'], + queueArn: 'arn:scality:bucketnotif:::destination1', + id: 'config-1', + filterRules: [], + }, + ], + }, + }); + const publishStub = sinon.stub(notificationQueuePopulator, 'publish'); + await notificationQueuePopulator._processObjectEntry( + 'example-bucket', + 'example-key', + objectEntry); + assert(publishStub.notCalled); + }); +});