small fixes

include/KafkaAvro_Provider.h

@@ -27,7 +27,7 @@ protected:
 	std::thread pollThread; /**< Thread for polling Kafka events. */
 	bool pollThreadFlag = false; /**< Flag to indicate if the poll thread is running. */
 	std::mutex closeMutex; /**< Mutex for thread-safe closing of the provider. */
-
+	bool _open = false; /**< internal variable to check the state of the provider */
 	class KafkaDeliveryReportCb; /**< Forward declaration of the KafkaDeliveryReportCb class. */
 	KafkaDeliveryReportCb *dr_cb = nullptr; /**< Pointer to the Kafka delivery report callback object. */
 	class KafkaConnectionCallback; /**< Forward declaration of the KafkaConnectionCallback class. */

src/KafkaAvro_Provider.cpp

@@ -25,6 +25,7 @@ public:
 		this->prov = prov;
 	}
 	void event_cb(RdKafka::Event &event) {
+		if(!this->prov->isOpen()) return;
 		if (event.err() == RdKafka::ERR__ALL_BROKERS_DOWN) {
 			time_t now = time(nullptr);
 			std::cerr << "["
@@ -140,6 +141,7 @@ int KafkaAvroProvider::open() {
 				&KafkaAvroProvider::pollingThreadFunction, this); // start polling
 		pollThreadFlag = true;
 	}
+	this->_open = true;
 	return 1;
 }
 
@@ -165,13 +167,15 @@ int KafkaAvroProvider::close() {
 				<< "]\t[KafkaAvro Provider]\t" << "Closed" << std::endl;
 		delete producer;
 		producer = nullptr;
+		this->_open = false;
 		this->closeMutex.unlock();
+		return 1;
 	}
-	return 1;
+	return -1;
 }
 
 bool KafkaAvroProvider::isOpen() {
-	return this->producer != nullptr;
+	return this->_open;
 }
 
 void KafkaAvroProvider::pollingThreadFunction() {