From 38ef0322cfb7f12ed67b6c3f91f828cbba4ef5ef Mon Sep 17 00:00:00 2001 From: sylvain senechal Date: Mon, 3 Aug 2026 16:30:12 +0200 Subject: [PATCH] configurable kafka producer parameters Issue: BB-685 --- .../ingestion/IngestionConfigValidator.js | 1 + lib/BackbeatProducer.js | 10 +- lib/config.joi.js | 1 + lib/queuePopulator/IngestionPopulator.js | 4 + tests/unit/backbeatProducer.js | 33 +++++++ tests/unit/ingestion/IngestionPopulator.js | 91 +++++++++++++++++++ 6 files changed, 137 insertions(+), 3 deletions(-) diff --git a/extensions/ingestion/IngestionConfigValidator.js b/extensions/ingestion/IngestionConfigValidator.js index b1a9b85caa..f01313038f 100644 --- a/extensions/ingestion/IngestionConfigValidator.js +++ b/extensions/ingestion/IngestionConfigValidator.js @@ -16,6 +16,7 @@ const joiSchema = joi.object({ processor: joi.object({ circuitBreaker: joi.object().optional(), }).optional(), + producerParams: joi.object().unknown(true).default({}), }); function configValidator(backbeatConfig, extConfig) { diff --git a/lib/BackbeatProducer.js b/lib/BackbeatProducer.js index 493924605e..807de84adf 100644 --- a/lib/BackbeatProducer.js +++ b/lib/BackbeatProducer.js @@ -58,6 +58,7 @@ class BackbeatProducer extends EventEmitter { maxRequestSize: joi.number().default(KAFKA_PRODUCER_MESSAGE_MAX_BYTES), compressionType: joi.string().default(KAFKA_PRODUCER_DEFAULT_COMPRESSION_TYPE), requiredAcks: joi.number().default(KAFKA_PRODUCER_DEFAULT_REQUIRED_ACKS), + producerParams: joi.object().unknown(true).default({}), } ); } @@ -75,7 +76,8 @@ class BackbeatProducer extends EventEmitter { } get producerConfig() { - const producerParams = { + const config = { + ...this._producerParams, 'metadata.broker.list': this._kafkaHosts, 'message.max.bytes': this._maxRequestSize, 'dr_cb': true, @@ -84,10 +86,10 @@ class BackbeatProducer extends EventEmitter { }; if (process.env.RDKAFKA_DEBUG_LOGS) { - producerParams.debug = process.env.RDKAFKA_DEBUG_LOGS; + config.debug = process.env.RDKAFKA_DEBUG_LOGS; } - return producerParams; + return config; } get topicConfig() { @@ -125,6 +127,7 @@ class BackbeatProducer extends EventEmitter { maxRequestSize, compressionType, requiredAcks, + producerParams, } = joiResult; this._kafkaHosts = kafka.hosts; this._topic = topic && withTopicPrefix(topic); @@ -132,6 +135,7 @@ class BackbeatProducer extends EventEmitter { this._maxRequestSize = maxRequestSize; this._compressionType = compressionType; this._requiredAcks = requiredAcks; + this._producerParams = producerParams; } connect() { diff --git a/lib/config.joi.js b/lib/config.joi.js index 2ea7a94169..378a5d7657 100644 --- a/lib/config.joi.js +++ b/lib/config.joi.js @@ -35,6 +35,7 @@ const joiSchema = joi.object({ site: joi.string(), compressionType: joi.string().default(KAFKA_PRODUCER_DEFAULT_COMPRESSION_TYPE), requiredAcks: joi.number().default(KAFKA_PRODUCER_DEFAULT_REQUIRED_ACKS), + producerParams: joi.object().unknown(true).default({}), }, transport: transportJoi, s3: hostPortJoi.optional(), diff --git a/lib/queuePopulator/IngestionPopulator.js b/lib/queuePopulator/IngestionPopulator.js index c9e1407c83..0b917cb8f0 100644 --- a/lib/queuePopulator/IngestionPopulator.js +++ b/lib/queuePopulator/IngestionPopulator.js @@ -173,6 +173,10 @@ class IngestionPopulator { maxRequestSize: this.kafkaConfig.maxRequestSize, compressionType: this.kafkaConfig.compressionType, requiredAcks: this.kafkaConfig.requiredAcks, + producerParams: { + ...this.kafkaConfig.producerParams, + ...this.ingestionConfig.producerParams, // Extension params override global params + }, topic, pollIntervalMs: POLL_INTERVAL_MS, }); diff --git a/tests/unit/backbeatProducer.js b/tests/unit/backbeatProducer.js index 793e2c8564..119e219753 100644 --- a/tests/unit/backbeatProducer.js +++ b/tests/unit/backbeatProducer.js @@ -111,6 +111,39 @@ describe('backbeatProducer', () => { 'custom-topic', [{ key: 'foo', message: 'bar' }], () => {}); }); + describe('producerParams', () => { + it('should include extra producerParams in producerConfig', () => { + const producer = new BackbeatProducer({ + kafka, + producerParams: { + 'queue.buffering.max.kbytes': 1048576, + 'queue.buffering.max.messages': 200000, + }, + }); + const config = producer.producerConfig; + assert.strictEqual(config['queue.buffering.max.kbytes'], 1048576); + assert.strictEqual(config['queue.buffering.max.messages'], 200000); + }); + + it('should not let producerParams override critical built-in params', () => { + const producer = new BackbeatProducer({ + kafka, + producerParams: { + 'metadata.broker.list': 'attacker:9092', + 'dr_cb': false, + }, + }); + const config = producer.producerConfig; + assert.strictEqual(config['metadata.broker.list'], kafka.hosts); + assert.strictEqual(config['dr_cb'], true); + }); + + it('should default to empty producerParams when not provided', () => { + const producer = new BackbeatProducer({ kafka }); + assert.deepStrictEqual(producer._producerParams, {}); + }); + }); + afterEach(() => { process.env.KAFKA_TOPIC_PREFIX = ''; }); diff --git a/tests/unit/ingestion/IngestionPopulator.js b/tests/unit/ingestion/IngestionPopulator.js index 270700b5db..ad10e19049 100644 --- a/tests/unit/ingestion/IngestionPopulator.js +++ b/tests/unit/ingestion/IngestionPopulator.js @@ -7,6 +7,7 @@ const config = require('../../../lib/Config'); const IngestionPopulator = require('../../../lib/queuePopulator/IngestionPopulator'); const IngestionReader = require('../../../lib/queuePopulator/IngestionReader'); +const BackbeatProducer = require('../../../lib/BackbeatProducer'); const fakeLogger = require('../../utils/fakeLogger'); const zkConfig = config.zookeeper; @@ -354,4 +355,94 @@ describe('Ingestion Populator', () => { }); }); }); + + describe('_setupProducer producerParams merge', () => { + let capturedProducerParams; + + beforeEach(() => { + sinon.stub(BackbeatProducer.prototype, 'setFromConfig').callsFake(function (cfg) { + capturedProducerParams = cfg.producerParams; + // Minimal instance state so producerConfig getter doesn't throw. + this._kafkaHosts = cfg.kafka.hosts; + this._topic = null; + this._pollIntervalMs = 2000; + this._maxRequestSize = 5000020; + this._compressionType = 'Zstd'; + this._requiredAcks = -1; + this._producerParams = cfg.producerParams || {}; + }); + }); + + afterEach(() => { + sinon.restore(); + capturedProducerParams = undefined; + }); + + it('should pass merged producerParams : extension overrides global', () => { + const globalParams = { + 'queue.buffering.max.kbytes': 1048576, + 'queue.buffering.max.ms': 100, + }; + const extParams = { + 'queue.buffering.max.messages': 200000, + 'queue.buffering.max.ms': 500, + }; + + const populator = new IngestionPopulator( + null, + zkConfig, + { ...kafkaConfig, producerParams: globalParams }, + qpConfig, + mConfig, + rConfig, + { ...ingestionConfig, producerParams: extParams }, + s3Config + ); + + populator._setupProducer(() => {}); + + assert.strictEqual(capturedProducerParams['queue.buffering.max.kbytes'], 1048576); + assert.strictEqual(capturedProducerParams['queue.buffering.max.messages'], 200000); + assert.strictEqual(capturedProducerParams['queue.buffering.max.ms'], 500, + 'extension producerParams should override global kafka.producerParams'); + }); + + it('should work when only global kafka.producerParams are set', () => { + const globalParams = { 'queue.buffering.max.kbytes': 524288 }; + + const populator = new IngestionPopulator( + null, + zkConfig, + { ...kafkaConfig, producerParams: globalParams }, + qpConfig, + mConfig, + rConfig, + ingestionConfig, + s3Config + ); + + populator._setupProducer(() => {}); + + assert.strictEqual(capturedProducerParams['queue.buffering.max.kbytes'], 524288); + }); + + it('should work when only extension producerParams are set', () => { + const extParams = { 'queue.buffering.max.messages': 100000 }; + + const populator = new IngestionPopulator( + null, + zkConfig, + kafkaConfig, + qpConfig, + mConfig, + rConfig, + { ...ingestionConfig, producerParams: extParams }, + s3Config + ); + + populator._setupProducer(() => {}); + + assert.strictEqual(capturedProducerParams['queue.buffering.max.messages'], 100000); + }); + }); });