feat: enable Kafka configuration via the environment or convict (#143)

src/kafka/config.js

@@ -0,0 +1,7 @@
+const insertDots = object => Object.fromEntries(Object.entries(object).map(([key, value]) => [key.replace(/[A-Z]/g, s => '.' + s.toLowerCase()), value]))
+
+module.exports = config => config && ({
+  ...config,
+  rdkafkaConf: config.rdkafkaConf && insertDots(config.rdkafkaConf),
+  topicConf: config.topicConf && insertDots(config.topicConf)
+})

src/kafka/consumer.js

@@ -43,6 +43,7 @@ const Logger = require('@mojaloop/central-services-logger')
 const Kafka = require('node-rdkafka')
 
 const Protocol = require('./protocol')
+const getConfig = require('./config')
 
 const connectedClients = new Set()
 require('async-exit-hook')(callback => Promise.allSettled(
@@ -205,6 +206,7 @@ exports.ENUMS = ENUMS
 class Consumer extends EventEmitter {
   constructor (topics = [], config = {}) {
     super()
+    config = getConfig(config)
     if (!config.options) {
       config.options = {
         mode: CONSUMER_MODES.recursive,

src/kafka/producer.js

@@ -40,6 +40,7 @@ const EventEmitter = require('events')
 const Logger = require('@mojaloop/central-services-logger')
 const Kafka = require('node-rdkafka')
 const Protocol = require('./protocol')
+const getConfig = require('./config')
 
 const connectedClients = new Set()
 require('async-exit-hook')(callback => Promise.allSettled(
@@ -153,6 +154,7 @@ const
 class Producer extends EventEmitter {
   constructor (config = {}) {
     super()
+    config = getConfig(config)
     if (!config) {
       config = {}
     }

src/util/consumer.js

@@ -63,6 +63,9 @@ const createHandler = async (topicName, config, command) => {
   if (config.rdkafkaConf !== undefined && config.rdkafkaConf['enable.auto.commit'] !== undefined) {
     autoCommitEnabled = config.rdkafkaConf['enable.auto.commit']
   }
+  if (config.rdkafkaConf !== undefined && config.rdkafkaConf.enableAutoCommit !== undefined) {
+    autoCommitEnabled = config.rdkafkaConf.enableAutoCommit
+  }
 
   let connectedTimeStamp = 0
   try {

test/unit/kafka/consumer.test.js

@@ -75,7 +75,7 @@ Test('Consumer test', (consumerTests) => {
         consumeTimeout: 1000
       },
       rdkafkaConf: {
-        'client.id': 'default-client',
+        clientId: 'default-client',
         'group.id': 'kafka-test',
         'metadata.broker.list': 'localhost:9092',
         'enable.auto.commit': false