diff --git a/extensions/mongoProcessor/MongoQueueProcessor.js b/extensions/mongoProcessor/MongoQueueProcessor.js index df701bbd5..fac7f03ff 100644 --- a/extensions/mongoProcessor/MongoQueueProcessor.js +++ b/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, diff --git a/lib/BackbeatConsumer.js b/lib/BackbeatConsumer.js index f20f1af8d..b750ff2cc 100644 --- a/lib/BackbeatConsumer.js +++ b/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 })); diff --git a/lib/config.joi.js b/lib/config.joi.js index 77857505d..735e4f963 100644 --- a/lib/config.joi.js +++ b/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, }; diff --git a/tests/unit/backbeatConsumer.js b/tests/unit/backbeatConsumer.js index 951fad035..e3b990448 100644 --- a/tests/unit/backbeatConsumer.js +++ b/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, {}); + }); + }); }); diff --git a/tests/unit/mongoProcessor/MongoQueueProcessor.spec.js b/tests/unit/mongoProcessor/MongoQueueProcessor.spec.js index 68fedecd9..576e56a7e 100644 --- a/tests/unit/mongoProcessor/MongoQueueProcessor.spec.js +++ b/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(); + }); + }); });