Upgrade with librdkafka v1.7.0

Author: kenneth-jia

.github/workflows/kafka_api_ci_tests.yml

@@ -5,7 +5,7 @@ on: [push, pull_request]
 env:
   KAFKA_SRC_LINK:     https://archive.apache.org/dist/kafka/2.6.0/kafka_2.13-2.6.0.tgz
   CPU_CORE_NUM:       2
-  LIBRDKAFKA_VERSION: 1.6.0
+  LIBRDKAFKA_VERSION: 1.7.0
   BUILD_SUB_DIR:      build/sub-build
 
 jobs:
@@ -179,7 +179,7 @@ jobs:
           fi
 
       - name: Test
-        timeout-minutes: 15
+        timeout-minutes: 20
         run: |
           cd ${BUILD_SUB_DIR}
 

README.md

@@ -4,7 +4,7 @@
 
 The `Modern C++ based Kafka API` (`modern-cpp-kafka`) 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` is compatible with `librdkafka` **v1.6.0**.
+- By now, `modern-cpp-kafka` is compatible with `librdkafka` **v1.7.0**.
 
 ```
 KAFKA is a registered trademark of The Apache Software Foundation and

tests/integration/TestTransaction.cc

@@ -234,7 +234,7 @@ TEST(Transaction, CatchException)
 
         producer.beginTransaction();
 
-        EXPECT_KAFKA_THROW(producer.commitTransaction(), RD_KAFKA_RESP_ERR_INVALID_TXN_STATE);
+        producer.commitTransaction();
     }
 }
 

tests/robustness/TestAdminClient.cc

@@ -34,7 +34,7 @@ TEST(AdminClient, BrokersTimeout)
         std::cout << "[" << Utility::getCurrentTime() << "] will ListTopics" << std::endl;
         {
             auto listResult = adminClient.listTopics(std::chrono::seconds(1));
-            std::cout << "[" << Utility::getCurrentTime() << "] ListTopics: result[" << listResult.message() << "]" << std::endl;
+            std::cout << "[" << Utility::getCurrentTime() << "] ListTopics: result[" << listResult.message() << "]. Result: " << listResult.message() << std::endl;
             EXPECT_TRUE(listResult.errorCode().value() == RD_KAFKA_RESP_ERR__TRANSPORT || listResult.errorCode().value() == RD_KAFKA_RESP_ERR__TIMED_OUT);
             EXPECT_EQ(0, listResult.topics.size());
         }

tests/unit/TestKafkaClientDefaultProperties.cc

@@ -107,7 +107,7 @@ TEST(KafkaClient, KafkaConsumerDefaultProperties)
             { Kafka::ConsumerConfig::ENABLE_PARTITION_EOF,  "false"      },
             { Kafka::ConsumerConfig::MAX_POLL_RECORDS,      "500"        },
             { Kafka::ConsumerConfig::QUEUED_MIN_MESSAGES,   "100000"     },
-            { Kafka::ConsumerConfig::SESSION_TIMEOUT_MS,    "10000"      },
+            { Kafka::ConsumerConfig::SESSION_TIMEOUT_MS,    "45000"      },
             { Kafka::ConsumerConfig::SOCKET_TIMEOUT_MS,     "60000"      },
             { Kafka::ConsumerConfig::SECURITY_PROTOCOL,     "plaintext"  },
             { "enable.auto.commit",                         "false"      },

tests/utils/TestUtility.h

@@ -172,9 +172,10 @@ CreateKafkaTopic(const Kafka::Topic& topic, int numPartitions, int replicationFa
 {
     Kafka::AdminClient adminClient(GetKafkaClientCommonConfig());
     auto createResult = adminClient.createTopics({topic}, numPartitions, replicationFactor);
+    std::cout << "[" << Kafka::Utility::getCurrentTime() << "] " << __FUNCTION__ << ": create topic[" << topic << "] "
+       << "with numPartitions[" << numPartitions << "], replicationFactor[" << replicationFactor << "]. Result: " << createResult.message() << std::endl;
     ASSERT_FALSE(createResult.errorCode());
-    std::cout << "[" << Kafka::Utility::getCurrentTime() << "] " << __FUNCTION__ << ": topic[" << topic << "] created with numPartitions[" << numPartitions << "], replicationFactor[" << replicationFactor << "]." << std::endl;
-    std::this_thread::sleep_for(std::chrono::seconds(1));
+    std::this_thread::sleep_for(std::chrono::seconds(5));
 }
 
 class JoiningThread {
@@ -220,20 +221,23 @@ signalToAllBrokers(int sig)
     else if (sig == SIGCONT)
     {
         std::cout << "[" << Kafka::Utility::getCurrentTime() << "] Brokers resumed"  << std::endl;
-        std::this_thread::sleep_for(std::chrono::seconds(1));
     }
 }
 
 inline void
 PauseBrokers()
 {
+    constexpr int WAIT_AFTER_PAUSE_MS = 100;
     signalToAllBrokers(SIGSTOP);
+    std::this_thread::sleep_for(std::chrono::milliseconds(WAIT_AFTER_PAUSE_MS));
 }
 
 inline void
 ResumeBrokers()
 {
+    constexpr int WAIT_AFTER_RESUME_SEC = 5;
     signalToAllBrokers(SIGCONT);
+    std::this_thread::sleep_for(std::chrono::seconds(WAIT_AFTER_RESUME_SEC));
 }
 
 inline std::shared_ptr<JoiningThread>