Message err was not set for on_ack interceptors on broker reply (#1892)

confluentinc · Aug 24, 2018 · ee8729a · ee8729a
1 parent c16998b
commit ee8729a
Show file tree

Hide file tree

Showing 3 changed files with 10 additions and 3 deletions.
diff --git a/src/rdkafka_broker.c b/src/rdkafka_broker.c
@@ -2076,7 +2076,7 @@ void rd_kafka_dr_msgq (rd_kafka_itopic_t *rkt,
 	    return;
 
         /* Call on_acknowledgement() interceptors */
-        rd_kafka_interceptors_on_acknowledgement_queue(rk, rkmq);
+        rd_kafka_interceptors_on_acknowledgement_queue(rk, rkmq, err);
 
         if ((rk->rk_conf.enabled_events & RD_KAFKA_EVENT_DR) &&
 	    (!rk->rk_conf.dr_err_only || err)) {

diff --git a/src/rdkafka_interceptor.c b/src/rdkafka_interceptor.c
@@ -453,14 +453,20 @@ rd_kafka_interceptors_on_acknowledgement (rd_kafka_t *rk,
 
 /**
  * @brief Call on_acknowledgement methods for all messages in queue.
+ *
+ * @param force_err If non-zero, sets this error on each message.
+ *
  * @locality broker thread
  */
 void
 rd_kafka_interceptors_on_acknowledgement_queue (rd_kafka_t *rk,
-                                                rd_kafka_msgq_t *rkmq) {
+                                                rd_kafka_msgq_t *rkmq,
+                                                rd_kafka_resp_err_t force_err) {
         rd_kafka_msg_t *rkm;
 
         RD_KAFKA_MSGQ_FOREACH(rkm, rkmq) {
+                if (force_err)
+                        rkm->rkm_err = force_err;
                 rd_kafka_interceptors_on_acknowledgement(rk,
                                                          &rkm->rkm_rkmessage);
         }

diff --git a/src/rdkafka_interceptor.h b/src/rdkafka_interceptor.h
@@ -50,7 +50,8 @@ rd_kafka_interceptors_on_acknowledgement (rd_kafka_t *rk,
                                           rd_kafka_message_t *rkmessage);
 void
 rd_kafka_interceptors_on_acknowledgement_queue (rd_kafka_t *rk,
-                                                rd_kafka_msgq_t *rkmq);
+                                                rd_kafka_msgq_t *rkmq,
+                                                rd_kafka_resp_err_t force_err);
 
 void rd_kafka_interceptors_on_consume (rd_kafka_t *rk,
                                        rd_kafka_message_t *rkmessage);