A new parameter for producer.send(...), -- ActionWhileQueueIsFull

kenneth-jia · kenneth-jia · commit 90d25442aa2e · 2021-10-20T16:07:35.000+08:00
diff --git a/include/kafka/KafkaProducer.h b/include/kafka/KafkaProducer.h
@@ -69,6 +69,11 @@ class KafkaProducer: public KafkaClient
      */
     enum class SendOption { NoCopyRecordValue, ToCopyRecordValue };
 
+    /**
+     * Choose the action while the sending buffer is full.
+     */
+    enum class ActionWhileQueueIsFull { Block, NoBlock };
+
     /**
      * Asynchronously send a record to a topic.
      *
@@ -87,7 +92,10 @@ class KafkaProducer: public KafkaClient
      *   Broker errors,
      *     - [Error Codes] (https://cwiki.apache.org/confluence/display/KAFKA/A+Guide+To+The+Kafka+Protocol#AGuideToTheKafkaProtocol-ErrorCodes)
      */
-    void send(const producer::ProducerRecord& record, const producer::Callback& deliveryCb, SendOption option = SendOption::NoCopyRecordValue);
+    void send(const producer::ProducerRecord& record,
+              const producer::Callback&       deliveryCb,
+              SendOption                      option = SendOption::NoCopyRecordValue,
+              ActionWhileQueueIsFull          action = ActionWhileQueueIsFull::Block);
 
     /**
      * Asynchronously send a record to a topic.
@@ -107,9 +115,13 @@ class KafkaProducer: public KafkaClient
      *   Broker errors,
      *     - [Error Codes] (https://cwiki.apache.org/confluence/display/KAFKA/A+Guide+To+The+Kafka+Protocol#AGuideToTheKafkaProtocol-ErrorCodes)
      */
-    void send(const producer::ProducerRecord& record, const producer::Callback& deliveryCb, Error& error, SendOption option = SendOption::NoCopyRecordValue)
+    void send(const producer::ProducerRecord& record,
+              const producer::Callback&       deliveryCb,
+              Error&                          error,
+              SendOption                      option = SendOption::NoCopyRecordValue,
+              ActionWhileQueueIsFull          action = ActionWhileQueueIsFull::Block)
     {
-        try { send(record, deliveryCb, option); } catch (const KafkaException& e) { error = e.error(); }
+        try { send(record, deliveryCb, option, action); } catch (const KafkaException& e) { error = e.error(); }
     }
 
     /**
@@ -183,8 +195,6 @@ class KafkaProducer: public KafkaClient
         const producer::Callback                     _deliveryCb;
     };
 
-    enum class ActionWhileQueueIsFull { Block, NoBlock };
-
     // Validate properties (and fix it if necesary)
     static Properties validateAndReformProperties(const Properties& properties);
 
@@ -320,10 +330,11 @@ KafkaProducer::deliveryCallback(rd_kafka_t* rk, const rd_kafka_message_t* rkmsg,
 inline void
 KafkaProducer::send(const producer::ProducerRecord& record,
                     const producer::Callback&       deliveryCb,
-                    SendOption                      option)
+                    SendOption                      option,
+                    ActionWhileQueueIsFull          action)
 {
     auto deliveryCbOpaque = std::make_unique<DeliveryCbOpaque>(record.id(), deliveryCb);
-    auto queueFullAction  = (isWithAutoEventsPolling() ? ActionWhileQueueIsFull::Block : ActionWhileQueueIsFull::NoBlock);
+    auto queueFullAction  = (isWithAutoEventsPolling() ? action : ActionWhileQueueIsFull::NoBlock);
 
     const auto* topic     = record.topic().c_str();
     const auto  partition = record.partition();
diff --git a/include/kafka/addons/KafkaRecoverableProducer.h b/include/kafka/addons/KafkaRecoverableProducer.h
@@ -205,14 +205,14 @@ class KafkaRecoverableProducer
      *   Broker errors,
      *     - [Error Codes] (https://cwiki.apache.org/confluence/display/KAFKA/A+Guide+To+The+Kafka+Protocol#AGuideToTheKafkaProtocol-ErrorCodes)
      */
-    void send(const producer::ProducerRecord& record,
-              const producer::Callback&       deliveryCb,
-              KafkaProducer::SendOption       option = KafkaProducer::SendOption::NoCopyRecordValue)
+    void send(const producer::ProducerRecord&       record,
+              const producer::Callback&             deliveryCb,
+              KafkaProducer::SendOption             option = KafkaProducer::SendOption::NoCopyRecordValue,
+              KafkaProducer::ActionWhileQueueIsFull action = KafkaProducer::ActionWhileQueueIsFull::Block)
     {
         std::lock_guard<std::mutex> lock(_producerMutex);
 
-        _producer->send(record, deliveryCb, option);
-
+        _producer->send(record, deliveryCb, option, action);
     }
 
     /**
@@ -234,14 +234,15 @@ class KafkaRecoverableProducer
      *     - [Error Codes] (https://cwiki.apache.org/confluence/display/KAFKA/A+Guide+To+The+Kafka+Protocol#AGuideToTheKafkaProtocol-ErrorCodes)
      */
 
-    void send(const producer::ProducerRecord& record,
-              const producer::Callback&       deliveryCb,
-              Error&                          error,
-              KafkaProducer::SendOption       option = KafkaProducer::SendOption::NoCopyRecordValue)
+    void send(const producer::ProducerRecord&       record,
+              const producer::Callback&             deliveryCb,
+              Error&                                error,
+              KafkaProducer::SendOption             option = KafkaProducer::SendOption::NoCopyRecordValue,
+              KafkaProducer::ActionWhileQueueIsFull action = KafkaProducer::ActionWhileQueueIsFull::Block)
     {
         std::lock_guard<std::mutex> lock(_producerMutex);
 
-        _producer->send(record, deliveryCb, error, option);
+        _producer->send(record, deliveryCb, error, option, action);
     }
 
     /**
