Disable automatic committing

This keeps CS and Kafka in sync by making sure all consumed messages have
been committed only after CS successfully applies them.

kafka-avro-adapter/src/controller.cpp

@@ -48,6 +48,7 @@ void Controller::run()
             if (m_queue.size() > 0 && should_flush())
             {
                 m_producer.write(m_queue);
+                m_consumer.commit();
                 m_last_flush = Clock::now();
             }
         }

kafka-avro-adapter/src/kafka_consumer.cpp

@@ -142,6 +142,7 @@ KafkaConsumer::KafkaConsumer(const Options& options):
     std::string errstr;
     conf->set("metadata.broker.list", m_options.broker, errstr);
     conf->set("group.id", m_options.group, errstr);
+    conf->set("enable.auto.commit", "false", errstr);
 
     if (!m_options.debug.empty()) 
     {
@@ -194,3 +195,13 @@ Result KafkaConsumer::read()
 
     return rval;
 }
+
+void KafkaConsumer::commit()
+{
+    RdKafka::ErrorCode err = m_consumer->commitSync();
+
+    if (err != RdKafka::ERR_NO_ERROR)
+    {
+        throw AdapterError(RdKafka::err2str(err));
+    }
+}

kafka-avro-adapter/src/kafka_consumer.h

@@ -28,6 +28,7 @@ class KafkaConsumer
     KafkaConsumer(const Options& options);
     ~KafkaConsumer();
     Result read();
+    void commit();
 
 private:
     const Options& m_options;