Fix flakyness of tests 0014 and 0085 (confluentinc#5189)

* Fix flakyness test 0085
* Errors that cause a refresh coordinator
like NOT_COORDINATOR during an offset fetch
should not be propagated to the application.

Author: emasab

src/rdkafka_partition.c

@@ -2162,6 +2162,7 @@ static rd_kafka_op_res_t rd_kafka_toppar_op_serve(rd_kafka_t *rk,
                 rd_kafka_toppar_lock(rktp);
 
                 if (rko->rko_err) {
+                        int actions;
                         rd_kafka_dbg(
                             rktp->rktp_rkt->rkt_rk, TOPIC, "OFFSET",
                             "Failed to fetch offset for "
@@ -2176,10 +2177,15 @@ static rd_kafka_op_res_t rd_kafka_toppar_op_serve(rd_kafka_t *rk,
                         rd_kafka_toppar_unlock(rktp);
 
 
-                        /* Propagate error to application */
+                        actions = rd_kafka_handle_OffsetFetch_err_action(
+                            NULL, rko->rko_err, NULL);
+                        /* Propagate error to application. Exclude
+                         * permanent errors that caused a coordinator
+                         * refresh like `NOT_COORDINATOR` */
                         if (rko->rko_err != RD_KAFKA_RESP_ERR__WAIT_COORD &&
                             rko->rko_err !=
-                                RD_KAFKA_RESP_ERR_UNSTABLE_OFFSET_COMMIT)
+                                RD_KAFKA_RESP_ERR_UNSTABLE_OFFSET_COMMIT &&
+                            !(actions & RD_KAFKA_ERR_ACTION_REFRESH))
                                 rd_kafka_consumer_err(
                                     rktp->rktp_fetchq, RD_KAFKA_NODEID_UA,
                                     rko->rko_err, 0, NULL, rktp,

src/rdkafka_request.c

@@ -1163,6 +1163,11 @@ void rd_kafka_OffsetForLeaderEpochRequest(
 }
 
 
+int rd_kafka_handle_OffsetFetch_err_action(rd_kafka_broker_t *rkb,
+                                           rd_kafka_resp_err_t err,
+                                           rd_kafka_buf_t *request) {
+        return rd_kafka_err_action(rkb, err, request, RD_KAFKA_ERR_ACTION_END);
+}
 
 /**
  * Generic handler for OffsetFetch responses.
@@ -1362,8 +1367,7 @@ rd_kafka_handle_OffsetFetch(rd_kafka_t *rk,
                            seen_cnt, (*offsets)->cnt, retry_unstable,
                            rd_kafka_err2str(err));
 
-        actions =
-            rd_kafka_err_action(rkb, err, request, RD_KAFKA_ERR_ACTION_END);
+        actions = rd_kafka_handle_OffsetFetch_err_action(rkb, err, request);
 
         if (actions & RD_KAFKA_ERR_ACTION_REFRESH) {
                 /* Re-query for coordinator */

src/rdkafka_request.h

@@ -245,6 +245,9 @@ void rd_kafka_OffsetForLeaderEpochRequest(
     rd_kafka_resp_cb_t *resp_cb,
     void *opaque);
 
+int rd_kafka_handle_OffsetFetch_err_action(rd_kafka_broker_t *rkb,
+                                           rd_kafka_resp_err_t err,
+                                           rd_kafka_buf_t *request);
 
 rd_kafka_resp_err_t
 rd_kafka_handle_OffsetFetch(rd_kafka_t *rk,

tests/0085-headers.cpp

@@ -34,7 +34,6 @@
 static RdKafka::Producer *producer;
 static RdKafka::KafkaConsumer *consumer;
 static std::string topic;
-static bool first_produce = true;
 
 static void assert_all_headers_match(RdKafka::Headers *actual,
                                      const RdKafka::Headers *expected) {
@@ -88,10 +87,6 @@ static void test_headers(RdKafka::Headers *produce_headers,
     Test::Fail("produce() failed: " + RdKafka::err2str(err));
 
   producer->flush(tmout_multip(10 * 1000));
-  if (first_produce) {
-    test_wait_topic_exists(producer->c_ptr(), topic.c_str(), 5000);
-    first_produce = false;
-  }
 
   if (producer->outq_len() > 0)
     Test::Fail(tostr() << "Expected producer to be flushed, "
@@ -363,6 +358,8 @@ int main_0085_headers(int argc, char **argv) {
 
   delete conf;
 
+  Test::create_topic_wait_exists(p, topic.c_str(), 1, -1, 5000);
+
   std::vector<RdKafka::TopicPartition *> parts;
   parts.push_back(RdKafka::TopicPartition::create(
       topic, 0, RdKafka::Topic::OFFSET_BEGINNING));
@@ -385,6 +382,7 @@ int main_0085_headers(int argc, char **argv) {
   test_failed_produce();
   test_assignment_op();
 
+  Test::delete_topic(p, topic.c_str());
   c->close();
   delete c;
   delete p;