coord_req()s would only trigger one connection attempt and then go idle

This caused things like send_offsets_to_transaction() to stall and time out
if the initial group coordinator connection was unsuccessful.

Author: edenhill

CHANGELOG.md

@@ -87,6 +87,12 @@ librdkafka v1.9.0 is a feature release:
  * If a metadata request triggered by `rd_kafka_metadata()` or consumer group rebalancing
    encountered a non-retriable error it would not be propagated to the caller and thus
    cause a stall or timeout, this has now been fixed. (@aiquestion, #3625)
+ * AdminAPI `DeleteGroups()` and `DeleteConsumerGroupOffsets()`:
+   if the given coordinator connection was not up by the time these calls were
+   initiated and the first connection attempt failed then no further connection
+   attempts were performed, ulimately leading to the calls timing out.
+   This is now fixed by keep retrying to connect to the group coordinator
+   until the connection is successful or the call times out.
  * Mock cluster `rd_kafka_mock_broker_set_down()` would previously
    accept and then disconnect new connections, it now refuses new connections.
 
@@ -128,7 +134,7 @@ librdkafka v1.9.0 is a feature release:
    (@kevinconaway)
 
 
-### Producer fixes
+### Transactional producer fixes
 
  * Fix message loss in idempotent/transactional producer.
    A corner case has been identified that may cause idempotent/transactional
@@ -162,6 +168,17 @@ librdkafka v1.9.0 is a feature release:
    broker (added in Apache Kafka 2.8), which could cause the producer to
    seemingly hang.
    This error code is now correctly handled by raising a fatal error.
+ * If the given group coordinator connection was not up by the time
+   `send_offsets_to_transactions()` was called, and the first connection
+   attempt failed then no further connection attempts were performed, ulimately
+   leading to `send_offsets_to_transactions()` timing out, and possibly
+   also the transaction timing out on the transaction coordinator.
+   This is now fixed by keep retrying to connect to the group coordinator
+   until the connection is successful or the call times out.
+
+
+### Producer fixes
+
  * Improved producer queue wakeup scheduling. This should significantly
    decrease the number of wakeups and thus syscalls for high message rate
    producers. (#3538, #2912)

src/rdkafka_broker.h

@@ -253,7 +253,7 @@ struct rd_kafka_broker_s { /* rd_kafka_broker_t */
         rd_ts_t rkb_ts_connect;
 
         /**< Persistent connection demand is tracked by
-         *   an counter for each type of demand.
+         *   a counter for each type of demand.
          *   The broker thread will maintain a persistent connection
          *   if any of the counters are non-zero, and revert to
          *   on-demand mode when they all reach zero.
@@ -276,7 +276,11 @@ struct rd_kafka_broker_s { /* rd_kafka_broker_t */
                  *   rdkafka main thread.
                  *
                  *   Producer: Broker is the transaction coordinator.
-                 *   Counter is maintained by rdkafka_idempotence.c. */
+                 *   Counter is maintained by rdkafka_idempotence.c.
+                 *
+                 *   All: A coord_req_t is waiting for this broker to come up.
+                 */
+
                 rd_atomic32_t coord;
         } rkb_persistconn;
 

src/rdkafka_coord.c

@@ -268,6 +268,15 @@ static rd_bool_t rd_kafka_coord_req_destroy(rd_kafka_t *rk,
                 return rd_false;
 
         rd_dassert(creq->creq_done);
+
+        /* Clear out coordinator we were waiting for. */
+        if (creq->creq_rkb) {
+                rd_kafka_broker_persistent_connection_del(
+                    creq->creq_rkb, &creq->creq_rkb->rkb_persistconn.coord);
+                rd_kafka_broker_destroy(creq->creq_rkb);
+                creq->creq_rkb = NULL;
+        }
+
         rd_kafka_replyq_destroy(&creq->creq_replyq);
         rd_free(creq->creq_coordkey);
         rd_free(creq);
@@ -447,6 +456,15 @@ static void rd_kafka_coord_req_fsm(rd_kafka_t *rk, rd_kafka_coord_req_t *creq) {
                         /* Cached coordinator is up, send request */
                         rd_kafka_replyq_t replyq;
 
+                        /* Clear out previous coordinator we waited for. */
+                        if (creq->creq_rkb) {
+                                rd_kafka_broker_persistent_connection_del(
+                                    creq->creq_rkb,
+                                    &creq->creq_rkb->rkb_persistconn.coord);
+                                rd_kafka_broker_destroy(creq->creq_rkb);
+                                creq->creq_rkb = NULL;
+                        }
+
                         rd_kafka_replyq_copy(&replyq, &creq->creq_replyq);
                         err = creq->creq_send_req_cb(rkb, creq->creq_rko,
                                                      replyq, creq->creq_resp_cb,
@@ -462,16 +480,37 @@ static void rd_kafka_coord_req_fsm(rd_kafka_t *rk, rd_kafka_coord_req_t *creq) {
                                                            rd_true /*done*/);
                         }
 
-                } else {
-                        /* No connection yet. We'll be re-triggered on
-                         * broker state broadcast. */
-                        rd_kafka_broker_schedule_connection(rkb);
+                } else if (creq->creq_rkb != rkb) {
+                        /* No connection yet.
+                         * Let broker thread know we need a connection.
+                         * We'll be re-triggered on broker state broadcast. */
+
+                        if (creq->creq_rkb) {
+                                /* Clear previous */
+                                rd_kafka_broker_persistent_connection_del(
+                                    rkb, &rkb->rkb_persistconn.coord);
+                                rd_kafka_broker_destroy(creq->creq_rkb);
+                        }
+
+                        rd_kafka_broker_keep(rkb);
+                        creq->creq_rkb = rkb;
+                        rd_kafka_broker_persistent_connection_add(
+                            rkb, &rkb->rkb_persistconn.coord);
                 }
 
                 rd_kafka_broker_destroy(rkb);
                 return;
+
+        } else if (creq->creq_rkb) {
+                /* No coordinator information, clear out the previous
+                 * coordinator we waited for. */
+                rd_kafka_broker_persistent_connection_del(
+                    creq->creq_rkb, &creq->creq_rkb->rkb_persistconn.coord);
+                rd_kafka_broker_destroy(creq->creq_rkb);
+                creq->creq_rkb = NULL;
         }
 
+
         /* Get any usable broker to look up the coordinator */
         rkb = rd_kafka_broker_any_usable(rk, RD_POLL_NOWAIT, RD_DO_LOCK,
                                          RD_KAFKA_FEATURE_BROKER_GROUP_COORD,

src/rdkafka_coord.h

@@ -107,6 +107,8 @@ typedef struct rd_kafka_coord_req_s {
                               *   FindCoordinator requests. */
         rd_bool_t creq_done; /**< True if request was sent */
 
+        rd_kafka_broker_t *creq_rkb; /**< creq is waiting for this broker to
+                                      *   come up. */
 } rd_kafka_coord_req_t;
 
 

tests/0105-transactions_mock.c

@@ -2666,8 +2666,8 @@ static void do_test_topic_disappears_for_awhile(void) {
         SUB_TEST_QUICK();
 
         rk = create_txn_producer(
-            &mcluster, txnid, 1, NULL, "batch.num.messages", "3", "linger.ms",
-            "100", "topic.metadata.refresh.interval.ms", "2000", NULL);
+            &mcluster, txnid, 1, "batch.num.messages", "3", "linger.ms", "100",
+            "topic.metadata.refresh.interval.ms", "2000", NULL);
 
         rd_kafka_mock_topic_create(mcluster, topic, partition_cnt, 1);
 
@@ -2742,6 +2742,104 @@ static void do_test_topic_disappears_for_awhile(void) {
 }
 
 
+/**
+ * @brief Test that group coordinator requests can handle an
+ *        untimely disconnect.
+ *
+ * The transaction manager makes use of librdkafka coord_req to commit
+ * transaction offsets to the group coordinator.
+ * If the connection to the given group coordinator is not up the
+ * coord_req code will request a connection once, but if this connection fails
+ * there will be no new attempts and the coord_req will idle until either
+ * destroyed or the connection is retried for other reasons.
+ * This in turn stalls the send_offsets_to_transaction() call until the
+ * transaction times out.
+ */
+static int delayed_up_cb(void *arg) {
+        rd_kafka_mock_cluster_t *mcluster = arg;
+        rd_sleep(3);
+        TEST_SAY("Bringing up group coordinator 2..\n");
+        rd_kafka_mock_broker_set_up(mcluster, 2);
+        return 0;
+}
+
+static void do_test_disconnected_group_coord(void) {
+        const char *topic       = "mytopic";
+        const char *txnid       = "myTxnId";
+        const char *grpid       = "myGrpId";
+        const int partition_cnt = 1;
+        rd_kafka_t *rk;
+        rd_kafka_mock_cluster_t *mcluster;
+        rd_kafka_topic_partition_list_t *offsets;
+        rd_kafka_consumer_group_metadata_t *cgmetadata;
+        test_timing_t timing;
+        thrd_t thrd;
+        int ret;
+
+        SUB_TEST_QUICK();
+
+        test_curr->is_fatal_cb = error_is_fatal_cb;
+        allowed_error          = RD_KAFKA_RESP_ERR__TRANSPORT;
+
+        rk = create_txn_producer(&mcluster, txnid, 3, NULL);
+
+        rd_kafka_mock_topic_create(mcluster, topic, partition_cnt, 1);
+
+        /* Broker 1: txn coordinator
+         * Broker 2: group coordinator
+         * Broker 3: partition leader */
+        rd_kafka_mock_coordinator_set(mcluster, "transaction", txnid, 1);
+        rd_kafka_mock_coordinator_set(mcluster, "group", grpid, 2);
+        rd_kafka_mock_partition_set_leader(mcluster, topic, 0, 3);
+
+        /* Bring down group coordinator so there are no undesired
+         * connections to it. */
+        rd_kafka_mock_broker_set_down(mcluster, 2);
+
+
+        TEST_CALL_ERROR__(rd_kafka_init_transactions(rk, -1));
+        TEST_CALL_ERROR__(rd_kafka_begin_transaction(rk));
+        TEST_CALL_ERR__(rd_kafka_producev(
+            rk, RD_KAFKA_V_TOPIC(topic), RD_KAFKA_V_PARTITION(0),
+            RD_KAFKA_V_VALUE("hi", 2), RD_KAFKA_V_END));
+        test_flush(rk, -1);
+
+        rd_sleep(1);
+
+        /* Run a background thread that after 3s, which should be enough
+         * to perform the first failed connection attempt, makes the
+         * group coordinator available again. */
+        thrd_create(&thrd, delayed_up_cb, mcluster);
+
+        TEST_SAY("Calling send_offsets_to_transaction()\n");
+        offsets = rd_kafka_topic_partition_list_new(1);
+        rd_kafka_topic_partition_list_add(offsets, topic, 0)->offset = 1;
+        cgmetadata = rd_kafka_consumer_group_metadata_new(grpid);
+
+        TIMING_START(&timing, "send_offsets_to_transaction(-1)");
+        TEST_CALL_ERROR__(
+            rd_kafka_send_offsets_to_transaction(rk, offsets, cgmetadata, -1));
+        TIMING_STOP(&timing);
+        TIMING_ASSERT(&timing, 0, 10 * 1000 /*10s*/);
+
+        rd_kafka_consumer_group_metadata_destroy(cgmetadata);
+        rd_kafka_topic_partition_list_destroy(offsets);
+        thrd_join(thrd, &ret);
+
+        /* Commit the transaction */
+        TIMING_START(&timing, "commit_transaction(-1)");
+        TEST_CALL_ERROR__(rd_kafka_commit_transaction(rk, -1));
+        TIMING_STOP(&timing);
+
+        rd_kafka_destroy(rk);
+
+        allowed_error          = RD_KAFKA_RESP_ERR_NO_ERROR;
+        test_curr->is_fatal_cb = NULL;
+
+        SUB_TEST_PASS();
+}
+
+
 int main_0105_transactions_mock(int argc, char **argv) {
         if (test_needs_auth()) {
                 TEST_SKIP("Mock cluster does not support SSL/SASL\n");
@@ -2815,5 +2913,7 @@ int main_0105_transactions_mock(int argc, char **argv) {
 
         do_test_topic_disappears_for_awhile();
 
+        do_test_disconnected_group_coord();
+
         return 0;
 }