Refine test KafkaRecoverableProducer.MockFatalError a bit

kenneth-jia · kenneth-jia · commit 5fa8bc01577a · 2021-10-20T16:07:35.000+08:00
diff --git a/include/kafka/addons/KafkaRecoverableProducer.h b/include/kafka/addons/KafkaRecoverableProducer.h
@@ -31,10 +31,10 @@ class KafkaRecoverableProducer
     {
         std::lock_guard<std::mutex> lock(_producerMutex);
 
-        _producer->close();
-
         _running = false;
         if (_pollThread.joinable()) _pollThread.join();
+
+        _producer->close();
     }
 
     /**
diff --git a/tests/integration/TestKafkaRecoverableProducer.cc b/tests/integration/TestKafkaRecoverableProducer.cc
@@ -133,6 +133,7 @@ TEST(KafkaRecoverableProducer, MockFatalError)
             std::this_thread::sleep_for(std::chrono::milliseconds(1));
         }
     }
+    EXPECT_EQ(sendCount, deliveryCount);
 
     // Prepare a consumer
     const auto consumerProps = KafkaTestUtility::GetKafkaClientCommonConfig().put(kafka::clients::consumer::Config::AUTO_OFFSET_RESET, "earliest");

Original file line number	Diff line number	Diff line change
`@@ -31,10 +31,10 @@ class KafkaRecoverableProducer`
`31`	`31`	`{`
`32`	`32`	`std::lock_guard<std::mutex> lock(_producerMutex);`
`33`	`33`
`34`		`- _producer->close();`
`35`		`-`
`36`	`34`	`_running = false;`
`37`	`35`	`if (_pollThread.joinable()) _pollThread.join();`
	`36`	`+`
	`37`	`+ _producer->close();`
`38`	`38`	`}`
`39`	`39`
`40`	`40`	`/**`
Original file line number	Diff line number	Diff line change
`@@ -133,6 +133,7 @@ TEST(KafkaRecoverableProducer, MockFatalError)`
`133`	`133`	`std::this_thread::sleep_for(std::chrono::milliseconds(1));`
`134`	`134`	`}`
`135`	`135`	`}`
	`136`	`+ EXPECT_EQ(sendCount, deliveryCount);`
`136`	`137`
`137`	`138`	`// Prepare a consumer`
`138`	`139`	`const auto consumerProps = KafkaTestUtility::GetKafkaClientCommonConfig().put(kafka::clients::consumer::Config::AUTO_OFFSET_RESET, "earliest");`