Improve clang-tidy config and enable more warnings

Author: kenneth-jia

.clang-tidy

@@ -2,42 +2,34 @@ Checks: "*,\
          -llvm-header-guard,\
          -llvm-namespace-comment,\
          -fuchsia-*,\
-         -google-default-arguments,\
          -google-readability-namespace-comments,\
          -google-build-using-namespace,\
          -google-runtime-references,\
          -google-readability-avoid-underscore-in-googletest-name,\
          -modernize-use-nodiscard,\
          -modernize-deprecated-headers,\
-         -modernize-avoid-c-arrays,\
          -modernize-use-trailing-return-type,\
          -hicpp-special-member-functions,\
          -hicpp-vararg,\
          -hicpp-no-malloc,\
-         -hicpp-avoid-c-arrays,\
          -hicpp-no-array-decay,\
          -hicpp-deprecated-headers,\
          -hicpp-braces-around-statements,\
          -cppcoreguidelines-special-member-function,\
          -cppcoreguidelines-macro-usage,\
          -cppcoreguidelines-avoid-magic-numbers,\
          -cppcoreguidelines-pro-type-vararg,\
-         -cppcoreguidelines-pro-type-cstyle-cast,\
          -cppcoreguidelines-pro-bounds-array-to-pointer-decay,\
          -cppcoreguidelines-pro-bounds-pointer-arithmetic,\
          -cppcoreguidelines-special-member-functions,\
          -cppcoreguidelines-owning-memory,\
-         -cppcoreguidelines-avoid-c-arrays,\
-         -cppcoreguidelines-no-malloc,\
          -cppcoreguidelines-non-private-member-variables-in-classes,\
-         -cppcoreguidelines-pro-bounds-constant-array-index,\
          -cppcoreguidelines-pro-type-union-access,\
          -misc-non-private-member-variables-in-classes,\
          -readability-magic-numbers,\
          -readability-implicit-bool-conversion,\
          -readability-braces-around-statements,\
          -readability-isolate-declaration,\
-         -bugprone-macro-parentheses,\
          -bugprone-unused-return-value,\
          -cert-err58-cpp,\
          -cert-err60-cpp"

CMakeLists.txt

@@ -3,9 +3,15 @@ cmake_minimum_required(VERSION "3.8")
 project("Modern C++ Kafka API" VERSION 1.0.0)
 
 get_property(parent_directory DIRECTORY PROPERTY PARENT_DIRECTORY)
-if(NOT parent_directory)
+if (NOT parent_directory)
     set(cppkafka_master_project ON)
-endif()
+    # Use Strict Options
+    if ((CMAKE_CXX_COMPILER_ID STREQUAL "Clang") OR (CMAKE_CXX_COMPILER_ID STREQUAL "GNU"))
+        add_compile_options("-Wall" "-Werror" "-Wextra" "-Wshadow" "-Wno-unused-result" "-Wsign-conversion")
+    elseif (CMAKE_CXX_COMPILER_ID STREQUAL "MSVC")
+        add_definitions(-D_CRT_SECURE_NO_WARNINGS)
+    endif ()
+endif ()
 
 option(CPPKAFKA_ENABLE_TESTS "Generate the test targets" ${cppkafka_master_project})
 
@@ -80,16 +86,6 @@ if (CMAKE_CXX_STANDARD EQUAL 14)
 endif ()
 
 
-#---------------------------
-# Build Options
-#---------------------------
-if ((CMAKE_CXX_COMPILER_ID STREQUAL "Clang") OR (CMAKE_CXX_COMPILER_ID STREQUAL "GNU"))
-    add_compile_options("-Wall" "-Werror" "-Wextra" "-Wshadow" "-Wno-unused-result")
-elseif (CMAKE_CXX_COMPILER_ID STREQUAL "MSVC")
-    add_definitions(-D_CRT_SECURE_NO_WARNINGS)
-endif ()
-
-
 #---------------------------
 # Build Option: UT stubs
 #---------------------------

examples/CMakeLists.txt

@@ -1,5 +1,6 @@
 project("kafka-examples")
 
+
 # Target: kafka_sync_producer
 add_executable("kafka_sync_producer" "kafka_sync_producer.cc")
 target_link_libraries("kafka_sync_producer" modern-cpp-kafka-api)

include/kafka/ConsumerRecord.h

@@ -107,7 +107,7 @@ ConsumerRecord::headers() const
     const char* name      = nullptr;
     const void* valuePtr  = nullptr;
     std::size_t valueSize = 0;
-    for (int i = 0; !rd_kafka_header_get_all(hdrs, i, &name, &valuePtr, &valueSize); i++)
+    for (std::size_t i = 0; !rd_kafka_header_get_all(hdrs, i, &name, &valuePtr, &valueSize); i++)
     {
         headers.emplace_back(name, Header::Value(valuePtr, valueSize));
     }

include/kafka/KafkaConsumer.h

@@ -205,6 +205,8 @@ class KafkaConsumer: public KafkaClient
      * Returns the polled records.
      * Note: 1) The result could be fetched through ConsumerRecord (with member function `error`).
      *       2) Make sure the `ConsumerRecord` be destructed before the `KafkaConsumer.close()`.
+     * Throws KafkaException with errors:
+     *   - RD_KAFKA_RESP_ERR__UNKNOWN_PARTITION: Unknow partition
      */
     std::vector<consumer::ConsumerRecord> poll(std::chrono::milliseconds timeout);
 
@@ -213,6 +215,8 @@ class KafkaConsumer: public KafkaClient
      * Returns the number of polled records (which have been saved into parameter `output`).
      * Note: 1) The result could be fetched through ConsumerRecord (with member function `error`).
      *       2) Make sure the `ConsumerRecord` be destructed before the `KafkaConsumer.close()`.
+     * Throws KafkaException with errors:
+     *   - RD_KAFKA_RESP_ERR__UNKNOWN_PARTITION: Unknow partition
      */
     std::size_t poll(std::chrono::milliseconds timeout, std::vector<consumer::ConsumerRecord>& output);
 
@@ -286,8 +290,8 @@ class KafkaConsumer: public KafkaClient
 
     std::string  _groupId;
 
-    unsigned int _maxPollRecords   = 500;   // From "max.poll.records" property, and here is the default for batch-poll
-    bool         _enableAutoCommit = false; // From "enable.auto.commit" property
+    std::size_t _maxPollRecords   = 500;   // From "max.poll.records" property, and here is the default for batch-poll
+    bool        _enableAutoCommit = false; // From "enable.auto.commit" property
 
     rd_kafka_queue_unique_ptr _rk_queue;
 
@@ -379,7 +383,7 @@ KafkaConsumer::KafkaConsumer(const Properties &properties, EventsPollingOption e
     if (auto maxPollRecordsProperty = properties.getProperty(consumer::Config::MAX_POLL_RECORDS))
     {
         const std::string maxPollRecords = *maxPollRecordsProperty;
-        _maxPollRecords = std::stoi(maxPollRecords);
+        _maxPollRecords = static_cast<std::size_t>(std::stoi(maxPollRecords));
     }
     _properties.put(consumer::Config::MAX_POLL_RECORDS, std::to_string(_maxPollRecords));
 
@@ -812,11 +816,15 @@ KafkaConsumer::pollMessages(int timeoutMs, std::vector<consumer::ConsumerRecord>
 
     // Poll messages with librdkafka's API
     std::vector<rd_kafka_message_t*> msgPtrArray(_maxPollRecords);
-    std::size_t msgReceived = rd_kafka_consume_batch_queue(_rk_queue.get(), timeoutMs, msgPtrArray.data(), _maxPollRecords);
+    auto msgReceived = rd_kafka_consume_batch_queue(_rk_queue.get(), timeoutMs, msgPtrArray.data(), _maxPollRecords);
+    if (msgReceived < 0)
+    {
+        KAFKA_THROW_ERROR(Error(rd_kafka_last_error()));
+    }
 
     // Wrap messages with ConsumerRecord
     output.clear();
-    output.reserve(msgReceived);
+    output.reserve(static_cast<std::size_t>(msgReceived));
     std::for_each(msgPtrArray.begin(), msgPtrArray.begin() + msgReceived, [&output](rd_kafka_message_t* rkMsg) { output.emplace_back(rkMsg); });
 
     // Store the offsets for all these polled messages (for "enable.auto.commit=true" case)

include/kafka/KafkaProducer.h

@@ -14,6 +14,7 @@
 
 #include <cassert>
 #include <condition_variable>
+#include <cstdint>
 #include <memory>
 #include <mutex>
 #include <unordered_map>
@@ -398,7 +399,7 @@ KafkaProducer::send(const producer::ProducerRecord& record,
         vu.vtype         = RD_KAFKA_VTYPE_HEADER;
         vu.u.header.name = header.key.c_str();
         vu.u.header.val  = header.value.data();
-        vu.u.header.size = header.value.size();
+        vu.u.header.size = static_cast<int64_t>(header.value.size());
     }
 
     assert(uvCount == rkVUs.size());

include/kafka/Log.h

@@ -4,6 +4,8 @@
 
 #include <kafka/Utility.h>
 
+#include <algorithm>
+#include <array>
 #include <cassert>
 #include <functional>
 #include <iostream>
@@ -25,57 +27,56 @@ struct Log
         Debug   = 7
     };
 
-    static const std::string& levelString(int level)
+    static const std::string& levelString(std::size_t level)
     {
-        static const std::vector<std::string> levelNames = {"EMERG", "ALERT", "CRIT", "ERR", "WARNING", "NOTICE", "INFO", "DEBUG"};
-        static const int                      numLevels  = static_cast<int>(levelNames.size());
-        static const std::string              invalid    = "INVALID";
+        static const std::vector<std::string> levelNames = {"EMERG", "ALERT", "CRIT", "ERR", "WARNING", "NOTICE", "INFO", "DEBUG", "INVALID"};
+        static const std::size_t              maxIndex   = levelNames.size() - 1;
 
-        return (level >= 0 && level < numLevels) ? levelNames[level] : invalid;
+        return levelNames[std::min(level, maxIndex)];
     }
 };
 
-template <int MAX_CAPACITY>
+template <std::size_t MAX_CAPACITY>
 class LogBuffer
 {
 public:
-    LogBuffer():_wptr(_buf) { _buf[0] = 0; } // NOLINT
+    LogBuffer():_wptr(_buf.data()) { _buf[0] = 0; } // NOLINT
 
     LogBuffer& clear()
     {
-        _wptr = _buf;
+        _wptr = _buf.data();
         _buf[0] = 0;
         return *this;
     }
 
     template<class ...Args>
     LogBuffer& print(const char* format, Args... args)
     {
-        assert(!(_buf[0] != 0 && _wptr == _buf)); // means it has already been used as a plain buffer (with `str()`)
+        assert(!(_buf[0] != 0 && _wptr == _buf.data())); // means it has already been used as a plain buffer (with `str()`)
 
         auto cnt = std::snprintf(_wptr, capacity(), format, args...); // returns number of characters written if successful (not including '\0')
         if (cnt > 0)
         {
-            _wptr = std::min(_wptr + cnt, _buf + MAX_CAPACITY - 1);
+            _wptr = std::min(_wptr + cnt, _buf.data() + MAX_CAPACITY - 1);
         }
         return *this;
     }
     LogBuffer& print(const char* format) { return print("%s", format); }
 
-    std::size_t capacity() const { return static_cast<size_t>(_buf + MAX_CAPACITY - _wptr); }
-    char* str() { return _buf; }
-    const char* c_str() const { return _buf; }
+    std::size_t capacity() const { return static_cast<size_t>(_buf.data() + MAX_CAPACITY - _wptr); }
+    char* str() { return _buf.data(); }
+    const char* c_str() const { return _buf.data(); }
 
 private:
-    char* _wptr;
-    char _buf[MAX_CAPACITY];
+    std::array<char, MAX_CAPACITY> _buf;
+    char*                          _wptr;
 };
 
 using Logger = std::function<void(int, const char*, int, const char* msg)>;
 
 inline void DefaultLogger(int level, const char* /*filename*/, int /*lineno*/, const char* msg)
 {
-    std::cout << "[" << utility::getCurrentTime() << "]" << Log::levelString(level) << " " << msg;
+    std::cout << "[" << utility::getCurrentTime() << "]" << Log::levelString(static_cast<std::size_t>(level)) << " " << msg;
     std::cout << std::endl;
 }
 

tests/integration/TestKafkaConsumer.cc

@@ -1564,15 +1564,15 @@ TEST(KafkaConsumer, OffsetsForTime)
 
     using namespace std::chrono;
 
-    constexpr int MESSAGES_NUM = 5;
+    constexpr std::size_t MESSAGES_NUM = 5;
 
     std::vector<time_point<system_clock>>     checkPoints;
     std::vector<kafka::TopicPartitionOffsets> expectedOffsets;
 
     std::cout << "Produce messages:" << std::endl;
     {
         kafka::clients::KafkaProducer producer(KafkaTestUtility::GetKafkaClientCommonConfig());
-        for (int i = 0; i < MESSAGES_NUM; ++i)
+        for (std::size_t i = 0; i < MESSAGES_NUM; ++i)
         {
             checkPoints.emplace_back(system_clock::now());
 
@@ -1602,7 +1602,7 @@ TEST(KafkaConsumer, OffsetsForTime)
     {
         kafka::clients::KafkaConsumer consumer(KafkaTestUtility::GetKafkaClientCommonConfig());
         consumer.subscribe({topic1, topic2});
-        for (int i = 0; i < MESSAGES_NUM; ++i)
+        for (std::size_t i = 0; i < MESSAGES_NUM; ++i)
         {
             const auto timepoint = checkPoints[i];
             const auto expected  = expectedOffsets[i];
@@ -1619,7 +1619,7 @@ TEST(KafkaConsumer, OffsetsForTime)
         kafka::clients::KafkaConsumer consumer(KafkaTestUtility::GetKafkaClientCommonConfig());
 
         // Here we doesn't subsribe to topic1 or topic2 (the result is undefined)
-        for (int i = 0; i < MESSAGES_NUM; ++i)
+        for (std::size_t i = 0; i < MESSAGES_NUM; ++i)
         {
             try
             {
@@ -1654,7 +1654,7 @@ TEST(KafkaConsumer, OffsetsForTime)
         kafka::clients::KafkaConsumer consumer(KafkaTestUtility::GetKafkaClientCommonConfig());
         consumer.subscribe({topic1, topic2});
 
-        for (int i = 0; i < MESSAGES_NUM; ++i)
+        for (std::size_t i = 0; i < MESSAGES_NUM; ++i)
         {
             const auto timepoint = checkPoints[i];
             const auto validTp   = kafka::TopicPartition{topic1, partition1};

tests/integration/TestKafkaProducer.cc

@@ -347,7 +347,7 @@ TEST(KafkaProducer, ThreadCount)
 TEST(KafkaProducer, MessageDeliveryCallback)
 {
     // Prepare messages to test
-    const std::vector<std::tuple<std::string, std::string, int>> messages = {
+    const std::vector<std::tuple<std::string, std::string, kafka::clients::producer::ProducerRecord::Id>> messages = {
         {"key1", "value1", 1},
         {"key2", "value2", 2},
         {"key3", "value3", 3},
@@ -379,7 +379,7 @@ TEST(KafkaProducer, MessageDeliveryCallback)
             auto record = kafka::clients::producer::ProducerRecord(topic, partition,
                                                                    kafka::Key(std::get<0>(msg).c_str(), std::get<0>(msg).size()),
                                                                    kafka::Value(std::get<1>(msg).c_str(), std::get<1>(msg).size()),
-                                                                 std::get<2>(msg));
+                                                                   std::get<2>(msg));
             std::cout << "[" <<kafka::utility::getCurrentTime() << "] ProducerRecord: " << record.toString() << std::endl;
             producer.send(record, drCallback);
         }
@@ -397,7 +397,7 @@ TEST(KafkaProducer, MessageDeliveryCallback)
 TEST(KafkaProducer, DeliveryCallback_ManuallyPollEvents)
 {
     // Prepare messages to test
-    const std::vector<std::tuple<std::string, std::string, int>> messages = {
+    const std::vector<std::tuple<std::string, std::string, kafka::clients::producer::ProducerRecord::Id>> messages = {
         {"key1", "value1", 1},
         {"key2", "value2", 2},
         {"key3", "value3", 3},

tests/integration/TestKafkaRecoverableProducer.cc

@@ -93,8 +93,8 @@ TEST(KafkaRecoverableProducer, MockFatalError)
     // Prepare messages to send
     static constexpr int MSG_NUM = 50;
     std::mutex messagesMutex;
-    std::list<int> messagesToSend;
-    for (int i = 0; i < MSG_NUM; ++i) messagesToSend.push_back(i);
+    std::list<kafka::clients::producer::ProducerRecord::Id> messagesToSend;
+    for (std::size_t i = 0; i < MSG_NUM; ++i) messagesToSend.push_back(i);
 
     int sendCount     = 0;
     int deliveryCount = 0;
@@ -122,7 +122,7 @@ TEST(KafkaRecoverableProducer, MockFatalError)
                              // Would resend the message
                              if (error) {
                                 std::lock_guard<std::mutex> lock(messagesMutex);
-                                messagesToSend.push_front(std::stoi(*payload));
+                                messagesToSend.push_front(static_cast<kafka::clients::producer::ProducerRecord::Id>(std::stoi(*payload)));
                              }
 
                              ++deliveryCount;