[KIP-848] Test 0147: mock tests specific for the 848 consumer group protocol (confluentinc#4920)

* [KIP-8484] Mock tests specific for the new consumer group protocol
* Expedite HB only after coordinator actually changes
  Expedite coordinator query when there's a
  DESTROY_BROKER error.
* Flaky test 0067 fix

Author: emasab

src/rdkafka_cgrp.c

@@ -654,6 +654,9 @@ static int rd_kafka_cgrp_coord_update(rd_kafka_cgrp_t *rkcg, int32_t coord_id) {
                 /* Clear previous broker handle, if any */
                 if (rkcg->rkcg_curr_coord)
                         rd_kafka_cgrp_coord_clear_broker(rkcg);
+
+                rd_kafka_cgrp_consumer_expedite_next_heartbeat(
+                    rkcg, "coordinator changed");
         }
 
 
@@ -788,6 +791,8 @@ static void rd_kafka_cgrp_handle_FindCoordinator(rd_kafka_t *rk,
 
             RD_KAFKA_ERR_ACTION_RETRY, RD_KAFKA_RESP_ERR__TIMED_OUT_QUEUE,
 
+            RD_KAFKA_ERR_ACTION_RETRY, RD_KAFKA_RESP_ERR__DESTROY_BROKER,
+
             RD_KAFKA_ERR_ACTION_END);
 
 
@@ -807,6 +812,12 @@ static void rd_kafka_cgrp_handle_FindCoordinator(rd_kafka_t *rk,
                         rd_kafka_cgrp_set_last_err(rkcg, ErrorCode);
                 }
 
+                if (ErrorCode == RD_KAFKA_RESP_ERR__DESTROY_BROKER) {
+                        /* This error is one-time and should cause
+                         * an immediate retry. */
+                        rd_interval_reset(&rkcg->rkcg_coord_query_intvl);
+                }
+
                 /* Retries are performed by the timer-intervalled
                  * coord queries, continue querying */
                 rd_kafka_cgrp_set_state(rkcg, RD_KAFKA_CGRP_STATE_QUERY_COORD);
@@ -3320,8 +3331,7 @@ void rd_kafka_cgrp_handle_ConsumerGroupHeartbeat(rd_kafka_t *rk,
                 rkcg->rkcg_consumer_flags |=
                     RD_KAFKA_CGRP_CONSUMER_F_SEND_FULL_REQUEST;
                 rd_kafka_cgrp_coord_query(rkcg, rd_kafka_err2str(err));
-                rd_kafka_cgrp_consumer_expedite_next_heartbeat(
-                    rkcg, "coordinator query");
+                /* If coordinator changes, HB will be expedited. */
         }
 
         if (actions & RD_KAFKA_ERR_ACTION_SPECIAL) {

tests/0067-empty_topic.cpp

@@ -70,6 +70,9 @@ static void do_test_empty_topic_consumer() {
                RdKafka::err2str(err));
   delete md;
 
+  /* Await propagation to the whole cluster */
+  test_wait_topic_exists(consumer->c_ptr(), topic.c_str(), 1 * 1000);
+
   /* Start consumer */
   err = consumer->start(rkt, partition, RdKafka::Topic::OFFSET_BEGINNING);
   if (err)