Update doc

Author: kenneth-jia

doc/KafkaConsumerQuickStart.md

@@ -6,27 +6,27 @@ 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.
 
-## KafkaAutoCommitConsumer
+## KafkaConsumer (`enable.auto.commit=true`)
 
-* Friendly for users, --would not care about when to commit the offsets for these received messages.
+* Automatically commits previously polled offsets on each `poll` (and the final `close`) operations.
 
-* Internally, it would commit the offsets (for received records) within the next `poll` and the final `close`.
-Note, each internal `commit` would "try its best", but "not guaranteed to succeed", -- it's supposed to be called periodically, thus occasional failure doesn't matter.
+    * Note, the internal `offset commit` is asynchronous, and is not guaranteed to succeed. It's supposed to be triggered (within each `poll` operation) periodically, thus the occasional failure doesn't quite matter.
 
 ### Example
 ```cpp
         // Create configuration object
         kafka::Properties props ({
-            {"bootstrap.servers", brokers},
+            {"bootstrap.servers",  brokers},
+            {"enable.auto.commit", "true"}
         });
 
-        // Create a consumer instance.
-        kafka::KafkaAutoCommitConsumer consumer(props);
+        // Create a consumer instance
+        kafka::clients::KafkaConsumer consumer(props);
 
         // Subscribe to topics
         consumer.subscribe({topic});
 
-        // Read messages from the topic.
+        // Read messages from the topic
         std::cout << "% Reading messages from topic: " << topic << std::endl;
         while (true) {
             auto records = consumer.poll(std::chrono::milliseconds(100));
@@ -48,17 +48,19 @@ Note, each internal `commit` would "try its best", but "not guaranteed to succee
                 }
             }
         }
+        
+        // consumer.close(); // No explicit close is needed, RAII will take care of it
 ```
 
-* `ConsumerConfig::BOOTSTRAP_SERVERS` is mandatory for `ConsumerConfig`.
+* `bootstrap.servers` property is mandatory for a Kafka client.
 
-* `subscribe` could take a topic list. And it's a blocking operation, -- would return after the rebalance event triggered callback was executed.
+* `subscribe` could take a topic list. It's a block operation, would wait the consumer to get partitions assigned.
 
-* `poll` must be periodically called, and it would trigger kinds of callback handling internally. As in this example, just put it in a "while loop" would be OK.
+* `poll` must be called periodically, thus to trigger kinds of callback handling internally. In practice, it could be put in a "while loop".
 
-* At the end, the user could `close` the consumer manually, or just leave it to the destructor (which would `close` anyway).
+* At the end, we could `close` the consumer explicitly, or just leave it to the destructor.
 
-## KafkaManualCommitConsumer
+## KafkaConsumer (`enable.auto.commit=false`)
 
 * Users must commit the offsets for received records manually.
 
@@ -69,15 +71,15 @@ Note, each internal `commit` would "try its best", but "not guaranteed to succee
             {"bootstrap.servers", brokers},
         });
 
-        // Create a consumer instance.
-        kafka::KafkaManualCommitConsumer consumer(props);
+        // Create a consumer instance
+        kafka::clients::KafkaConsumer consumer(props);
 
         // Subscribe to topics
         consumer.subscribe({topic});
 
         auto lastTimeCommitted = std::chrono::steady_clock::now();
 
-        // Read messages from the topic.
+        // Read messages from the topic
         std::cout << "% Reading messages from topic: " << topic << std::endl;
         bool allCommitted = true;
         bool running      = true;
@@ -110,31 +112,33 @@ Note, each internal `commit` would "try its best", but "not guaranteed to succee
                 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: " << kafka::utility::getCurrentTime() << std::endl;
                     consumer.commitSync(); // or commitAsync()
 
                     lastTimeCommitted = now;
                     allCommitted      = true;
                 }
             }
         }
+
+        // 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.
 
 * `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.
 
-## KafkaManualCommitConsumer with `KafkaClient::EventsPollingOption::Manual`
+## `KafkaConsumer` with `kafka::clients::KafkaClient::EventsPollingOption`
 
-While we construct a `KafkaManualCommitConsumer` with option `KafkaClient::EventsPollingOption::AUTO` (default), an internal thread would be created for `OffsetCommit` callbacks handling.
+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.
 
 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 `KafkaClient::EventsPollingOption::Manual`, thus the `OffsetCommit` callbacks would be called within member function `pollEvents()`.
+Here we have another choice, -- using `kafka::clients::KafkaClient::EventsPollingOption::Manual`, thus the `OffsetCommit` callbacks would be called within member function `pollEvents()`.
 
 ### Example
 ```cpp
-    KafkaManualCommitConsumer consumer(props, KafkaClient::EventsPollingOption::Manual);
+    KafkaConsumer consumer(props, kafka::clients::KafkaClient::EventsPollingOption::Manual);
 
     consumer.subscribe({"topic1", "topic2"});
 
@@ -156,17 +160,17 @@ Here we have another choice, -- using `KafkaClient::EventsPollingOption::Manual`
 
 ## Error handling
 
-No exception would be thrown by `KafkaProducer::poll()`.
+No exception would be thrown from a consumer's `poll` operation.
 
-Once an error occurs, the `ErrorCode` would be embedded in the `Consumer::ConsumerRecord`.
+Instead, once an error occurs, the `Error` would be embedded in the `Consumer::ConsumerRecord`.
 
-There're 2 cases,
+About `Error`'s `value()`s, there are 2 cases
 
 1. Success
 
-    - RD_KAFKA_RESP_ERR__NO_ERROR (0),    -- got a message successfully
+    - `RD_KAFKA_RESP_ERR__NO_ERROR` (`0`),  -- got a message successfully
 
-    - RD_KAFKA_RESP_ERR__PARTITION_EOF,   -- reached the end of a partition (no message got)
+    - `RD_KAFKA_RESP_ERR__PARTITION_EOF`,   -- reached the end of a partition (no message got)
 
 2. Failure
 
@@ -187,21 +191,20 @@ There're 2 cases,
 
 * How many threads would be created by a KafkaConsumer?
 
-    Excluding the user's main thread, `KafkaAutoCommitConsumer` would start another (N + 2) threads in the background, while `KafkaManualConsumer` would start (N + 3) background threads. (N means the number of BOOTSTRAP_SERVERS)
+    Excluding the user's main thread, if `enable.auto.commit` is `false`, the `KafkaConsumer` would start another (N + 2) threads in the background; otherwise, the `KafkaConsumer` would start (N + 3) background threads. (N means the number of BOOTSTRAP_SERVERS)
 
     1. Each broker (in the list of BOOTSTRAP_SERVERS) would take a seperate thread to transmit messages towards a kafka cluster server.
 
     2. Another 3 threads will handle internal operations, consumer group operations, and kinds of timers, etc.
 
-    3. KafkaManualConsumer has one more thread, which keeps polling the offset-commit callback event.
+    3. To enable the auto commit, one more thread would be create, which keeps polling/processing the offset-commit callback event.
 
-    E.g, if a KafkaAutoCommitConsumer was created with property of `BOOTSTRAP_SERVERS=127.0.0.1:8888,127.0.0.1:8889,127.0.0.1:8890`, it would take 6 threads in total (including the main thread).
+    E.g, if a KafkaConsumer was created with property of `BOOTSTRAP_SERVERS=127.0.0.1:8888,127.0.0.1:8889,127.0.0.1:8890`, it would take 6 threads in total (including the main thread).
 
 * Which one of these threads will handle the callbacks?
 
     There are 2 kinds of callbacks for a KafkaConsumer,
 
     1. `RebalanceCallback` will be triggered internally by the user's thread, -- within the `poll` function.
 
-    2. `OffsetCommitCallback` (only available for `KafkaManualCommitConsumer`) will be triggered by a background thread, not by the user's thread.
-
+    2. If `enable.auto.commit=true`, the `OffsetCommitCallback` will be triggered by the user's `poll` thread; otherwise, it would be triggered by a background thread.

doc/KafkaProducerQuickStart.md

@@ -6,42 +6,50 @@ We'd recommend users to cross-reference them, --especially the examples.
 
 ## KafkaProducer
 
-* The `send` is an unblocking operation, and the result (including errors) could only be got from the delivery callback.
+* The `send` is an unblock operation, and the result (including errors) could only be got from the delivery callback.
 
 ### Example
 ```cpp
+        using namespace kafka::clients;
+
         // Create configuration object
         kafka::Properties props ({
             {"bootstrap.servers",  brokers},
             {"enable.idempotence", "true"},
         });
 
-        // Create a producer instance.
-        kafka::KafkaProducer producer(props);
+        // Create a producer instance
+        KafkaProducer producer(props);
 
-        // Read messages from stdin and produce to the broker.
+        // Read messages from stdin and produce to the broker
         std::cout << "% Type message value and hit enter to produce message. (empty line to quit)" << std::endl;
 
-        for (std::string line; std::getline(std::cin, line);) {
+        for (auto line = std::make_shared<std::string>();
+             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 = kafka::ProducerRecord(topic,
-                                                kafka::NullKey,
-                                                kafka::Value(line.c_str(), line.size()));
-            // Send the message.
+            auto record = producer::ProducerRecord(topic,
+                                                   kafka::NullKey,
+                                                   kafka::Value(line->c_str(), line->size()));
+
+            // Send the message
             producer.send(record,
                           // The delivery report handler
-                          [](const kafka::Producer::RecordMetadata& metadata, std::error_code ec) {
-                              if (!ec) {
+                          // 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) {
+                              if (!error) {
                                   std::cout << "% Message delivered: " << metadata.toString() << std::endl;
                               } else {
-                                  std::cerr << "% Message delivery failed: " << ec.message() << std::endl;
+                                  std::cerr << "% Message delivery failed: " << error.message() << std::endl;
                               }
-                          },
-                          // The memory block given by record.value() would be copied
-                          kafka::KafkaProducer::SendOption::ToCopyRecordValue);
+                          });
 
-            if (line.empty()) break;
+            if (line->empty()) break;
         }
+
+        // producer.close(); // No explicit close is needed, RAII will take care of it
 ```
 
 * User must guarantee the memory block for `ProducerRecord`'s `key` is valid until being `send`.
@@ -50,40 +58,42 @@ We'd recommend users to cross-reference them, --especially the examples.
 
 * It's guaranteed that the delivery callback would be triggered anyway after `send`, -- a producer would even be waiting for it before `close`. So, it's a good way to release these memory resources in the `Producer::Callback` function.
 
-## `KafkaProducer` with `KafkaClient::EventsPollingOption::Manual`
+## `KafkaProducer` with `kafka::clients::KafkaClient::EventsPollingOption`
 
-While we construct a `KafkaProducer` with option `KafkaClient::EventsPollingOption::Auto` (default), an internal thread would be created for `MessageDelivery` callbacks handling.
+While we construct a `KafkaProducer` with `kafka::clients::KafkaClient::EventsPollingOption::Auto` (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 `KafkaClient::EventsPollingOption::Manual`, thus the `MessageDelivery` callbacks would be called within member function `pollEvents()`.
+Here we have another choice, -- using `kafka::clients::KafkaClient::EventsPollingOption::Manual`, 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`).
 
 ### Example
 ```cpp
-    kafak::KafkaProducer producer(props, KafkaClient::EventsPollingOption::Manual);
+    using namespace kafka::clients;
+
+    KafkaProducer producer(props, KafkaClient::EventsPollingOption::Manual);
 
     // Prepare "msgsToBeSent"
     auto std::map<int, std::pair<Key, Value>> msgsToBeSent = ...;
 
     for (const auto& msg : msgsToBeSent) {
-        auto record = kafak::ProducerRecord(topic, partition, msg.second.first, msg.second.second, msg.first);
-        std::error_code ec;
-        producer.send(ec,
+        auto record = producer::ProducerRecord(topic, partition, msg.second.first, msg.second.second, msg.first);
+        kafka::Error sendError;
+        producer.send(sendError,
                       record,
                       // Ack callback
-                      [&msg](const kafka::Producer::RecordMetadata& metadata, std::error_code ec) {
+                      [&msg](const producer::RecordMetadata& metadata, const kafka::Error& deliveryError) {
                            // the message could be identified by `metadata.recordId()`
-                           if (ec)  {
-                               LOG_ERROR("Cannot send out message with recordId={0}", metadata.recordId());
+                           if (deliveryError)  {
+                               std::cerr << "% Message delivery failed: " << deliveryError.message() << std::endl;
                            } else {
                                msgsToBeSend.erase(metadata.recordId()); // Quite safe here
                            }
                        });
-        if (ec) break;
+        if (sendError) break;
     }
 
     // Here we call the `MessageDelivery` callbacks
@@ -99,9 +109,11 @@ This makes sense, since you might want to call `pollEvents()` later, thus delive
 
 ### Example
 ```cpp
+    using namespace kafka::clients;
+
     kafak::KafkaProducer producer(props);
 
-    auto record = kafka::ProducerRecord(topic, partition, Key(), Value());
+    auto record = producer::ProducerRecord(topic, partition, Key(), Value());
 
     for (const auto& msg : msgsToBeSent) {
         // Prepare record headers
@@ -117,9 +129,9 @@ This makes sense, since you might want to call `pollEvents()` later, thus delive
 
         producer.send(record,
                       // Ack callback
-                      [&msg](const kafka::Producer::RecordMetadata& metadata, std::error_code ec) {
-                           if (ec)  {
-                               LOG_ERROR("Cannot send out message: {0}, err: {1}", metadata.toString(), ec);
+                      [&msg](const kafka::Producer::RecordMetadata& metadata, , const kafka::Error& error) {
+                           if (error)  {
+                               std::cerr << "% Message delivery failed: " << error.message() << std::endl;
                            }
                        });
     }
@@ -133,17 +145,17 @@ This makes sense, since you might want to call `pollEvents()` later, thus delive
 
 2. Delivery `Error` would be passed through the delivery-callback.
 
-There are 2 kinds of possible errors,
+About `Error`'s `value()`s, there are 2 cases
 
 1. Local errors,
 
-    - RD_KAFKA_RESP_ERR__UNKNOWN_TOPIC      -- The topic doesn't exist
+    - `RD_KAFKA_RESP_ERR__UNKNOWN_TOPIC`      -- The topic doesn't exist
 
-    - RD_KAFKA_RESP_ERR__UNKNOWN_PARTITION  -- The partition doesn't exist
+    - `RD_KAFKA_RESP_ERR__UNKNOWN_PARTITION`  -- The partition doesn't exist
 
-    - RD_KAFKA_RESP_ERR__INVALID_ARG        -- Invalid topic (topic is null or the length is too long (>512))
+    - `RD_KAFKA_RESP_ERR__INVALID_ARG`        -- Invalid topic (topic is null or the length is too long (>512))
 
-    - RD_KAFKA_RESP_ERR__MSG_TIMED_OUT      -- No ack received within the time limit
+    - `RD_KAFKA_RESP_ERR__MSG_TIMED_OUT`      -- No ack received within the time limit
 
 2. Broker errors,
 
@@ -205,5 +217,5 @@ E.g, if a `KafkaProducer` was created with property of `BOOTSTRAP_SERVERS=127.0.
 
 It will be handled by a background thread, not by the user's thread.
 
-Note, should be careful if both the `KafkaProducer::send()` and the `Producer::Callback` might access the same container at the same time.
+Note, should be careful if both the `KafkaProducer::send()` and the `producer::Callback` might access the same container at the same time.
 

examples/kafka_async_producer_copy_payload.cc

@@ -24,18 +24,18 @@ int main(int argc, char **argv)
             {"enable.idempotence", "true"},
         });
 
-        // Create a producer instance.
+        // Create a producer instance
         KafkaProducer producer(props);
 
-        // Read messages from stdin and produce to the broker.
+        // Read messages from stdin and produce to the broker
         std::cout << "% Type message value and hit enter to produce message. (empty line to quit)" << std::endl;
 
         for (std::string line; std::getline(std::cin, line);) {
             // 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()));
-            // Send the message.
+            // Send the message
             producer.send(record,
                           // The delivery report handler
                           [](const producer::RecordMetadata& metadata, const kafka::Error& error) {