diff --git a/extensions/gc/GarbageCollector.js b/extensions/gc/GarbageCollector.js index 82c9a2b1b..b1388d89b 100644 --- a/extensions/gc/GarbageCollector.js +++ b/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, diff --git a/extensions/lifecycle/bucketProcessor/LifecycleBucketProcessor.js b/extensions/lifecycle/bucketProcessor/LifecycleBucketProcessor.js index adc4d45d7..38c07d704 100644 --- a/extensions/lifecycle/bucketProcessor/LifecycleBucketProcessor.js +++ b/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, diff --git a/extensions/lifecycle/objectProcessor/LifecycleObjectProcessor.js b/extensions/lifecycle/objectProcessor/LifecycleObjectProcessor.js index 794449bb5..de7cc37d0 100644 --- a/extensions/lifecycle/objectProcessor/LifecycleObjectProcessor.js +++ b/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, diff --git a/extensions/mongoProcessor/MongoQueueProcessor.js b/extensions/mongoProcessor/MongoQueueProcessor.js index 09b465832..d0b659354 100644 --- a/extensions/mongoProcessor/MongoQueueProcessor.js +++ b/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, diff --git a/extensions/notification/queueProcessor/QueueProcessor.js b/extensions/notification/queueProcessor/QueueProcessor.js index 6228b9f55..1b72f3d3b 100644 --- a/extensions/notification/queueProcessor/QueueProcessor.js +++ b/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, diff --git a/extensions/replication/failedCRR/FailedCRRConsumer.js b/extensions/replication/failedCRR/FailedCRRConsumer.js index f837f9ef7..e3da25a40 100644 --- a/extensions/replication/failedCRR/FailedCRRConsumer.js +++ b/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', diff --git a/extensions/replication/queueProcessor/QueueProcessor.js b/extensions/replication/queueProcessor/QueueProcessor.js index 24edf7d05..0ca51882b 100644 --- a/extensions/replication/queueProcessor/QueueProcessor.js +++ b/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, diff --git a/extensions/replication/replicationStatusProcessor/ReplicationStatusProcessor.js b/extensions/replication/replicationStatusProcessor/ReplicationStatusProcessor.js index 77f7b0a9a..a6036a149 100644 --- a/extensions/replication/replicationStatusProcessor/ReplicationStatusProcessor.js +++ b/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, diff --git a/lib/BackbeatConsumer.js b/lib/BackbeatConsumer.js index f20f1af8d..267bec7a8 100644 --- a/lib/BackbeatConsumer.js +++ b/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', () => { diff --git a/lib/MetricsConsumer.js b/lib/MetricsConsumer.js index 360904d4c..1b2dab82e 100644 --- a/lib/MetricsConsumer.js +++ b/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}`, diff --git a/lib/config.joi.js b/lib/config.joi.js index 77857505d..5f988066d 100644 --- a/lib/config.joi.js +++ b/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, }; diff --git a/lib/queuePopulator/KafkaLogConsumer/LogConsumer.js b/lib/queuePopulator/KafkaLogConsumer/LogConsumer.js index 13a89caa8..d295c53fb 100644 --- a/lib/queuePopulator/KafkaLogConsumer/LogConsumer.js +++ b/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. diff --git a/lib/queuePopulator/KafkaLogReader.js b/lib/queuePopulator/KafkaLogReader.js index 6a2061010..b8378bf5f 100644 --- a/lib/queuePopulator/KafkaLogReader.js +++ b/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', diff --git a/tests/unit/backbeatConsumer.js b/tests/unit/backbeatConsumer.js index 951fad035..a060cabe2 100644 --- a/tests/unit/backbeatConsumer.js +++ b/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, {}); + }); + }); }); diff --git a/tests/unit/lib/queuePopulator/kafkaLogConsumer/LogConsumer.js b/tests/unit/lib/queuePopulator/kafkaLogConsumer/LogConsumer.js index ce36d1a20..28abf35af 100644 --- a/tests/unit/lib/queuePopulator/kafkaLogConsumer/LogConsumer.js +++ b/tests/unit/lib/queuePopulator/kafkaLogConsumer/LogConsumer.js @@ -63,6 +63,33 @@ describe('LogConsumer', () => { sinon.restore(); }); + describe('setup', () => { + it('should forward the configured consumer params to rdkafka', done => { + const fakeConsumer = { + connect: () => {}, + once: (event, cb) => (event === 'ready' ? cb() : undefined), + subscribe: () => {}, + }; + const ctor = sinon.stub(kafka, 'KafkaConsumer').returns(fakeConsumer); + const consumer = new LogConsumer({ + ...kafkaConfig, + consumerParams: { + 'security.protocol': 'ssl', + 'partition.assignment.strategy': 'hijacked', + }, + }, logger); + + consumer.setup(() => { + const params = ctor.firstCall.args[0]; + assert.strictEqual(params['security.protocol'], 'ssl'); + // the keys the consumer sets itself still win + assert.strictEqual(params['partition.assignment.strategy'], + 'range,roundrobin'); + done(); + }); + }); + }); + describe('_resetRecordStream', () => { it('should initialize record stream', () => { logConsumer._resetRecordStream(); diff --git a/tests/unit/mongoProcessor/MongoQueueProcessor.spec.js b/tests/unit/mongoProcessor/MongoQueueProcessor.spec.js index b58f77a79..54c6c1e44 100644 --- a/tests/unit/mongoProcessor/MongoQueueProcessor.spec.js +++ b/tests/unit/mongoProcessor/MongoQueueProcessor.spec.js @@ -243,14 +243,14 @@ describe('MongoQueueProcessor.start', () => { 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', @@ -272,4 +272,17 @@ describe('MongoQueueProcessor.start', () => { 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(); + }); + }); });