Unify KafkaConsumers (i.e. KafkaAutoCommitConsumer & KafkaManualCommitConsumer)

Author: kenneth-jia

CMakeLists.txt

@@ -1,6 +1,6 @@
 cmake_minimum_required(VERSION "3.8")
 
-project("Modern C++ based Kafka API" VERSION 1.0.0)
+project("Modern C++ Kafka API" VERSION 1.0.0)
 
 include(CTest)
 include(CheckCXXCompilerFlag)

NOTICE

@@ -1,3 +1,3 @@
-Modern C++ based Kafka API
+Modern C++ Kafka API
 Copyright 2020 Morgan Stanley
 This project includes software developed at Morgan Stanley.

README.md

@@ -176,12 +176,14 @@ Eventually, we worked out the ***modern-cpp-kafka***, -- a header-only library t
 
                     * `max.poll.records` (default: `500`): The maxmum number of records that a single call to `poll()` would return
 
+                * Property which overrides the one from ***librdkafka***
+
+                    * `enable.auto.commit` (default: `false`): To automatically commit the previously polled offsets on each `poll` operation
+
                 * Properties not supposed to be used (internally shadowed by ***modern-cpp-kafka***)
 
                     * `enable.auto.offset.store`
 
-                    * `enable.auto.commit`
-
                     * `auto.commit.interval.ms`
 
             * KafkaProducer

doc/GoodPracticesToUseKafkaConsumer.md

@@ -16,7 +16,7 @@ If we want to achieve high performance/availability, here're some rules of thumb
 
 ## How to avoid polling duplicated messages
 
-* A `KafkaManualCommitConsumer` could help to commit the offsets more frequently (e.g, always do commit after finishing processing a message).
+* To commit the offsets more frequently (e.g, always do commit after finishing processing a message).
 
-* Don't use quite a large `MAX_POLL_RECORDS` for a `KafkaAutoCommitConsumer`, -- you might fail to commit all these messages before crash, thus more duplications with the next `poll`.
+* Don't use quite a large `MAX_POLL_RECORDS` for a `KafkaConsumer` (with `enable.auto.commit=true`) -- you might fail to commit all these messages before crash, thus more duplications with the next `poll`.
 

doc/KafkaConsumerQuickStart.md

@@ -1,6 +1,6 @@
 # KafkaConsumer Quick Start
 
-Generally speaking, The `Modern C++ based Kafka API` is quite similar with [Kafka Java's API](https://kafka.apache.org/22/javadoc/org/apache/kafka/clients/consumer/KafkaConsumer.html)
+Generally speaking, The `Modern C++ Kafka API` is quite similar with [Kafka Java's API](https://kafka.apache.org/22/javadoc/org/apache/kafka/clients/consumer/KafkaConsumer.html)
 
 We'd recommend users to cross-reference them, --especially the examples.
 

doc/KafkaProducerQuickStart.md

@@ -1,6 +1,6 @@
 # KafkaProducer Quick Start
 
-Generally speaking, The `Modern C++ based Kafka API` is quite similar to the [Kafka Java's API](https://kafka.apache.org/10/javadoc/org/apache/kafka/clients/producer/KafkaProducer.html).
+Generally speaking, The `Modern C++ Kafka API` is quite similar to the [Kafka Java's API](https://kafka.apache.org/10/javadoc/org/apache/kafka/clients/producer/KafkaProducer.html).
 
 We'd recommend users to cross-reference them, --especially the examples.
 

examples/kafka_auto_commit_consumer.cc

@@ -17,11 +17,12 @@ int main(int argc, char **argv)
 
         // Create configuration object
         kafka::Properties props ({
-            {"bootstrap.servers", brokers},
+            {"bootstrap.servers",  brokers},
+            {"enable.auto.commit", "true"}
         });
 
         // Create a consumer instance.
-        kafka::KafkaAutoCommitConsumer consumer(props);
+        kafka::KafkaConsumer consumer(props);
 
         // Subscribe to topics
         consumer.subscribe({topic});

examples/kafka_manual_commit_consumer.cc

@@ -21,7 +21,7 @@ int main(int argc, char **argv)
         });
 
         // Create a consumer instance.
-        kafka::KafkaManualCommitConsumer consumer(props);
+        kafka::KafkaConsumer consumer(props);
 
         // Subscribe to topics
         consumer.subscribe({topic});

include/kafka/AdminClient.h

@@ -89,7 +89,8 @@ class AdminClient: public KafkaClient
 {
 public:
     explicit AdminClient(const Properties& properties)
-        : KafkaClient(ClientType::AdminClient, KafkaClient::validateAndReformProperties(properties))
+        : KafkaClient(ClientType::AdminClient,
+                      KafkaClient::validateAndReformProperties(properties))
     {
     }
 

include/kafka/ConsumerConfig.h

@@ -35,6 +35,11 @@ class ConsumerConfig: public Properties
      */
     static const constexpr char* CLIENT_ID               = "client.id";
 
+    /**
+     * Automatically commits previously polled offsets on each `poll` operation.
+     */
+    static const constexpr char* ENABLE_AUTO_COMMIT      = "enable.auto.commit";
+
     /**
      * This property controls the behavior of the consumer when it starts reading a partition for which it doesn't have a valid committed offset.
      * The "latest" means the consumer will begin reading the newest records written after the consumer started. While "earliest" means that the consumer will read from the very beginning.