Modify default transactional timeout for KafkaProducer

.github/workflows/kafka_api_bazel_build.yml

@@ -9,7 +9,7 @@ on:
 env:
   KAFKA_SRC_LINK: https://dlcdn.apache.org/kafka/3.2.0/kafka_2.13-3.2.0.tgz
   CPU_CORE_NUM:   2
-  LIBRDKAFKA_TAG: v1.9.0
+  LIBRDKAFKA_TAG: v1.9.2
 
 jobs:
   kafka-api-bazel-build:

.github/workflows/kafka_api_ci_tests.yml

@@ -9,7 +9,7 @@ on:
 env:
   KAFKA_SRC_LINK: https://dlcdn.apache.org/kafka/3.2.0/kafka_2.13-3.2.0.tgz
   CPU_CORE_NUM:   2
-  LIBRDKAFKA_TAG: v1.9.0
+  LIBRDKAFKA_TAG: v1.9.2
   BUILD_SUB_DIR:  builds/sub-build
 
 jobs:
@@ -310,7 +310,8 @@ jobs:
 
           $Env:GTEST_ROOT='C:\VCPKG\INSTALLED\x86-windows\'
           $Env:BOOST_ROOT='C:\VCPKG\INSTALLED\x86-windows\'
-          $Env:LIBRDKAFKA_ROOT='C:\VCPKG\INSTALLED\x86-windows\'
+          $Env:LIBRDKAFKA_INCLUDE_DIR='C:\VCPKG\INSTALLED\x86-windows\include\'
+          $Env:LIBRDKAFKA_LIBRARY_DIR='C:\VCPKG\INSTALLED\x86-windows\lib\'
           $Env:RAPIDJSON_INCLUDE_DIRS='C:\VCPKG\INSTALLED\x86-windows\include\'
 
           cmake -B ./ -A Win32 -S ../.. "-DCMAKE_TOOLCHAIN_FILE=C:/vcpkg/scripts/buildsystems/vcpkg.cmake"

CMakeLists.txt

@@ -39,24 +39,27 @@ if (NOT BUILD_OPTION_DOC_ONLY)
     #---------------------------
     # librdkafka library
     #---------------------------
-    if (DEFINED ENV{LIBRDKAFKA_ROOT})
-        set(LIBRDKAFKA_INCLUDE_DIR $ENV{LIBRDKAFKA_ROOT}/include)
-        set(LIBRDKAFKA_LIBRARY_DIR $ENV{LIBRDKAFKA_ROOT}/lib)
+    if (DEFINED ENV{LIBRDKAFKA_INCLUDE_DIR})
+        set(LIBRDKAFKA_INCLUDE_DIR $ENV{LIBRDKAFKA_INCLUDE_DIR})
     else ()
         set(LIBRDKAFKA_INCLUDE_DIR /usr/local/include)
+    endif ()
+    if (DEFINED ENV{LIBRDKAFKA_LIBRARY_DIR})
+        set(LIBRDKAFKA_LIBRARY_DIR $ENV{LIBRDKAFKA_LIBRARY_DIR})
+    else ()
         set(LIBRDKAFKA_LIBRARY_DIR /usr/local/lib)
     endif ()
 
     if (EXISTS "${LIBRDKAFKA_INCLUDE_DIR}/librdkafka/rdkafka.h")
         message(STATUS "librdkafka include directory: ${LIBRDKAFKA_INCLUDE_DIR}")
     else ()
-        message(FATAL_ERROR "Could not find headers: librdkafka!")
+        message(FATAL_ERROR "Could not find headers for librdkafka!")
     endif ()
 
     if (EXISTS "${LIBRDKAFKA_LIBRARY_DIR}/librdkafka.a" OR EXISTS "${LIBRDKAFKA_LIBRARY_DIR}/librdkafka.so" OR EXISTS "${LIBRDKAFKA_LIBRARY_DIR}/rdkafka.lib" )
         message(STATUS "librdkafka library directory: ${LIBRDKAFKA_LIBRARY_DIR}")
     else ()
-        message(FATAL_ERROR "Could not find library: librdkafka!")
+        message(FATAL_ERROR "Could not find library for librdkafka!")
     endif ()
 
 
@@ -68,7 +71,7 @@ if (NOT BUILD_OPTION_DOC_ONLY)
         if (PTHREAD_LIB)
             message(STATUS "pthread library: ${PTHREAD_LIB}")
         else ()
-            message(FATAL_ERROR "Could not find library: pthread!")
+            message(FATAL_ERROR "Could not find library for pthread!")
         endif ()
     endif ()
 

README.md

@@ -6,7 +6,7 @@
 
 The [Modern C++ Kafka API](http://opensource.morganstanley.com/modern-cpp-kafka/doxygen/annotated.html) is a layer of C++ wrapper based on [librdkafka](https://github.com/edenhill/librdkafka) (the C part), with high quality, but more friendly to users.
 
-- By now, [modern-cpp-kafka](https://github.com/morganstanley/modern-cpp-kafka) is compatible with [librdkafka v1.9.0](https://github.com/edenhill/librdkafka/releases/tag/v1.9.0).
+- By now, [modern-cpp-kafka](https://github.com/morganstanley/modern-cpp-kafka) is compatible with [librdkafka v1.9.2](https://github.com/edenhill/librdkafka/releases/tag/v1.9.2).
 
 ```
 KAFKA is a registered trademark of The Apache Software Foundation and
@@ -65,7 +65,9 @@ Eventually, we worked out the ***modern-cpp-kafka***, -- a header-only library t
 
     * Specify library locations with environment variables
 
-        * `LIBRDKAFKA_ROOT`                 -- ***librdkafka*** headers and libraries
+        * `LIBRDKAFKA_INCLUDE_DIR`          -- ***librdkafka*** headers
+
+        * `LIBRDKAFKA_LIBRARY_DIR`          -- ***librdkafka*** libraries
 
         * `GTEST_ROOT`                      -- ***googletest*** headers and libraries
 

include/kafka/KafkaProducer.h

@@ -141,7 +141,7 @@ class KafkaProducer: public KafkaClient
     /**
      * Needs to be called before any other methods when the transactional.id is set in the configuration.
      */
-    void initTransactions(std::chrono::milliseconds timeout = std::chrono::milliseconds(KafkaProducer::DEFAULT_INIT_TRANSACTIONS_TIMEOUT_MS));
+    void initTransactions(std::chrono::milliseconds timeout = std::chrono::milliseconds::max());
 
     /**
      * Should be called before the start of each new transaction.
@@ -151,7 +151,7 @@ class KafkaProducer: public KafkaClient
     /**
      * Commit the ongoing transaction.
      */
-    void commitTransaction(std::chrono::milliseconds timeout = std::chrono::milliseconds(KafkaProducer::DEFAULT_COMMIT_TRANSACTION_TIMEOUT_MS));
+    void commitTransaction(std::chrono::milliseconds timeout = std::chrono::milliseconds::max());
 
     /**
      * Abort the ongoing transaction.
@@ -166,14 +166,6 @@ class KafkaProducer: public KafkaClient
                                   const consumer::ConsumerGroupMetadata& groupMetadata,
                                   std::chrono::milliseconds              timeout);
 
-#if COMPILER_SUPPORTS_CPP_17
-    static constexpr int DEFAULT_INIT_TRANSACTIONS_TIMEOUT_MS  = 10000;
-    static constexpr int DEFAULT_COMMIT_TRANSACTION_TIMEOUT_MS = 10000;
-#else
-    enum { DEFAULT_INIT_TRANSACTIONS_TIMEOUT_MS  = 10000 };
-    enum { DEFAULT_COMMIT_TRANSACTION_TIMEOUT_MS = 10000 };
-#endif
-
 private:
     void pollCallbacks(int timeoutMs)
     {

include/kafka/addons/KafkaRecoverableProducer.h

@@ -247,7 +247,7 @@ class KafkaRecoverableProducer
     /**
      * Needs to be called before any other methods when the transactional.id is set in the configuration.
      */
-    void initTransactions(std::chrono::milliseconds timeout = std::chrono::milliseconds(KafkaProducer::DEFAULT_INIT_TRANSACTIONS_TIMEOUT_MS))
+    void initTransactions(std::chrono::milliseconds timeout = std::chrono::milliseconds::max())
     {
         std::lock_guard<std::mutex> lock(_producerMutex);
 
@@ -267,7 +267,7 @@ class KafkaRecoverableProducer
     /**
      * Commit the ongoing transaction.
      */
-    void commitTransaction(std::chrono::milliseconds timeout = std::chrono::milliseconds(KafkaProducer::DEFAULT_COMMIT_TRANSACTION_TIMEOUT_MS))
+    void commitTransaction(std::chrono::milliseconds timeout = std::chrono::milliseconds::max())
     {
         std::lock_guard<std::mutex> lock(_producerMutex);