Improve doc

kenneth-jia · kenneth-jia · commit 5ad3e90cff5a · 2022-12-07T16:53:52.000+08:00
diff --git a/README.md b/README.md
@@ -163,44 +163,50 @@ Eventually, we worked out the ***modern-cpp-kafka***, -- a header-only library t
 
     * [Kafka Client API](http://opensource.morganstanley.com/modern-cpp-kafka/doxygen/annotated.html)
 
+    * `Properties` for Kafka clients
 
-    * Kafka Client Properties
+        * `Properties` is a map which contains all configuration info needed to initialize a Kafka client. These configuration items are key-value pairs, -- the "key" is a `std::string`, while the "value" could be a `std::string`, a `std::function<...>`, or an `Interceptors`.
 
-        * In most cases, the `Properties` settings for ***modern-cpp-kafka*** are identical with [librdkafka configuration](https://github.com/edenhill/librdkafka/blob/master/CONFIGURATION.md)
+            * K-V Types: `std::string` -> `std::string`
 
-        * With following exceptions
+                * Most are identical with [librdkafka configuration](https://github.com/edenhill/librdkafka/blob/master/CONFIGURATION.md)
 
-            * KafkaConsumer
+                * But with Exceptions
 
-                * Properties with random string as default
+                    * Default Value Changes
 
-                    * `client.id`
+                        * `log_level`(default: `5`): default was `6` from **librdkafka**
 
-                    * `group.id`
+                        * `client.id` (default: random string): no default string from **librdkafka**
 
-                * More properties than ***librdkafka***
+                        * `group.id` (default: random string, for `KafkaConsumer` only): no default string from **librdkafka**
 
-                    * `max.poll.records` (default: `500`): The maxmum number of records that a single call to `poll()` would return
+                    * Additional Options
 
-                * Property which overrides the one from ***librdkafka***
+                        * `enable.manual.events.poll` (default: `false`): To poll the (offset-commit/message-delivery callback) events manually
 
-                    * `enable.auto.commit` (default: `false`): To automatically commit the previously polled offsets on each `poll` operation
+                        *  `max.poll.records` (default: `500`, for `KafkaConsumer` only): The maxmum number of records that a single call to `poll()` would return
 
-                * Properties not supposed to be used (internally shadowed by ***modern-cpp-kafka***)
+                    * Ignored Options
 
-                    * `enable.auto.offset.store`
+                        * `enable.auto.offset.store`: ***modern-cpp-kafka*** will save the offsets in its own way
 
-                    * `auto.commit.interval.ms`
+                        * `auto.commit.interval.ms`: ***modern-cpp-kafka*** will not commit the offsets periodically, instead, it would do it in the next `poll()`.
 
-            * KafkaProducer
 
-                * Properties with random string as default
+            * K-V Types: `std::string` -> `std::function<...>`
 
-                    * `client.id`
+                * `log_cb` -> `LogCallback` (`std::function<void(int, const char*, int, const char* msg)>`)
 
-            * Log level
+                * `error_cb` -> `ErrorCallback` (`std::function<void(const Error&)>`)
 
-                * The default `log_level` is `NOTICE` (`5`) for all these clients
+                * `stats_cb` -> `StatsCallback` (`std::function<void(const std::string&)>`)
+
+                * `oauthbearer_token_refresh_cb` -> `OauthbearerTokenRefreshCallback` (`std::function<SaslOauthbearerToken(const std::string&)>`)
+
+            * K-V Types: `std::string` -> `Interceptors`
+
+                * `interceptors`: takes `Interceptors` as the value type
 
 * Test Environment (ZooKeeper/Kafka cluster) Setup
 
diff --git a/doc/HowToMakeKafkaProducerReliable.md b/doc/HowToMakeKafkaProducerReliable.md
@@ -153,13 +153,16 @@ The `msgid` is used, (along with a base `msgid` value stored at the time the `PI
 ### `KafkaProducer` demo
 
 ```cpp
+    using namespace kafka::clients;
+    using namespace kafka::clients::producer;
+
     std::atomic<bool> running = true;
 
     KafkaProducer producer(
         Properties({
-            { ProducerConfig::BOOTSTRAP_SERVERS,  "192.168.0.1:9092,192.168.0.2:9092,192.168.0.3:9092" },
-            { ProducerConfig::ENABLE_IDEMPOTENCE, "true" },
-            { ProducerConfig::MESSAGE_TIMEOUT_MS, "86400000"}  // as long as 1 day
+            { Config::BOOTSTRAP_SERVERS,  {"192.168.0.1:9092,192.168.0.2:9092,192.168.0.3:9092"} },
+            { ProducerConfig::ENABLE_IDEMPOTENCE, {"true"} },
+            { ProducerConfig::MESSAGE_TIMEOUT_MS, {"86400000"} }  // as long as 1 day
         })
     );
 
@@ -168,7 +171,7 @@ The `msgid` is used, (along with a base `msgid` value stored at the time the `PI
         auto record = ProducerRecord(topic, msg.key, msg.value, msg.id);
         producer.send(record,
                       // Ack callback
-                      [&msg](const Producer::RecordMetadata& metadata, std::error_code ec) {
+                      [&msg](const RecordMetadata& metadata, std::error_code ec) {
                            // the message could be identified by `metadata.recordId()`
                            auto recordId = metadata.recordId();
                            if (ec)  {
diff --git a/doc/KafkaConsumerQuickStart.md b/doc/KafkaConsumerQuickStart.md
@@ -4,8 +4,6 @@ Generally speaking, The `Modern C++ Kafka API` is quite similar with [Kafka Java
 
 We'd recommend users to cross-reference them, --especially the examples.
 
-Unlike Java's KafkaConsumer, here we introduced two derived classes, --KafkaAutoCommitConsumer and KafkaManualCommitConsumer, --depending on whether users should call `commit` manually.
-
 ## KafkaConsumer (`enable.auto.commit=true`)
 
 * Automatically commits previously polled offsets on each `poll` (and the final `close`) operations.
@@ -15,13 +13,12 @@ Unlike Java's KafkaConsumer, here we introduced two derived classes, --KafkaAuto
 ### Example
 ```cpp
         // Create configuration object
-        kafka::Properties props ({
-            {"bootstrap.servers",  brokers},
-            {"enable.auto.commit", "true"}
+        const Properties props ({
+            {"bootstrap.servers", {brokers}},
         });
 
         // Create a consumer instance
-        kafka::clients::KafkaConsumer consumer(props);
+        KafkaConsumer consumer(props);
 
         // Subscribe to topics
         consumer.subscribe({topic});
@@ -40,15 +37,15 @@ Unlike Java's KafkaConsumer, here we introduced two derived classes, --KafkaAuto
                     std::cout << "    Partition: " << record.partition() << std::endl;
                     std::cout << "    Offset   : " << record.offset() << std::endl;
                     std::cout << "    Timestamp: " << record.timestamp().toString() << std::endl;
-                    std::cout << "    Headers  : " << kafka::toString(record.headers()) << std::endl;
+                    std::cout << "    Headers  : " << toString(record.headers()) << std::endl;
                     std::cout << "    Key   [" << record.key().toString() << "]" << std::endl;
                     std::cout << "    Value [" << record.value().toString() << "]" << std::endl;
                 } else {
                     std::cerr << record.toString() << std::endl;
                 }
             }
         }
-        
+
         // consumer.close(); // No explicit close is needed, RAII will take care of it
 ```
 
@@ -67,12 +64,13 @@ Unlike Java's KafkaConsumer, here we introduced two derived classes, --KafkaAuto
 ### Example
 ```cpp
         // Create configuration object
-        kafka::Properties props ({
-            {"bootstrap.servers", brokers},
+        const Properties props ({
+            {"bootstrap.servers",  {brokers}},
+            {"enable.auto.commit", {"false" }}
         });
 
         // Create a consumer instance
-        kafka::clients::KafkaConsumer consumer(props);
+        KafkaConsumer consumer(props);
 
         // Subscribe to topics
         consumer.subscribe({topic});
@@ -98,7 +96,7 @@ Unlike Java's KafkaConsumer, here we introduced two derived classes, --KafkaAuto
                     std::cout << "    Partition: " << record.partition() << std::endl;
                     std::cout << "    Offset   : " << record.offset() << std::endl;
                     std::cout << "    Timestamp: " << record.timestamp().toString() << std::endl;
-                    std::cout << "    Headers  : " << kafka::toString(record.headers()) << std::endl;
+                    std::cout << "    Headers  : " << toString(record.headers()) << std::endl;
                     std::cout << "    Key   [" << record.key().toString() << "]" << std::endl;
                     std::cout << "    Value [" << record.value().toString() << "]" << std::endl;
 
@@ -112,7 +110,7 @@ Unlike Java's KafkaConsumer, here we introduced two derived classes, --KafkaAuto
                 auto now = std::chrono::steady_clock::now();
                 if (now - lastTimeCommitted > std::chrono::seconds(1)) {
                     // Commit offsets for messages polled
-                    std::cout << "% syncCommit offsets: " << kafka::utility::getCurrentTime() << std::endl;
+                    std::cout << "% syncCommit offsets: " << utility::getCurrentTime() << std::endl;
                     consumer.commitSync(); // or commitAsync()
 
                     lastTimeCommitted = now;
@@ -124,21 +122,21 @@ Unlike Java's KafkaConsumer, here we introduced two derived classes, --KafkaAuto
         // consumer.close(); // No explicit close is needed, RAII will take care of it
 ```
 
-* The example is quite similar with the KafkaAutoCommitConsumer, with only 1 more line added for manual-commit.
+* The example is quite similar with the previous `enable.auto.commit=true` case, but has to call `commitSync`(or `commitAsync`) manually.
 
-* `commitSync` and `commitAsync` are both available for a KafkaManualConsumer. Normally, use `commitSync` to guarantee the commitment, or use `commitAsync`(with `OffsetCommitCallback`) to get a better performance.
+* `commitSync` and `commitAsync` are both available here. Normally, use `commitSync` to guarantee the commitment, or use `commitAsync`(with `OffsetCommitCallback`) to get a better performance.
 
-## `KafkaConsumer` with `kafka::clients::KafkaClient::EventsPollingOption`
+## Option `enable.manual.events.poll`
 
-While we construct a `KafkaConsumer` with `kafka::clients::KafkaClient::EventsPollingOption::Auto` (i.e. the default option), an internal thread would be created for `OffsetCommit` callbacks handling.
+While we construct a `KafkaConsumer` with `enable.manual.events.poll=false` (i.e. the default option), an internal thread would be created for `OffsetCommit` callbacks handling.
 
 This might not be what you want, since then you have to use 2 different threads to process the messages and handle the `OffsetCommit` responses.
 
-Here we have another choice, -- using `kafka::clients::KafkaClient::EventsPollingOption::Manual`, thus the `OffsetCommit` callbacks would be called within member function `pollEvents()`.
+Here we have another choice, -- using `enable.manual.events.poll=true`, thus the `OffsetCommit` callbacks would be called within member function `pollEvents()`.
 
 ### Example
 ```cpp
-    KafkaConsumer consumer(props, kafka::clients::KafkaClient::EventsPollingOption::Manual);
+    KafkaConsumer consumer(props.put("enable.manual.events.poll", "true"));
 
     consumer.subscribe({"topic1", "topic2"});
 
@@ -153,7 +151,7 @@ Here we have another choice, -- using `kafka::clients::KafkaClient::EventsPollin
         }
 
         // Here we call the `OffsetCommit` callbacks
-        // Note, we can only do this while the consumer was constructed with `EventsPollingOption::Manual`.
+        // Note, we can only do this while the consumer was constructed with `enable.manual.events.poll=true`.
         consumer.pollEvents();
     }
 ```
diff --git a/doc/KafkaProducerQuickStart.md b/doc/KafkaProducerQuickStart.md
@@ -10,12 +10,14 @@ We'd recommend users to cross-reference them, --especially the examples.
 
 ### Example
 ```cpp
+        using namespace kafka;
         using namespace kafka::clients;
+        using namespace kafka::clients::producer;
 
         // Create configuration object
-        kafka::Properties props ({
-            {"bootstrap.servers",  brokers},
-            {"enable.idempotence", "true"},
+        const Properties props ({
+            {"bootstrap.servers",  {brokers}},
+            {"enable.idempotence", {"true" }},
         });
 
         // Create a producer instance
@@ -28,17 +30,17 @@ We'd recommend users to cross-reference them, --especially the examples.
              std::getline(std::cin, *line);
              line = std::make_shared<std::string>()) {
             // The ProducerRecord doesn't own `line`, it is just a thin wrapper
-            auto record = producer::ProducerRecord(topic,
-                                                   kafka::NullKey,
-                                                   kafka::Value(line->c_str(), line->size()));
+            auto record = ProducerRecord(topic,
+                                         NullKey,
+                                         Value(line->c_str(), line->size()));
 
             // Send the message
             producer.send(record,
                           // The delivery report handler
                           // Note: Here we capture the shared_pointer of `line`,
                           //       which holds the content for `record.value()`.
                           //       It makes sure the memory block is valid until the lambda finishes.
-                          [line](const producer::RecordMetadata& metadata, const kafka::Error& error) {
+                          [line](const RecordMetadata& metadata, const Error& error) {
                               if (!error) {
                                   std::cout << "% Message delivered: " << metadata.toString() << std::endl;
                               } else {
@@ -60,32 +62,32 @@ We'd recommend users to cross-reference them, --especially the examples.
 
 ## `KafkaProducer` with `kafka::clients::KafkaClient::EventsPollingOption`
 
-While we construct a `KafkaProducer` with `kafka::clients::KafkaClient::EventsPollingOption::Auto` (the default option), an internal thread would be created for `MessageDelivery` callbacks handling.
+While we construct a `KafkaProducer` with `enable.manual.events.poll=false` (the default option), an internal thread would be created for `MessageDelivery` callbacks handling.
 
 This might not be what you want, since then you have to use 2 different threads to send the messages and handle the `MessageDelivery` responses.
 
-Here we have another choice, -- using `kafka::clients::KafkaClient::EventsPollingOption::Manual`, thus the `MessageDelivery` callbacks would be called within member function `pollEvents()`.
+Here we have another choice, -- using `enable.manual.events.poll=true`, thus the `MessageDelivery` callbacks would be called within member function `pollEvents()`.
 
-* Note, if you constructed the `KafkaProducer` with `EventsPollingOption::Manual`, the `send()` would be an `unblocked` operation.
-I.e, once the `message buffering queue` becomes full, the `send()` operation would throw an exception (or return an `error code` with the input reference parameter), -- instead of blocking there.
-This makes sense, since you might want to call `pollEvents()` later, thus delivery-callback could be called for some messages (which could then be removed from the `message buffering queue`).
+* Note, if you constructed the `KafkaProducer` with `enable.manual.events.poll=true`, the `send()` will be an `unblocked` operation even if the `message buffering queue` is full. In that case, the `send()` operation would throw an exception (or return an `error code` with the input reference parameter), -- instead of blocking there. And you might want to call `pollEvents()`, thus delivery-callback could be called for some messages (which could then be removed from the `message buffering queue`).
 
 ### Example
 ```cpp
+    using namespace kafka;
     using namespace kafka::clients;
+    using namespace kafka::clients::producer;
 
-    KafkaProducer producer(props, KafkaClient::EventsPollingOption::Manual);
+    KafkaProducer producer(props.put("enable.manual.events.poll", "true"));
 
     // Prepare "msgsToBeSent"
     auto std::map<int, std::pair<Key, Value>> msgsToBeSent = ...;
     
     for (const auto& msg : msgsToBeSent) {
-        auto record = producer::ProducerRecord(topic, partition, msg.second.first, msg.second.second, msg.first);
+        auto record = ProducerRecord(topic, partition, msg.second.first, msg.second.second, msg.first);
         kafka::Error sendError;
         producer.send(sendError,
                       record,
                       // Ack callback
-                      [&msg](const producer::RecordMetadata& metadata, const kafka::Error& deliveryError) {
+                      [&msg](const RecordMetadata& metadata, const Error& deliveryError) {
                            // the message could be identified by `metadata.recordId()`
                            if (deliveryError)  {
                                std::cerr << "% Message delivery failed: " << deliveryError.message() << std::endl;
@@ -97,7 +99,7 @@ This makes sense, since you might want to call `pollEvents()` later, thus delive
     }
     
     // Here we call the `MessageDelivery` callbacks
-    // Note, we can only do this while the producer was constructed with `EventsPollingOption::MANUAL`.
+    // Note, we can only do this while the producer was constructed with `enable.manual.events.poll=true`.
     producer.pollEvents();
 ```
 
@@ -111,7 +113,7 @@ This makes sense, since you might want to call `pollEvents()` later, thus delive
 ```cpp
     using namespace kafka::clients;
 
-    kafak::KafkaProducer producer(props);
+    kafak::producer::KafkaProducer producer(props);
 
     auto record = producer::ProducerRecord(topic, partition, Key(), Value());
 
@@ -129,7 +131,7 @@ This makes sense, since you might want to call `pollEvents()` later, thus delive
 
         producer.send(record,
                       // Ack callback
-                      [&msg](const kafka::Producer::RecordMetadata& metadata, , const kafka::Error& error) {
+                      [&msg](const kafka::producer::RecordMetadata& metadata, , const kafka::Error& error) {
                            if (error)  {
                                std::cerr << "% Message delivery failed: " << error.message() << std::endl;
                            }
diff --git a/examples/kafka_auto_commit_consumer.cc b/examples/kafka_auto_commit_consumer.cc
@@ -21,8 +21,7 @@ int main(int argc, char **argv)
 
         // Create configuration object
         const Properties props ({
-            {"bootstrap.servers",  {brokers}},
-            {"enable.auto.commit", {"true" }}
+            {"bootstrap.servers", {brokers}}
         });
 
         // Create a consumer instance
diff --git a/examples/kafka_manual_commit_consumer.cc b/examples/kafka_manual_commit_consumer.cc
@@ -21,7 +21,8 @@ int main(int argc, char **argv)
 
         // Create configuration object
         const Properties props ({
-            {"bootstrap.servers", {brokers}},
+            {"bootstrap.servers",  {brokers}},
+            {"enable.auto.commit", {"false"}}
         });
 
         // Create a consumer instance
