Configurable kafka producer parameters

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) {

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,13 +127,15 @@ class BackbeatProducer extends EventEmitter {
             maxRequestSize,
             compressionType,
             requiredAcks,
+            producerParams,
         } = joiResult;
         this._kafkaHosts = kafka.hosts;
         this._topic = topic && withTopicPrefix(topic);
         this._pollIntervalMs = pollIntervalMs;
         this._maxRequestSize = maxRequestSize;
         this._compressionType = compressionType;
         this._requiredAcks = requiredAcks;
+        this._producerParams = producerParams;
     }
 
     connect() {

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(),

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,
         });

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 = '';
     });

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);
+        });
+    });
 });