Add error notification for some testcases

kenneth-jia · kenneth-jia · commit ea5d02c77bb7 · 2021-08-24T15:52:56.000+08:00
diff --git a/include/kafka/Error.h b/include/kafka/Error.h
@@ -73,6 +73,23 @@ class Error
                  (_rkError ? rd_kafka_error_string(_rkError.get()) : rd_kafka_err2str(_respErr));
     }
 
+    /**
+     * Detailed error string.
+     */
+    std::string     toString()     const
+    {
+        std::ostringstream oss;
+
+        oss << rd_kafka_err2str(static_cast<rd_kafka_resp_err_t>(value())) << " [" << value() << "]";
+
+        if (auto fatal = isFatal())         oss << " | " << (*fatal ? "fatal" : "non-fatal");
+        if (transactionRequiresAbort())     oss << " | transaction-requires-abort";
+        if (auto retriable = isRetriable()) oss << " | " << (*retriable ? "retriable" : "non-retriable");
+        if (_message)                       oss << " | " << *_message;
+
+        return oss.str();
+    }
+
     /**
      * Fatal error indicates that the client instance is no longer usable.
      */
@@ -89,6 +106,18 @@ class Error
         return _rkError ? rd_kafka_error_is_retriable(_rkError.get()) : Optional<bool>{};
     }
 
+    /**
+     * Show whether the error is an abortable transaction error.
+     *
+     * Note:
+     * 1. Only valid for transactional API.
+     * 2. If `true`, the producer must call `abortTransaction` and start a new transaction with `beginTransaction` to proceed with transactions.
+     */
+    bool transactionRequiresAbort() const
+    {
+        return _rkError ? rd_kafka_error_txn_requires_abort(_rkError.get()) : false;
+    }
+
 private:
     rd_kafka_error_shared_ptr _rkError;     // For error with rich info
     rd_kafka_resp_err_t       _respErr{};   // For error with a simple response code
diff --git a/include/kafka/KafkaException.h b/include/kafka/KafkaException.h
@@ -38,8 +38,7 @@ class KafkaException: public std::exception
      */
     const char* what() const noexcept override
     {
-        _what = Utility::getLocalTimeString(_when)  + ": " + _error->message() + " [" + std::to_string(_error->value())
-                  + "] (" + std::string(_filename) + ":" + std::to_string(_lineno) + ")";
+        _what = Utility::getLocalTimeString(_when) + ": " + _error->toString() + " (" + std::string(_filename) + ":" + std::to_string(_lineno) + ")";
         return _what.c_str();
     }
 
diff --git a/tests/integration/TestKafkaConsumer.cc b/tests/integration/TestKafkaConsumer.cc
@@ -1861,12 +1861,46 @@ TEST(KafkaAutoCommitConsumer, AutoCreateTopics)
     KafkaAutoCommitConsumer consumer(KafkaTestUtility::GetKafkaClientCommonConfig()
                                      .put("allow.auto.create.topics", "true"));
 
+    // The error would be triggered while consumer tries to subscribe a non-existed topic.
+    consumer.setErrorCallback([](const Error& error) {
+                                  std::cout << "consumer met an error: " << error.toString() << std::endl;
+                                  EXPECT_EQ(RD_KAFKA_RESP_ERR_UNKNOWN_TOPIC_OR_PART, error.value());
+                              });
+
     // Subscribe topics, but would never make it!
-    EXPECT_KAFKA_THROW(consumer.subscribe({topic}, Consumer::NullRebalanceCallback, std::chrono::seconds(10)), RD_KAFKA_RESP_ERR__TIMED_OUT);
+    EXPECT_KAFKA_THROW(consumer.subscribe({topic}, Consumer::NullRebalanceCallback, std::chrono::seconds(10)),
+                       RD_KAFKA_RESP_ERR__TIMED_OUT);
 
     EXPECT_TRUE(consumer.assignment().empty());
 }
 
+TEST(KafkaAutoCommitConsumer, CreateTopicAfterSubscribe)
+{
+    const Topic topic = Utility::getRandomString();
+
+    auto createTopicAfterSeconds = [topic](int seconds) {
+        std::this_thread::sleep_for(std::chrono::seconds(seconds));
+        KafkaTestUtility::CreateKafkaTopic(topic, 1, 1);
+    };
+
+    KafkaAutoCommitConsumer consumer(KafkaTestUtility::GetKafkaClientCommonConfig());
+
+    // The error would be triggered while consumer tries to subscribe a non-existed topic.
+    consumer.setErrorCallback([](const Error& error) {
+                                 KafkaTestUtility::DumpError(error);
+                                 EXPECT_EQ(RD_KAFKA_RESP_ERR_UNKNOWN_TOPIC_OR_PART, error.value());
+                              });
+
+    // The topic would be created after 5 seconds
+    KafkaTestUtility::JoiningThread consumer1Thread(createTopicAfterSeconds, 5);
+
+    std::cout << "[" << Utility::getCurrentTime() << "] Consumer will subscribe" << std::endl;
+    EXPECT_KAFKA_NO_THROW(consumer.subscribe({topic}));
+    std::cout << "[" << Utility::getCurrentTime() << "] Consumer just subscribed" << std::endl;
+
+    EXPECT_FALSE(consumer.assignment().empty());
+}
+
 TEST(KafkaAutoCommitConsumer, CooperativeRebalance)
 {
     constexpr int NUM_TOPICS     = 3;
diff --git a/tests/robustness/TestKafkaConsumer.cc b/tests/robustness/TestKafkaConsumer.cc
@@ -42,6 +42,7 @@ TEST(KafkaManualCommitConsumer, AlwaysFinishClosing_ManuallyPollEvents)
     {
         // Start a consumer (which need to call `pollEvents()` to trigger the commit callback)
         KafkaManualCommitConsumer consumer(props, KafkaClient::EventsPollingOption::Manual);
+        consumer.setErrorCallback(KafkaTestUtility::DumpError);
         std::cout << "[" << Utility::getCurrentTime() << "] " << consumer.name() << " started" << std::endl;
 
         // Subscribe the topic
@@ -106,6 +107,7 @@ TEST(KafkaManualCommitConsumer, CommitOffsetWhileBrokersStop)
     {
         // Start a consumer
         KafkaManualCommitConsumer consumer(props);
+        consumer.setErrorCallback(KafkaTestUtility::DumpError);
         std::cout << "[" << Utility::getCurrentTime() << "] " << consumer.name() << " started" << std::endl;
 
         // Subscribe th topic
@@ -165,6 +167,7 @@ TEST(KafkaAutoCommitConsumer, BrokerStopBeforeConsumerStart)
 
     // Start the consumer
     KafkaAutoCommitConsumer consumer(props);
+    consumer.setErrorCallback(KafkaTestUtility::DumpError);
     std::cout << "[" << Utility::getCurrentTime() << "] " << consumer.name() << " started" << std::endl;
 
 
@@ -202,21 +205,21 @@ TEST(KafkaAutoCommitConsumer, BrokerStopBeforeConsumerStart)
 
 TEST(KafkaAutoCommitConsumer, BrokerStopBeforeSubscription)
 {
+    const Topic topic = Utility::getRandomString();
+    KafkaTestUtility::CreateKafkaTopic(topic, 5, 3);
+
     // Consumer properties
     const auto props = KafkaTestUtility::GetKafkaClientCommonConfig()
                             .put(ConsumerConfig::SESSION_TIMEOUT_MS,   "30000")
                             .put(ConsumerConfig::ENABLE_PARTITION_EOF, "true");
 
     // Start the consumer
     KafkaAutoCommitConsumer consumer(props);
+    consumer.setErrorCallback(KafkaTestUtility::DumpError);
     std::cout << "[" << Utility::getCurrentTime() << "] " << consumer.name() << " started" << std::endl;
 
     // Pause the brokers for a while
     auto asyncTask = KafkaTestUtility::PauseBrokersForAWhile(std::chrono::seconds(5));
-
-    const Topic topic = Utility::getRandomString();
-    KafkaTestUtility::CreateKafkaTopic(topic, 5, 3);
-
     TopicPartitions assignment;
     // In some corner cases, the assigned partitions might be empty (due to "Local: Broker node update" error), and we'll retry
     while (assignment.empty())
@@ -261,6 +264,7 @@ TEST(KafkaAutoCommitConsumer, BrokerStopBeforeSeek)
 
     // Start the consumer
     KafkaAutoCommitConsumer consumer(props);
+    consumer.setErrorCallback(KafkaTestUtility::DumpError);
     std::cout << "[" << Utility::getCurrentTime() << "] " << consumer.name() << " started" << std::endl;
 
     // Subscribe the topic
@@ -325,6 +329,7 @@ TEST(KafkaAutoCommitConsumer, BrokerStopDuringMsgPoll)
 
     // Start the consumer
     KafkaAutoCommitConsumer consumer(props);
+    consumer.setErrorCallback(KafkaTestUtility::DumpError);
     std::cout << "[" << Utility::getCurrentTime() << "] " << consumer.name() << " started" << std::endl;
 
     // Subscribe the topic
diff --git a/tests/robustness/TestKafkaProducer.cc b/tests/robustness/TestKafkaProducer.cc
@@ -32,6 +32,7 @@ TEST(KafkaSyncProducer, RecordTimestamp)
 
     // Prepare a producer
     KafkaSyncProducer producer(KafkaTestUtility::GetKafkaClientCommonConfig());
+    producer.setErrorCallback(KafkaTestUtility::DumpError);
 
     constexpr int TIME_LAPSE_THRESHOLD_MS = 1000;
     using namespace std::chrono;
@@ -124,6 +125,7 @@ TEST(KafkaAsyncProducer, NoMissedDeliveryCallback)
 
     {
         KafkaAsyncProducer producer(KafkaTestUtility::GetKafkaClientCommonConfig().put(ProducerConfig::MESSAGE_TIMEOUT_MS, "5000"));
+        producer.setErrorCallback(KafkaTestUtility::DumpError);
 
         // Pause the brokers for a while
         auto asyncTask = KafkaTestUtility::PauseBrokersForAWhile(std::chrono::seconds(5));
@@ -157,6 +159,7 @@ TEST(KafkaAsyncProducer, MightMissDeliveryCallbackIfCloseWithLimitedTimeout)
     std::size_t deliveryCount = 0;
     {
         KafkaAsyncProducer producer(KafkaTestUtility::GetKafkaClientCommonConfig());
+        producer.setErrorCallback(KafkaTestUtility::DumpError);
 
         KafkaTestUtility::PauseBrokers();
 
@@ -202,6 +205,7 @@ TEST(KafkaAsyncProducer, BrokerStopWhileSendingMessages)
     std::size_t deliveryCount = 0;
     {
         KafkaAsyncProducer producer(KafkaTestUtility::GetKafkaClientCommonConfig());
+        producer.setErrorCallback(KafkaTestUtility::DumpError);
 
         // Pause the brokers for a while (shorter then the default "MESSAGE_TIMEOUT_MS" for producer, which is 10 seconds)
         auto asyncTask = KafkaTestUtility::PauseBrokersForAWhile(std::chrono::seconds(5));
@@ -249,6 +253,7 @@ TEST(KafkaAsyncProducer, Send_AckTimeout)
     {
         KafkaAsyncProducer producer(KafkaTestUtility::GetKafkaClientCommonConfig()
                                         .put(ProducerConfig::MESSAGE_TIMEOUT_MS, "3000")); // If with no response, the delivery would fail in a short time
+        producer.setErrorCallback(KafkaTestUtility::DumpError);
 
         // Pause the brokers for a while
         auto asyncTask = KafkaTestUtility::PauseBrokersForAWhile(std::chrono::seconds(5));
@@ -287,6 +292,7 @@ TEST(KafkaAsyncProducer, ManuallyPollEvents_AckTimeout)
         KafkaAsyncProducer producer(KafkaTestUtility::GetKafkaClientCommonConfig()
                                         .put(ProducerConfig::MESSAGE_TIMEOUT_MS, "3000"), // If with no response, the delivery would fail in a short time
                                     KafkaClient::EventsPollingOption::Manual);            // Manually call `pollEvents()`
+        producer.setErrorCallback(KafkaTestUtility::DumpError);
 
         // Pause the brokers for a while
         auto asyncTask = KafkaTestUtility::PauseBrokersForAWhile(std::chrono::seconds(5));
@@ -333,6 +339,7 @@ TEST(KafkaAsyncProducer, ManuallyPollEvents_AlwaysFinishClosing)
         KafkaAsyncProducer producer(KafkaTestUtility::GetKafkaClientCommonConfig()
                                         .put(ProducerConfig::MESSAGE_TIMEOUT_MS, "3000"), // If with no response, the delivery would fail in a short time
                                     KafkaClient::EventsPollingOption::Manual);            // Manually call `pollEvents()`
+        producer.setErrorCallback(KafkaTestUtility::DumpError);
 
         // Pause the brokers for a while
         auto asyncTask = KafkaTestUtility::PauseBrokersForAWhile(std::chrono::seconds(5));
@@ -362,6 +369,7 @@ TEST(KafkaSyncProducer, Send_AckTimeout)
     KafkaTestUtility::CreateKafkaTopic(topic, 5, 3);
 
     KafkaSyncProducer producer(KafkaTestUtility::GetKafkaClientCommonConfig().put(ProducerConfig::MESSAGE_TIMEOUT_MS, "3000"));
+    producer.setErrorCallback(KafkaTestUtility::DumpError);
 
     // Pause the brokers for a while
     auto asyncTask = KafkaTestUtility::PauseBrokersForAWhile(std::chrono::seconds(5));
diff --git a/tests/robustness/TestTransaction.cc b/tests/robustness/TestTransaction.cc
@@ -26,6 +26,7 @@ TEST(Transaction, DeliveryFailure)
         KafkaAsyncProducer producer(KafkaTestUtility::GetKafkaClientCommonConfig()
                                       .put(ProducerConfig::MESSAGE_TIMEOUT_MS, "3000")  // The delivery would fail in a short timeout
                                       .put(ProducerConfig::TRANSACTIONAL_ID, transactionId));
+        producer.setErrorCallback(KafkaTestUtility::DumpError);
 
         std::cout << "[" << Utility::getCurrentTime() << "] Producer created." << std::endl;
 
diff --git a/tests/unit/TestKafkaException.cc b/tests/unit/TestKafkaException.cc
@@ -48,7 +48,7 @@ TEST(KafkaException, Basic)
     }
     catch (const Kafka::KafkaException& e)
     {
-        std::regex reMatch(R"(.*something wrong here \[7\] \(some_filename:100\))");
+        std::regex reMatch(R"(.*Broker: Request timed out \[7\] \| something wrong here \(some_filename:100\))");
         EXPECT_TRUE(std::regex_match(e.what(), reMatch));
     }
 }
diff --git a/tests/utils/TestUtility.h b/tests/utils/TestUtility.h
@@ -9,6 +9,7 @@
 #include <boost/algorithm/string.hpp>
 
 #include <cstdlib>
+#include <functional>
 #include <list>
 #include <regex>
 #include <signal.h>
@@ -19,31 +20,21 @@
     do {                                                                \
         try {                                                           \
             expr;                                                       \
-        } catch (const KafkaException& e) {                                                                  \
-            const auto& error = e.error();                                                                   \
-            std::cout << "Exception caught: " << e.what()                                                    \
-                << (error.isFatal() ? (*error.isFatal() ? ", fatal" : ", non-fatal") : "")                   \
-                << (error.isRetriable() ? (*error.isRetriable() ? ",  retriable" : ", non-retriable") : "")  \
-                << std::endl;                                                                                \
-            EXPECT_EQ(err, e.error().value());                                                               \
-            break;                                                                                           \
+        } catch (const KafkaException& e) {                             \
+            std::cout << "Exception caught: " << e.what() << std::endl; \
+            EXPECT_EQ(err, e.error().value());                          \
+            break;                                                      \
         } catch (...){                                                  \
         }                                                               \
         EXPECT_FALSE(true);                                             \
     } while(false)
 
-#define EXPECT_KAFKA_NO_THROW(expr)   \
-    try {                             \
-        expr;                         \
-    } catch (...){                    \
-        EXPECT_FALSE(true);           \
-    }
-
-#define EXPECT_KAFKA_NO_THROW(expr)   \
-    try {                             \
-        expr;                         \
-    } catch (...){                    \
-        EXPECT_FALSE(true);           \
+#define EXPECT_KAFKA_NO_THROW(expr)                                 \
+    try {                                                           \
+        expr;                                                       \
+    } catch (const KafkaException& e) {                             \
+        std::cerr << "Exception caught: " << e.what() << std::endl; \
+        EXPECT_FALSE(true);                                         \
     }
 
 #define RETRY_FOR_ERROR(expr, errToRetry, maxRetries)                           \
@@ -69,9 +60,19 @@
         }                                                           \
     }
 
+
 namespace Kafka = KAFKA_API;
 
 namespace KafkaTestUtility {
+
+inline void
+DumpError(const Kafka::Error& error)
+{
+    // https://en.wikipedia.org/wiki/ANSI_escape_code
+    std::cerr << "\033[1;31m" << "[" << Kafka::Utility::getCurrentTime() << "] ==> Met Error: " << "\033[0m";
+    std::cerr << "\033[4;35m" << error.toString() << "\033[0m" << std::endl;
+};
+
 inline void
 PrintDividingLine(const std::string& description = "")
 {

Original file line number	Diff line number	Diff line change
`@@ -38,8 +38,7 @@ class KafkaException: public std::exception`
`38`	`38`	`*/`
`39`	`39`	`const char* what() const noexcept override`
`40`	`40`	`{`
`41`		`- _what = Utility::getLocalTimeString(_when) + ": " + _error->message() + " [" + std::to_string(_error->value())`
`42`		`- + "] (" + std::string(_filename) + ":" + std::to_string(_lineno) + ")";`
	`41`	`+ _what = Utility::getLocalTimeString(_when) + ": " + _error->toString() + " (" + std::string(_filename) + ":" + std::to_string(_lineno) + ")";`
`43`	`42`	`return _what.c_str();`
`44`	`43`	`}`
`45`	`44`
Original file line number	Diff line number	Diff line change
`@@ -48,7 +48,7 @@ TEST(KafkaException, Basic)`
`48`	`48`	`}`
`49`	`49`	`catch (const Kafka::KafkaException& e)`
`50`	`50`	`{`
`51`		`- std::regex reMatch(R"(.*something wrong here \[7\] \(some_filename:100\))");`
	`51`	`+ std::regex reMatch(R"(.*Broker: Request timed out \[7\] \\| something wrong here \(some_filename:100\))");`
`52`	`52`	`EXPECT_TRUE(std::regex_match(e.what(), reMatch));`
`53`	`53`	`}`
`54`	`54`	`}`