Use config from environment

Author: james0209

autoreduce_qp/queue_processor/confluent_consumer.py

@@ -11,11 +11,12 @@
 from autoreduce_utils.clients.connection_exception import ConnectionException
 from autoreduce_utils.message.message import Message
 from autoreduce_utils.clients.producer import Publisher
+from autoreduce_utils.clients.kafka_utils import kafka_config_from_env
 from autoreduce_qp.queue_processor.handle_message import HandleMessage
 
 TRANSACTIONS_TOPIC = os.getenv('KAFKA_TOPIC')
 KAFKA_BROKER_URL = os.getenv("KAFKA_BROKER_URL")
-GROUP_ID = 'mygroup'
+GROUP_ID = 'data_ready-group'
 
 
 class Consumer(threading.Thread):
@@ -49,14 +50,12 @@ def __init__(self, consumer=None):
             try:
                 self.logger.debug("Getting the kafka consumer")
 
-                config = {
-                    'bootstrap.servers': KAFKA_BROKER_URL,
-                    'group.id': GROUP_ID,
-                    'auto.offset.reset': "earliest",
-                    "on_commit": self.on_commit,
-                    'key.deserializer': StringDeserializer('utf_8'),
-                    'value.deserializer': StringDeserializer('utf_8')
-                }
+                config = kafka_config_from_env()
+
+                config['key.deserializer'] = StringDeserializer('utf_8')
+                config['value.deserializer'] = StringDeserializer('utf_8')
+                config['on_commit'] = self.on_commit
+                config['group.id'] = GROUP_ID
                 self.consumer = DeserializingConsumer(config)
             except KafkaException as err:
                 self.logger.error("Could not initialize the consumer: %s", err)