Set defalut log_level as NOTICE(5)

Author: kenneth-jia

README.md

@@ -163,6 +163,9 @@ Eventually, we worked out the `modern-cpp-kafka`, -- a header-only library that
 
                     * `client.id`
 
+            * Log level
+
+                * The default `log_level` is `NOTICE` (`5`) for all these clients
 
 * Test Environment (ZooKeeper/Kafka cluster) Setup
 

include/kafka/Consumer.h

@@ -23,12 +23,31 @@ namespace Consumer
     /**
      * A callback interface that the user can implement to trigger custom actions when the set of partitions assigned to the consumer changes.
      */
-    using RebalanceCallback    = std::function<void(RebalanceEventType eventType, const TopicPartitions& topicPartitions)>;
+    using RebalanceCallback = std::function<void(RebalanceEventType eventType, const TopicPartitions& topicPartitions)>;
+
+    /**
+     * Null RebalanceCallback
+     */
+#if __cplusplus >= 201703L
+    const inline RebalanceCallback NullRebalanceCallback = RebalanceCallback{};
+#else
+    const static RebalanceCallback NullRebalanceCallback = RebalanceCallback{};
+#endif
+
     /**
      * A callback interface that the user can implement to trigger custom actions when a commit request completes.
      */
     using OffsetCommitCallback = std::function<void(const TopicPartitionOffsets& topicPartitionOffsets, std::error_code ec)>;
 
+    /**
+     * Null OffsetCommitCallback
+     */
+#if __cplusplus >= 201703L
+    const inline OffsetCommitCallback NullOffsetCommitCallback = OffsetCommitCallback{};
+#else
+    const static OffsetCommitCallback NullOffsetCommitCallback = OffsetCommitCallback{};
+#endif
+
     /**
      * A metadata struct containing the consumer group information.
      */

include/kafka/KafkaClient.h

@@ -390,6 +390,12 @@ KafkaClient::validateAndReformProperties(const Properties& origProperties)
         }
     }
 
+    // If no "log_level" configured, use LOG_NOTICE as default
+    if (!properties.getProperty(LOG_LEVEL))
+    {
+        properties.put(LOG_LEVEL, std::to_string(LOG_NOTICE));
+    }
+
     return properties;
 }
 

include/kafka/KafkaConsumer.h

@@ -24,10 +24,6 @@ namespace KAFKA_API {
 class KafkaConsumer: public KafkaClient
 {
 protected:
-    using RebalanceEventType    = Consumer::RebalanceEventType;
-    using RebalanceCallback     = Consumer::RebalanceCallback;
-    using OffsetCommitCallback  = Consumer::OffsetCommitCallback;
-
     // Default value for property "max.poll.records" (which is same with Java API)
     static const constexpr char* DEFAULT_MAX_POLL_RECORDS_VALUE = "500";
 
@@ -75,7 +71,7 @@ class KafkaConsumer: public KafkaClient
      * An exception would be thrown if assign is called previously (without a subsequent call to unsubscribe())
      */
     void subscribe(const Topics&               topics,
-                   Consumer::RebalanceCallback cb      = Consumer::RebalanceCallback(),
+                   Consumer::RebalanceCallback cb      = Consumer::NullRebalanceCallback,
                    std::chrono::milliseconds   timeout = std::chrono::milliseconds(DEFAULT_SUBSCRIBE_TIMEOUT_MS));
     /**
      * Get the current subscription.
@@ -293,7 +289,7 @@ class KafkaConsumer: public KafkaClient
     // Rebalance Callback (for class instance)
     void onRebalance(rd_kafka_resp_err_t err, rd_kafka_topic_partition_list_t* rk_partitions);
 
-    RebalanceCallback _rebalanceCb;
+    Consumer::RebalanceCallback _rebalanceCb;
 };
 
 
@@ -360,7 +356,7 @@ KafkaConsumer::close()
 
 // Subscription
 inline void
-KafkaConsumer::subscribe(const Topics& topics, RebalanceCallback cb, std::chrono::milliseconds timeout)
+KafkaConsumer::subscribe(const Topics& topics, Consumer::RebalanceCallback cb, std::chrono::milliseconds timeout)
 {
     std::string topicsStr = toString(topics);
 
@@ -746,8 +742,8 @@ KafkaConsumer::onRebalance(rd_kafka_resp_err_t err, rd_kafka_topic_partition_lis
 
     if (_rebalanceCb)
     {
-        RebalanceEventType et =
-            (err == RD_KAFKA_RESP_ERR__ASSIGN_PARTITIONS ? RebalanceEventType::PartitionsAssigned : RebalanceEventType::PartitionsRevoked);
+        Consumer::RebalanceEventType et =
+            (err == RD_KAFKA_RESP_ERR__ASSIGN_PARTITIONS ? Consumer::RebalanceEventType::PartitionsAssigned : Consumer::RebalanceEventType::PartitionsRevoked);
         _rebalanceCb(et, tps);
     }
 }
@@ -895,17 +891,17 @@ class KafkaManualCommitConsumer: public KafkaConsumer
      * Commit offsets returned on the last poll() for all the subscribed list of topics and partition.
      * Note: If a callback is provided, it's guaranteed to be triggered (before closing the consumer).
      */
-    void commitAsync(const Consumer::OffsetCommitCallback& cb = OffsetCommitCallback());
+    void commitAsync(const Consumer::OffsetCommitCallback& cb = Consumer::NullOffsetCommitCallback);
     /**
      * Commit the specified offsets for the specified records
      * Note: If a callback is provided, it's guaranteed to be triggered (before closing the consumer).
      */
-    void commitAsync(const ConsumerRecord& record, const Consumer::OffsetCommitCallback& cb = OffsetCommitCallback());
+    void commitAsync(const ConsumerRecord& record, const Consumer::OffsetCommitCallback& cb = Consumer::NullOffsetCommitCallback);
     /**
      * Commit the specified offsets for the specified list of topics and partitions to Kafka.
      * Note: If a callback is provided, it's guaranteed to be triggered (before closing the consumer).
      */
-    void commitAsync(const TopicPartitionOffsets& tpos, const Consumer::OffsetCommitCallback& cb = OffsetCommitCallback());
+    void commitAsync(const TopicPartitionOffsets& tpos, const Consumer::OffsetCommitCallback& cb = Consumer::NullOffsetCommitCallback);
 
     /**
      * Call the OffsetCommit callbacks (if any)
@@ -968,16 +964,16 @@ KafkaManualCommitConsumer::commitSync(const TopicPartitionOffsets& tpos)
 }
 
 inline void
-KafkaManualCommitConsumer::commitAsync(const TopicPartitionOffsets& tpos, const OffsetCommitCallback& cb)
+KafkaManualCommitConsumer::commitAsync(const TopicPartitionOffsets& tpos, const Consumer::OffsetCommitCallback& cb)
 {
     auto rk_tpos = rd_kafka_topic_partition_list_unique_ptr(tpos.empty() ? nullptr : createRkTopicPartitionList(tpos));
 
-    rd_kafka_resp_err_t err = rd_kafka_commit_queue(getClientHandle(), rk_tpos.get(), getCommitCbQueue(), &KafkaConsumer::offsetCommitCallback, new OffsetCommitCallback(cb));
+    rd_kafka_resp_err_t err = rd_kafka_commit_queue(getClientHandle(), rk_tpos.get(), getCommitCbQueue(), &KafkaConsumer::offsetCommitCallback, new Consumer::OffsetCommitCallback(cb));
     KAFKA_THROW_IF_WITH_RESP_ERROR(err);
 }
 
 inline void
-KafkaManualCommitConsumer::commitAsync(const ConsumerRecord& record, const OffsetCommitCallback& cb)
+KafkaManualCommitConsumer::commitAsync(const ConsumerRecord& record, const Consumer::OffsetCommitCallback& cb)
 {
     TopicPartitionOffsets tpos;
     // committed offset should be "current received record's offset" + 1
@@ -986,7 +982,7 @@ KafkaManualCommitConsumer::commitAsync(const ConsumerRecord& record, const Offse
 }
 
 inline void
-KafkaManualCommitConsumer::commitAsync(const OffsetCommitCallback& cb)
+KafkaManualCommitConsumer::commitAsync(const Consumer::OffsetCommitCallback& cb)
 {
     commitAsync(TopicPartitionOffsets(), cb);
 }

include/kafka/Types.h

@@ -14,6 +14,7 @@
 #include <string>
 #include <vector>
 
+
 // Use `boost::optional` for C++14, which doesn't support `std::optional`
 #if __cplusplus >= 201703L
 #include <optional>
@@ -25,9 +26,38 @@ template<class T>
 using Optional = boost::optional<T>;
 #endif
 
+
 namespace KAFKA_API {
 
-class ConstBuffer;
+// Which is similar with `boost::const_buffer` (thus avoid the dependency towards `boost`)
+class ConstBuffer
+{
+public:
+    explicit ConstBuffer(const void* data = nullptr, std::size_t size = 0): _data(data), _size(size) {}
+    const void* data()     const { return _data; }
+    std::size_t size()     const { return _size; }
+    std::string toString() const
+    {
+        if (_size == 0) return _data ? "[empty]" : "[NULL]";
+
+        std::ostringstream oss;
+
+        auto printChar = [&oss](const unsigned char c) {
+            if (std::isprint(c)) {
+                oss << c;
+            } else {
+                oss << "[0x" << std::hex << std::setfill('0') << std::setw(2) << static_cast<int>(c) << "]";
+            }
+        };
+        const auto* beg = static_cast<const unsigned char*>(_data);
+        std::for_each(beg, beg + _size, printChar);
+
+        return oss.str();
+    }
+private:
+    const void* _data;
+    std::size_t _size;
+};
 
 /**
  * Topic name.
@@ -50,12 +80,30 @@ using Offset    = std::int64_t;
 using Key       = ConstBuffer;
 using KeySize   = std::size_t;
 
+/**
+ * Null Key.
+ */
+#if __cplusplus >= 201703L
+const inline Key NullKey = Key{};
+#else
+const static Key NullKey = Key{};
+#endif
+
 /**
  * Record value.
  */
 using Value     = ConstBuffer;
 using ValueSize = std::size_t;
 
+/**
+ * Null Value.
+ */
+#if __cplusplus >= 201703L
+const inline Value NullValue = Value{};
+#else
+const static Value NullValue = Value{};
+#endif
+
 /**
  * Topic set.
  */
@@ -82,53 +130,6 @@ using TopicPartitionOffset  = std::tuple<Topic, Partition, Offset>;
 using TopicPartitionOffsets = std::map<TopicPartition, Offset>;
 
 
-class ConstBuffer
-{
-public:
-    explicit ConstBuffer(const void* data = nullptr, std::size_t size = 0): _data(data), _size(size) {}
-    const void* data()     const { return _data; }
-    std::size_t size()     const { return _size; }
-    std::string toString() const
-    {
-        if (_size == 0) return _data ? "[empty]" : "[NULL]";
-
-        std::ostringstream oss;
-
-        auto printChar = [&oss](const unsigned char c) {
-            if (std::isprint(c)) {
-                oss << c;
-            } else {
-                oss << "[0x" << std::hex << std::setfill('0') << std::setw(2) << static_cast<int>(c) << "]";
-            }
-        };
-        const auto* beg = static_cast<const unsigned char*>(_data);
-        std::for_each(beg, beg + _size, printChar);
-
-        return oss.str();
-    }
-private:
-    const void* _data;
-    std::size_t _size;
-};
-
-
-/**
- * Empty Key.
- */
-#if __cplusplus >= 201703L
-const inline Key NullKey = Key();
-#else
-const static Key NullKey = Key();
-#endif
-/**
- * Empty Value.
- */
-#if __cplusplus >= 201703L
-const inline Value NullValue = Value();
-#else
-const static Value NullValue = Value();
-#endif
-
 /**
  * Obtains explanatory string for Topics.
  */

tests/integration/TestKafkaConsumer.cc

@@ -1682,22 +1682,20 @@ TEST(KafkaAutoCommitConsumer, AutoCreateTopics)
     const Topic topic = Utility::getRandomString();
 
     KafkaAutoCommitConsumer consumer(KafkaTestUtility::GetKafkaClientCommonConfig()
-                                     .put("allow.auto.create.topics", "true")
-                                     /* .put("debug",                    "all")
-                                        .put("log_level",                "7")  */ );
+                                     .put("allow.auto.create.topics", "true"));
 
     constexpr int MAX_RETRIES = 2;
     for (int i = 0; i < MAX_RETRIES; ++i)
     {
         // Subscribe topics
-        consumer.subscribe({topic});
+        consumer.subscribe({topic}, Consumer::NullRebalanceCallback, std::chrono::milliseconds(10));
         std::cout << "[" << Utility::getCurrentTime() << "] " << consumer.name() << " subscribed" << std::endl;
         std::cout << "[" << Utility::getCurrentTime() << "] " << consumer.name() << " subscription: " << toString(consumer.subscription()) << std::endl;
         std::cout << "[" << Utility::getCurrentTime() << "] " << consumer.name() << " assignment: " << toString(consumer.assignment()) << std::endl;
 
         if (!consumer.assignment().empty()) break;
 
-        consumer.unsubscribe();
+        consumer.unsubscribe(std::chrono::milliseconds(5));
     }
 
     // Would never make it!