Support mTLS on the Kafka consumer

extensions/mongoProcessor/MongoQueueProcessor.js

@@ -102,6 +102,7 @@ class MongoQueueProcessor {
                 site: this.kafkaConfig.site,
                 compressionType: this.kafkaConfig.compressionType,
                 requiredAcks: this.kafkaConfig.requiredAcks,
+                consumerParams: this.kafkaConfig.consumerParams,
             },
             queueProcessor: this.processKafkaEntry.bind(this),
             circuitBreaker: this.mongoProcessorConfig.circuitBreaker,

lib/BackbeatConsumer.js

@@ -15,6 +15,7 @@ const {
     startCircuitBreakerMetricsExport,
 } = require('./CircuitBreaker');
 const { observeKafkaStats } = require('./util/probe');
+const { KAFKA_CONSUMER_PARAMS_SCHEMA } = require('./config.joi');
 const {
     unassignStatus,
     backbeatConsumer: {
@@ -98,6 +99,7 @@ class BackbeatConsumer extends EventEmitter {
                 // Kafka producer params
                 compressionType: joi.string(),
                 requiredAcks: joi.number(),
+                consumerParams: KAFKA_CONSUMER_PARAMS_SCHEMA,
             }).required(),
             topic: joi.string().required(),
             groupId: joi.string().required(),
@@ -146,6 +148,7 @@ class BackbeatConsumer extends EventEmitter {
         this._producerCompressionType = kafka.compressionType;
         this._producerRequiredAcks = kafka.requiredAcks;
         this._site = kafka.site;
+        this._consumerParams = kafka.consumerParams;
         this._fromOffset = fromOffset;
         this._log = new Logger(clientId);
         this._topic = withTopicPrefix(topic);
@@ -221,7 +224,7 @@ class BackbeatConsumer extends EventEmitter {
         process.nextTick(this._checkIfReady.bind(this));
     }
 
-    _initConsumer() {
+    get consumerConfig() {
         // TODO: Ask Rahul/Jonathan if at least once delivery is acceptable.
         // We automatically and periodically commit offsets in the background
         // every 5 seconds (default value of "auto.commit.interval.ms").
@@ -233,6 +236,7 @@ class BackbeatConsumer extends EventEmitter {
         // - same object/version could be replayed multiple times (if
         // replication_status_processor crashes).
         const consumerParams = {
+            ...this._consumerParams,
             'metadata.broker.list': this._kafkaHosts,
             'group.id': this._groupId,
             // This is the default in our current librdkafka version, but we
@@ -255,25 +259,35 @@ class BackbeatConsumer extends EventEmitter {
             'metadata.max.age.ms': 5000,
             'max.poll.interval.ms': this._maxPollIntervalMs,
         };
-        const topicParams = {};
-        if (this._fromOffset !== undefined) {
-            topicParams['auto.offset.reset'] = this._fromOffset;
-        }
         if (this._fetchMaxBytes !== undefined) {
             consumerParams['fetch.message.max.bytes'] = this._fetchMaxBytes;
         }
         if (process.env.RDKAFKA_DEBUG_LOGS) {
             consumerParams.debug = process.env.RDKAFKA_DEBUG_LOGS;
         }
+        if (this._site) {
+            consumerParams['client.rack'] = this._site;
+        }
+        return consumerParams;
+    }
+
+    get topicConfig() {
+        const topicParams = {};
+        if (this._fromOffset !== undefined) {
+            topicParams['auto.offset.reset'] = this._fromOffset;
+        }
+        return topicParams;
+    }
+
+    _initConsumer() {
         if (this._site) {
             this._log.info('follower fetching enabled for topic/consumer group', {
                 site: this._site,
                 topic: this._topic,
                 groupId: this._groupId,
             });
-            consumerParams['client.rack'] = this._site;
         }
-        this._consumer = new kafka.KafkaConsumer(consumerParams, topicParams);
+        this._consumer = new kafka.KafkaConsumer(this.consumerConfig, this.topicConfig);
 
         this._consumer.on('event', event => this._log.info('rdkafka.event', { event }));
         this._consumer.on('event.log', log => this._log.info('rdkafka.log', { log }));

lib/config.joi.js

@@ -22,6 +22,24 @@ const KAFKA_PRODUCER_PARAMS_SCHEMA = joi.object({
     'compression.type': joi.forbidden(),
     'statistics.interval.ms': joi.forbidden(),
 }).unknown(true).default({});
+const KAFKA_CONSUMER_PARAMS_SCHEMA = joi.object({
+    'metadata.broker.list': joi.forbidden(),
+    'group.id': joi.forbidden(),
+    'partition.assignment.strategy': joi.forbidden(),
+    'enable.auto.offset.store': joi.forbidden(),
+    'offset_commit_cb': joi.forbidden(),
+    'allow.auto.create.topics': joi.forbidden(),
+    'statistics.interval.ms': joi.forbidden(),
+    'rebalance_cb': joi.forbidden(),
+    'metadata.max.age.ms': joi.forbidden(),
+    'max.poll.interval.ms': joi.forbidden(),
+    'fetch.message.max.bytes': joi.forbidden(),
+    'client.rack': joi.forbidden(),
+    'debug': joi.forbidden(),
+    'enable.auto.commit': joi.forbidden(),
+    'auto.commit.interval.ms': joi.forbidden(),
+    'auto.offset.reset': joi.forbidden(),
+}).unknown(true).default({});
 const logSourcesJoi = joi.string().valid('bucketd', 'ingestion', 'dmd', 'kafka');
 
 const joiSchema = joi.object({
@@ -42,6 +60,7 @@ const joiSchema = joi.object({
         compressionType: joi.string().default(KAFKA_PRODUCER_DEFAULT_COMPRESSION_TYPE),
         requiredAcks: joi.number().default(KAFKA_PRODUCER_DEFAULT_REQUIRED_ACKS),
         producerParams: KAFKA_PRODUCER_PARAMS_SCHEMA,
+        consumerParams: KAFKA_CONSUMER_PARAMS_SCHEMA,
     },
     transport: transportJoi,
     s3: hostPortJoi.optional(),
@@ -116,4 +135,5 @@ module.exports = {
     KAFKA_PRODUCER_DEFAULT_COMPRESSION_TYPE,
     KAFKA_PRODUCER_DEFAULT_REQUIRED_ACKS,
     KAFKA_PRODUCER_PARAMS_SCHEMA,
+    KAFKA_CONSUMER_PARAMS_SCHEMA,
 };

tests/unit/backbeatConsumer.js

@@ -397,4 +397,118 @@ describe('backbeatConsumer', () => {
             });
         });
     });
+
+    describe('consumerParams', () => {
+        it('should include extra consumerParams in consumerConfig', () => {
+            const consumer = new BackbeatConsumerMock({
+                kafka: {
+                    ...kafka,
+                    consumerParams: {
+                        'security.protocol': 'ssl',
+                        'ssl.ca.location': '/kafka-certs/ca/ca.crt',
+                    },
+                },
+                groupId: 'unittest-group',
+                topic: 'my-test-topic',
+            });
+            const config = consumer.consumerConfig;
+            assert.strictEqual(config['security.protocol'], 'ssl');
+            assert.strictEqual(config['ssl.ca.location'],
+                '/kafka-certs/ca/ca.crt');
+        });
+
+        it('should reject critical built-in params via Joi', () => {
+            assert.throws(
+                () => new BackbeatConsumerMock({
+                    kafka: {
+                        ...kafka,
+                        consumerParams: { 'group.id': 'hijacked-group' },
+                    },
+                    groupId: 'unittest-group',
+                    topic: 'my-test-topic',
+                }),
+                /group\.id/,
+            );
+            assert.throws(
+                () => new BackbeatConsumerMock({
+                    kafka: {
+                        ...kafka,
+                        consumerParams: { 'max.poll.interval.ms': 1000 },
+                    },
+                    groupId: 'unittest-group',
+                    topic: 'my-test-topic',
+                }),
+                /max\.poll\.interval\.ms/,
+            );
+            assert.throws(
+                () => new BackbeatConsumerMock({
+                    kafka: {
+                        ...kafka,
+                        consumerParams: { 'enable.auto.commit': false },
+                    },
+                    groupId: 'unittest-group',
+                    topic: 'my-test-topic',
+                }),
+                /enable\.auto\.commit/,
+            );
+            assert.throws(
+                () => new BackbeatConsumerMock({
+                    kafka: {
+                        ...kafka,
+                        consumerParams: { 'auto.offset.reset': 'latest' },
+                    },
+                    groupId: 'unittest-group',
+                    topic: 'my-test-topic',
+                }),
+                /auto\.offset\.reset/,
+            );
+        });
+
+        it('should throw on invalid librdkafka consumerParams keys', () => {
+            let err;
+            try {
+                new BackbeatConsumer({
+                    kafka: {
+                        ...kafka,
+                        consumerParams: { 'not.a.valid.librdkafka.option': 'value' },
+                    },
+                    groupId: 'unittest-group',
+                    topic: 'my-test-topic',
+                });
+            } catch (e) {
+                err = e;
+            }
+            assert(err, 'expected BackbeatConsumer to throw on invalid consumerParams key');
+            assert.match(err.message,
+                /No such configuration property: "not\.a\.valid\.librdkafka\.option"/);
+        });
+
+        it('should throw on out-of-range librdkafka consumerParams values', () => {
+            let err;
+            try {
+                new BackbeatConsumer({
+                    kafka: {
+                        ...kafka,
+                        consumerParams: { 'fetch.min.bytes': -1 },
+                    },
+                    groupId: 'unittest-group',
+                    topic: 'my-test-topic',
+                });
+            } catch (e) {
+                err = e;
+            }
+            assert(err, 'expected BackbeatConsumer to throw on out-of-range value');
+            assert.match(err.message,
+                /Configuration property "fetch\.min\.bytes" value -1 is outside allowed range/);
+        });
+
+        it('should default to empty consumerParams when not provided', () => {
+            const consumer = new BackbeatConsumerMock({
+                kafka,
+                groupId: 'unittest-group',
+                topic: 'my-test-topic',
+            });
+            assert.deepStrictEqual(consumer._consumerParams, {});
+        });
+    });
 });

tests/unit/mongoProcessor/MongoQueueProcessor.spec.js

@@ -243,14 +243,14 @@ describe('MongoQueueProcessor._getConsumerOptions', () => {
         sinon.restore();
     });
 
-    function startProcessor() {
+    function startProcessor(kafkaConfig = { hosts: 'localhost:9092' }) {
         sinon.stub(BackbeatConsumer.prototype, '_init');
         sinon.stub(Config, 'getBootstrapList').returns([]);
         sinon.stub(Config, 'on');
 
         const proc = _makeProcessor([]);
         proc.logger = { info: () => {}, error: () => {}, fatal: () => {} };
-        proc.kafkaConfig = { hosts: 'localhost:9092' };
+        proc.kafkaConfig = kafkaConfig;
         proc.mongoProcessorConfig = {
             topic: 'backbeat-ingestion',
             groupId: 'backbeat-ingestion-group',
@@ -287,4 +287,17 @@ describe('MongoQueueProcessor._getConsumerOptions', () => {
             done();
         });
     });
+
+    it('forwards the configured kafka consumer params', done => {
+        const proc = startProcessor({
+            hosts: 'localhost:9092',
+            consumerParams: { 'security.protocol': 'ssl' },
+        });
+
+        setImmediate(() => {
+            assert.deepStrictEqual(proc._consumer._consumerParams,
+                { 'security.protocol': 'ssl' });
+            done();
+        });
+    });
 });