Support mTLS on the Kafka consumer

extensions/gc/GarbageCollector.js

@@ -131,6 +131,7 @@ class GarbageCollector extends EventEmitter {
                 site: this._kafkaConfig.site,
                 compressionType: this._kafkaConfig.compressionType,
                 requiredAcks: this._kafkaConfig.requiredAcks,
+                consumerParams: this._kafkaConfig.consumerParams,
             },
             topic: this._gcConfig.topic,
             groupId: this._gcConfig.consumer.groupId,

extensions/lifecycle/bucketProcessor/LifecycleBucketProcessor.js

@@ -447,6 +447,7 @@ class LifecycleBucketProcessor {
                 backlogMetrics: this._kafkaConfig.backlogMetrics,
                 compressionType: this._kafkaConfig.compressionType,
                 requiredAcks: this._kafkaConfig.requiredAcks,
+                consumerParams: this._kafkaConfig.consumerParams,
             },
             topic: this._lcConfig.bucketTasksTopic,
             groupId: this._lcConfig.bucketProcessor.groupId,

extensions/lifecycle/objectProcessor/LifecycleObjectProcessor.js

@@ -79,6 +79,7 @@ class LifecycleObjectProcessor extends EventEmitter {
                 backlogMetrics: this._kafkaConfig.backlogMetrics,
                 compressionType: this._kafkaConfig.compressionType,
                 requiredAcks: this._kafkaConfig.requiredAcks,
+                consumerParams: this._kafkaConfig.consumerParams,
             },
             topic,
             groupId: this._processConfig.groupId,

extensions/mongoProcessor/MongoQueueProcessor.js

@@ -139,6 +139,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,

extensions/notification/queueProcessor/QueueProcessor.js

@@ -170,6 +170,7 @@ class QueueProcessor extends EventEmitter {
                         site: this.kafkaConfig.site,
                         compressionType: this.kafkaConfig.compressionType,
                         requiredAcks: this.kafkaConfig.requiredAcks,
+                        consumerParams: this.kafkaConfig.consumerParams,
                     },
                     topic: internalTopic,
                     groupId: consumerGroupId,

extensions/replication/failedCRR/FailedCRRConsumer.js

@@ -42,6 +42,7 @@ class FailedCRRConsumer {
                 site: this._kafkaConfig.site,
                 compressionType: this._kafkaConfig.compressionType,
                 requiredAcks: this._kafkaConfig.requiredAcks,
+                consumerParams: this._kafkaConfig.consumerParams,
             },
             topic: this._topic,
             groupId: 'backbeat-retry-group',

extensions/replication/queueProcessor/QueueProcessor.js

@@ -395,6 +395,7 @@ class QueueProcessor extends EventEmitter {
                     this.kafkaConfig.backlogMetrics : undefined,
                 compressionType: this.kafkaConfig.compressionType,
                 requiredAcks: this.kafkaConfig.requiredAcks,
+                consumerParams: this.kafkaConfig.consumerParams,
                 maxPollIntervalMs: this.repConfig.queueProcessor.maxPollIntervalMs,
             },
             topic,

extensions/replication/replicationStatusProcessor/ReplicationStatusProcessor.js

@@ -358,6 +358,7 @@ class ReplicationStatusProcessor {
                 site: this.kafkaConfig.site,
                 compressionType: this.kafkaConfig.compressionType,
                 requiredAcks: this.kafkaConfig.requiredAcks,
+                consumerParams: this.kafkaConfig.consumerParams,
             },
             topic: this.repConfig.replicationStatusTopic,
             groupId: this.repConfig.replicationStatusProcessor.groupId,

lib/BackbeatConsumer.js

@@ -15,6 +15,8 @@ const {
     startCircuitBreakerMetricsExport,
 } = require('./CircuitBreaker');
 const { observeKafkaStats } = require('./util/probe');
+const { KAFKA_CONSUMER_PARAMS_SCHEMA, KAFKA_PRODUCER_PARAMS_SCHEMA } =
+    require('./config.joi');
 const {
     unassignStatus,
     backbeatConsumer: {
@@ -98,6 +100,8 @@ class BackbeatConsumer extends EventEmitter {
                 // Kafka producer params
                 compressionType: joi.string(),
                 requiredAcks: joi.number(),
+                producerParams: KAFKA_PRODUCER_PARAMS_SCHEMA,
+                consumerParams: KAFKA_CONSUMER_PARAMS_SCHEMA,
             }).required(),
             topic: joi.string().required(),
             groupId: joi.string().required(),
@@ -145,7 +149,9 @@ class BackbeatConsumer extends EventEmitter {
         this._maxPollIntervalMs = kafka.maxPollIntervalMs;
         this._producerCompressionType = kafka.compressionType;
         this._producerRequiredAcks = kafka.requiredAcks;
+        this._producerParams = kafka.producerParams;
         this._site = kafka.site;
+        this._consumerParams = kafka.consumerParams;
         this._fromOffset = fromOffset;
         this._log = new Logger(clientId);
         this._topic = withTopicPrefix(topic);
@@ -221,7 +227,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 +239,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
@@ -243,6 +250,9 @@ class BackbeatConsumer extends EventEmitter {
             // contiguous offset fully processed by a worker, so
             // disabling automatic offset store is needed
             'enable.auto.offset.store': false,
+            // the offsets we store are flushed by auto-commit: the default,
+            // pinned explicitly because that mechanism depends on it
+            'enable.auto.commit': true,
             // this function is called periodically based on
             // auto-commit of stored offsets
             'offset_commit_cb': this._onOffsetCommit.bind(this),
@@ -255,25 +265,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 }));
@@ -1042,6 +1062,7 @@ class BackbeatConsumer extends EventEmitter {
             kafka: { hosts: this._kafkaHosts },
             compressionType: this._producerCompressionType,
             requiredAcks: this._producerRequiredAcks,
+            producerParams: this._producerParams,
             topic: this._topic,
         });
         producer.on('ready', () => {
@@ -1176,6 +1197,7 @@ class BackbeatConsumer extends EventEmitter {
                 kafka: { hosts: this._kafkaHosts },
                 compressionType: this._producerCompressionType,
                 requiredAcks: this._producerRequiredAcks,
+                producerParams: this._producerParams,
                 topic: this._topic,
             });
             return producer.on('ready', () => {

lib/MetricsConsumer.js

@@ -65,6 +65,7 @@ class MetricsConsumer {
                 site: this.kafkaConfig.site,
                 compressionType: this.kafkaConfig.compressionType,
                 requiredAcks: this.kafkaConfig.requiredAcks,
+                consumerParams: this.kafkaConfig.consumerParams,
             },
             topic: this.mConfig.topic,
             groupId: `${this.mConfig.groupIdPrefix}-${this._id}`,

lib/config.joi.js

@@ -22,6 +22,23 @@ 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(),
+    'auto.offset.reset': joi.forbidden(),
+    'enable.auto.commit': joi.forbidden(),
+}).unknown(true).default({});
 const logSourcesJoi = joi.string().valid('bucketd', 'ingestion', 'dmd', 'kafka');
 
 const joiSchema = joi.object({
@@ -42,6 +59,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 +134,5 @@ module.exports = {
     KAFKA_PRODUCER_DEFAULT_COMPRESSION_TYPE,
     KAFKA_PRODUCER_DEFAULT_REQUIRED_ACKS,
     KAFKA_PRODUCER_PARAMS_SCHEMA,
+    KAFKA_CONSUMER_PARAMS_SCHEMA,
 };

lib/queuePopulator/KafkaLogConsumer/LogConsumer.js

@@ -21,12 +21,15 @@ class LogConsumer extends EventEmitter {
      * @param {string} kafkaConfig.hosts kafka hosts
      * @param {string} kafkaConfig.topic kafka oplog topic
      * @param {string} kafkaConfig.consumerGroupId consumer group id
+     * @param {Object} [kafkaConfig.consumerParams] extra rdkafka params
      * @param {Logger} logger logger
      */
     constructor(kafkaConfig, logger) {
         super();
-        const { hosts, topic, consumerGroupId, maxPollIntervalMs } = kafkaConfig;
+        const { hosts, topic, consumerGroupId, maxPollIntervalMs,
+            consumerParams } = kafkaConfig;
         this._kafkaHosts = hosts;
+        this._consumerParams = consumerParams;
         this._maxPollIntervalMs = maxPollIntervalMs || 300000; // default to 5 minutes
         this._topic = topic;
         this._consumerGroupId = consumerGroupId;
@@ -55,6 +58,7 @@ class LogConsumer extends EventEmitter {
     setup(done) {
         // partition offsets will be managed by kafka
         const consumerParams = {
+            ...this._consumerParams,
             // This is the default in our current librdkafka version, but we
             // pin it explicitly because we depend on eager rebalancing and
             // don't want it changed implicitly by future version updates.

lib/queuePopulator/KafkaLogReader.js

@@ -23,6 +23,7 @@ class KafkaLogReader extends LogReader {
         // conf contains global kafka and queuePoplator kafka configs
         const conf = {
             hosts: kafkaConfig.hosts,
+            consumerParams: kafkaConfig.consumerParams,
             ...qpKafkaConfig
         };
         logger.info('initializing kafka log reader',

tests/unit/backbeatConsumer.js

@@ -34,6 +34,17 @@ describe('backbeatConsumer', () => {
         });
         assert.strictEqual(backbeatConsumer._topic, 'testing.my-test-topic');
     });
+
+    it('should commit the offsets it stores itself', () => {
+        const consumer = new BackbeatConsumerMock({
+            kafka,
+            groupId: 'unittest-group',
+            topic: 'my-test-topic',
+        });
+        assert.strictEqual(consumer.consumerConfig['enable.auto.offset.store'],
+            false);
+        assert.strictEqual(consumer.consumerConfig['enable.auto.commit'], true);
+    });
 
     describe('pause/resume topic partitions on circuit breaker', () => {
         let consumer;
@@ -397,4 +408,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: { 'auto.offset.reset': 'latest' },
+                    },
+                    groupId: 'unittest-group',
+                    topic: 'my-test-topic',
+                }),
+                /auto\.offset\.reset/,
+            );
+            assert.throws(
+                () => new BackbeatConsumerMock({
+                    kafka: {
+                        ...kafka,
+                        consumerParams: { 'enable.auto.commit': false },
+                    },
+                    groupId: 'unittest-group',
+                    topic: 'my-test-topic',
+                }),
+                /enable\.auto\.commit/,
+            );
+        });
+
+        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, {});
+        });
+    });
 });