path: root/fluent-bit/lib/librdkafka-2.1.0/tests/0105-transactions_mock.c

/*
 * librdkafka - Apache Kafka C library
 *
 * Copyright (c) 2019, Magnus Edenhill
 * All rights reserved.
 *
 * Redistribution and use in source and binary forms, with or without
 * modification, are permitted provided that the following conditions are met:
 *
 * 1. Redistributions of source code must retain the above copyright notice,
 *    this list of conditions and the following disclaimer.
 * 2. Redistributions in binary form must reproduce the above copyright notice,
 *    this list of conditions and the following disclaimer in the documentation
 *    and/or other materials provided with the distribution.
 *
 * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
 * AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
 * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
 * ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE
 * LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
 * CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
 * SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
 * INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
 * CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
 * ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
 * POSSIBILITY OF SUCH DAMAGE.
 */

#include "test.h"

#include "rdkafka.h"

#include "../src/rdkafka_proto.h"
#include "../src/rdstring.h"
#include "../src/rdunittest.h"

#include <stdarg.h>


/**
 * @name Producer transaction tests using the mock cluster
 *
 */


static int allowed_error;

/**
 * @brief Decide what error_cb's will cause the test to fail.
 */
static int
error_is_fatal_cb(rd_kafka_t *rk, rd_kafka_resp_err_t err, const char *reason) {
        if (err == allowed_error ||
            /* If transport errors are allowed then it is likely
             * that we'll also see ALL_BROKERS_DOWN. */
            (allowed_error == RD_KAFKA_RESP_ERR__TRANSPORT &&
             err == RD_KAFKA_RESP_ERR__ALL_BROKERS_DOWN)) {
                TEST_SAY("Ignoring allowed error: %s: %s\n",
                         rd_kafka_err2name(err), reason);
                return 0;
        }
        return 1;
}


static rd_kafka_resp_err_t (*on_response_received_cb)(rd_kafka_t *rk,
                                                      int sockfd,
                                                      const char *brokername,
                                                      int32_t brokerid,
                                                      int16_t ApiKey,
                                                      int16_t ApiVersion,
                                                      int32_t CorrId,
                                                      size_t size,
                                                      int64_t rtt,
                                                      rd_kafka_resp_err_t err,
                                                      void *ic_opaque);

/**
 * @brief Simple on_response_received interceptor that simply calls the
 *        sub-test's on_response_received_cb function, if set.
 */
static rd_kafka_resp_err_t
on_response_received_trampoline(rd_kafka_t *rk,
                                int sockfd,
                                const char *brokername,
                                int32_t brokerid,
                                int16_t ApiKey,
                                int16_t ApiVersion,
                                int32_t CorrId,
                                size_t size,
                                int64_t rtt,
                                rd_kafka_resp_err_t err,
                                void *ic_opaque) {
        TEST_ASSERT(on_response_received_cb != NULL, "");
        return on_response_received_cb(rk, sockfd, brokername, brokerid, ApiKey,
                                       ApiVersion, CorrId, size, rtt, err,
                                       ic_opaque);
}


/**
 * @brief on_new interceptor to add an on_response_received interceptor.
 */
static rd_kafka_resp_err_t on_new_producer(rd_kafka_t *rk,
                                           const rd_kafka_conf_t *conf,
                                           void *ic_opaque,
                                           char *errstr,
                                           size_t errstr_size) {
        rd_kafka_resp_err_t err = RD_KAFKA_RESP_ERR_NO_ERROR;

        if (on_response_received_cb)
                err = rd_kafka_interceptor_add_on_response_received(
                    rk, "on_response_received", on_response_received_trampoline,
                    ic_opaque);

        return err;
}


/**
 * @brief Create a transactional producer and a mock cluster.
 *
 * The var-arg list is a NULL-terminated list of
 * (const char *key, const char *value) config properties.
 *
 * Special keys:
 *   "on_response_received", "" - enable the on_response_received_cb
 *                                interceptor,
 *                                which must be assigned prior to
 *                                calling create_tnx_producer().
 */
static RD_SENTINEL rd_kafka_t *
create_txn_producer(rd_kafka_mock_cluster_t **mclusterp,
                    const char *transactional_id,
                    int broker_cnt,
                    ...) {
        rd_kafka_conf_t *conf;
        rd_kafka_t *rk;
        char numstr[8];
        va_list ap;
        const char *key;
        rd_bool_t add_interceptors = rd_false;

        rd_snprintf(numstr, sizeof(numstr), "%d", broker_cnt);

        test_conf_init(&conf, NULL, 60);

        test_conf_set(conf, "transactional.id", transactional_id);
        /* When mock brokers are set to down state they're still binding
         * the port, just not listening to it, which makes connection attempts
         * stall until socket.connection.setup.timeout.ms expires.
         * To speed up detection of brokers being down we reduce this timeout
         * to just a couple of seconds. */
        test_conf_set(conf, "socket.connection.setup.timeout.ms", "5000");
        /* Speed up reconnects */
        test_conf_set(conf, "reconnect.backoff.max.ms", "2000");
        test_conf_set(conf, "test.mock.num.brokers", numstr);
        rd_kafka_conf_set_dr_msg_cb(conf, test_dr_msg_cb);

        test_curr->ignore_dr_err = rd_false;

        va_start(ap, broker_cnt);
        while ((key = va_arg(ap, const char *))) {
                if (!strcmp(key, "on_response_received")) {
                        add_interceptors = rd_true;
                        (void)va_arg(ap, const char *);
                } else {
                        test_conf_set(conf, key, va_arg(ap, const char *));
                }
        }
        va_end(ap);

        /* Add an on_.. interceptors */
        if (add_interceptors)
                rd_kafka_conf_interceptor_add_on_new(conf, "on_new_producer",
                                                     on_new_producer, NULL);

        rk = test_create_handle(RD_KAFKA_PRODUCER, conf);

        if (mclusterp) {
                *mclusterp = rd_kafka_handle_mock_cluster(rk);
                TEST_ASSERT(*mclusterp, "failed to create mock cluster");

                /* Create some of the common consumer "input" topics
                 * that we must be able to commit to with
                 * send_offsets_to_transaction().
                 * The number depicts the number of partitions in the topic. */
                TEST_CALL_ERR__(
                    rd_kafka_mock_topic_create(*mclusterp, "srctopic4", 4, 1));
                TEST_CALL_ERR__(rd_kafka_mock_topic_create(
                    *mclusterp, "srctopic64", 64, 1));
        }

        return rk;
}


/**
 * @brief Test recoverable errors using mock broker error injections
 *        and code coverage checks.
 */
static void do_test_txn_recoverable_errors(void) {
        rd_kafka_t *rk;
        rd_kafka_mock_cluster_t *mcluster;
        rd_kafka_topic_partition_list_t *offsets;
        rd_kafka_consumer_group_metadata_t *cgmetadata;
        const char *groupid = "myGroupId";
        const char *txnid   = "myTxnId";

        SUB_TEST_QUICK();

        rk = create_txn_producer(&mcluster, txnid, 3, "batch.num.messages", "1",
                                 NULL);

        /* Make sure transaction and group coordinators are different.
         * This verifies that AddOffsetsToTxnRequest isn't sent to the
         * transaction coordinator but the group coordinator. */
        rd_kafka_mock_coordinator_set(mcluster, "group", groupid, 1);
        rd_kafka_mock_coordinator_set(mcluster, "transaction", txnid, 2);

        /*
         * Inject som InitProducerId errors that causes retries
         */
        rd_kafka_mock_push_request_errors(
            mcluster, RD_KAFKAP_InitProducerId, 3,
            RD_KAFKA_RESP_ERR_COORDINATOR_NOT_AVAILABLE,
            RD_KAFKA_RESP_ERR_NOT_COORDINATOR,
            RD_KAFKA_RESP_ERR_COORDINATOR_LOAD_IN_PROGRESS);

        TEST_CALL_ERROR__(rd_kafka_init_transactions(rk, 5000));

        (void)RD_UT_COVERAGE_CHECK(0); /* idemp_request_pid_failed(retry) */
        (void)RD_UT_COVERAGE_CHECK(1); /* txn_idemp_state_change(READY) */

        /*
         * Start a transaction
         */
        TEST_CALL_ERROR__(rd_kafka_begin_transaction(rk));


        /* Produce a message without error first */
        TEST_CALL_ERR__(rd_kafka_producev(
            rk, RD_KAFKA_V_TOPIC("mytopic"), RD_KAFKA_V_PARTITION(0),
            RD_KAFKA_V_VALUE("hi", 2), RD_KAFKA_V_END));

        rd_kafka_flush(rk, -1);

        /*
         * Produce a message, let it fail with a non-idempo/non-txn
         * retryable error
         */
        rd_kafka_mock_push_request_errors(
            mcluster, RD_KAFKAP_Produce, 1,
            RD_KAFKA_RESP_ERR_NOT_ENOUGH_REPLICAS);

        TEST_CALL_ERR__(rd_kafka_producev(
            rk, RD_KAFKA_V_TOPIC("mytopic"), RD_KAFKA_V_PARTITION(0),
            RD_KAFKA_V_VALUE("hi", 2), RD_KAFKA_V_END));

        /* Make sure messages are produced */
        rd_kafka_flush(rk, -1);

        /*
         * Send some arbitrary offsets, first with some failures, then
         * succeed.
         */
        offsets = rd_kafka_topic_partition_list_new(4);
        rd_kafka_topic_partition_list_add(offsets, "srctopic4", 3)->offset = 12;
        rd_kafka_topic_partition_list_add(offsets, "srctopic64", 39)->offset =
            999999111;
        rd_kafka_topic_partition_list_add(offsets, "srctopic4", 0)->offset =
            999;
        rd_kafka_topic_partition_list_add(offsets, "srctopic64", 19)->offset =
            123456789;

        rd_kafka_mock_push_request_errors(
            mcluster, RD_KAFKAP_AddPartitionsToTxn, 1,
            RD_KAFKA_RESP_ERR_UNKNOWN_TOPIC_OR_PART);

        rd_kafka_mock_push_request_errors(
            mcluster, RD_KAFKAP_TxnOffsetCommit, 2,
            RD_KAFKA_RESP_ERR_NOT_COORDINATOR,
            RD_KAFKA_RESP_ERR_CONCURRENT_TRANSACTIONS);

        cgmetadata = rd_kafka_consumer_group_metadata_new("mygroupid");

        TEST_CALL_ERROR__(
            rd_kafka_send_offsets_to_transaction(rk, offsets, cgmetadata, -1));

        rd_kafka_consumer_group_metadata_destroy(cgmetadata);
        rd_kafka_topic_partition_list_destroy(offsets);

        /*
         * Commit transaction, first with som failures, then succeed.
         */
        rd_kafka_mock_push_request_errors(
            mcluster, RD_KAFKAP_EndTxn, 3,
            RD_KAFKA_RESP_ERR_COORDINATOR_NOT_AVAILABLE,
            RD_KAFKA_RESP_ERR_NOT_COORDINATOR,
            RD_KAFKA_RESP_ERR_COORDINATOR_LOAD_IN_PROGRESS);

        TEST_CALL_ERROR__(rd_kafka_commit_transaction(rk, 5000));

        /* All done */

        rd_kafka_destroy(rk);

        SUB_TEST_PASS();
}


/**
 * @brief KIP-360: Test that fatal idempotence errors triggers abortable
 *        transaction errors and that the producer can recover.
 */
static void do_test_txn_fatal_idempo_errors(void) {
        rd_kafka_t *rk;
        rd_kafka_mock_cluster_t *mcluster;
        rd_kafka_error_t *error;
        const char *txnid = "myTxnId";

        SUB_TEST_QUICK();

        rk = create_txn_producer(&mcluster, txnid, 3, "batch.num.messages", "1",
                                 NULL);

        test_curr->ignore_dr_err = rd_true;
        test_curr->is_fatal_cb   = error_is_fatal_cb;
        allowed_error            = RD_KAFKA_RESP_ERR_UNKNOWN_PRODUCER_ID;

        TEST_CALL_ERROR__(rd_kafka_init_transactions(rk, 5000));

        /*
         * Start a transaction
         */
        TEST_CALL_ERROR__(rd_kafka_begin_transaction(rk));


        /* Produce a message without error first */
        TEST_CALL_ERR__(rd_kafka_producev(
            rk, RD_KAFKA_V_TOPIC("mytopic"), RD_KAFKA_V_PARTITION(0),
            RD_KAFKA_V_VALUE("hi", 2), RD_KAFKA_V_END));

        /* Produce a message, let it fail with a fatal idempo error. */
        rd_kafka_mock_push_request_errors(
            mcluster, RD_KAFKAP_Produce, 1,
            RD_KAFKA_RESP_ERR_UNKNOWN_PRODUCER_ID);

        TEST_CALL_ERR__(rd_kafka_producev(
            rk, RD_KAFKA_V_TOPIC("mytopic"), RD_KAFKA_V_PARTITION(0),
            RD_KAFKA_V_VALUE("hi", 2), RD_KAFKA_V_END));

        /* Commit the transaction, should fail */
        error = rd_kafka_commit_transaction(rk, -1);
        TEST_ASSERT(error != NULL, "Expected commit_transaction() to fail");

        TEST_SAY("commit_transaction() failed (expectedly): %s\n",
                 rd_kafka_error_string(error));

        TEST_ASSERT(!rd_kafka_error_is_fatal(error),
                    "Did not expect fatal error");
        TEST_ASSERT(rd_kafka_error_txn_requires_abort(error),
                    "Expected abortable error");
        rd_kafka_error_destroy(error);

        /* Abort the transaction */
        TEST_CALL_ERROR__(rd_kafka_abort_transaction(rk, -1));

        /* Run a new transaction without errors to verify that the
         * producer can recover. */
        TEST_CALL_ERROR__(rd_kafka_begin_transaction(rk));

        TEST_CALL_ERR__(rd_kafka_producev(
            rk, RD_KAFKA_V_TOPIC("mytopic"), RD_KAFKA_V_PARTITION(0),
            RD_KAFKA_V_VALUE("hi", 2), RD_KAFKA_V_END));

        TEST_CALL_ERROR__(rd_kafka_commit_transaction(rk, -1));

        /* All done */

        rd_kafka_destroy(rk);

        allowed_error = RD_KAFKA_RESP_ERR_NO_ERROR;

        SUB_TEST_PASS();
}


/**
 * @brief KIP-360: Test that fatal idempotence errors triggers abortable
 *        transaction errors, but let the broker-side bumping of the
 *        producer PID take longer than the remaining transaction timeout
 *        which should raise a retriable error from abort_transaction().
 *
 * @param with_sleep After the first abort sleep longer than it takes to
 *                   re-init the pid so that the internal state automatically
 *                   transitions.
 */
static void do_test_txn_slow_reinit(rd_bool_t with_sleep) {
        rd_kafka_t *rk;
        rd_kafka_mock_cluster_t *mcluster;
        rd_kafka_error_t *error;
        int32_t txn_coord = 2;
        const char *txnid = "myTxnId";
        test_timing_t timing;

        SUB_TEST("%s sleep", with_sleep ? "with" : "without");

        rk = create_txn_producer(&mcluster, txnid, 3, "batch.num.messages", "1",
                                 NULL);

        rd_kafka_mock_coordinator_set(mcluster, "transaction", txnid,
                                      txn_coord);

        test_curr->ignore_dr_err = rd_true;
        test_curr->is_fatal_cb   = NULL;

        TEST_CALL_ERROR__(rd_kafka_init_transactions(rk, -1));

        /*
         * Start a transaction
         */
        TEST_CALL_ERROR__(rd_kafka_begin_transaction(rk));


        /* Produce a message without error first */
        TEST_CALL_ERR__(rd_kafka_producev(
            rk, RD_KAFKA_V_TOPIC("mytopic"), RD_KAFKA_V_PARTITION(0),
            RD_KAFKA_V_VALUE("hi", 2), RD_KAFKA_V_END));

        test_flush(rk, -1);

        /* Set transaction coordinator latency higher than
         * the abort_transaction() call timeout so that the automatic
         * re-initpid takes longer than abort_transaction(). */
        rd_kafka_mock_broker_push_request_error_rtts(
            mcluster, txn_coord, RD_KAFKAP_InitProducerId, 1,
            RD_KAFKA_RESP_ERR_NO_ERROR, 10000 /*10s*/);

        /* Produce a message, let it fail with a fatal idempo error. */
        rd_kafka_mock_push_request_errors(
            mcluster, RD_KAFKAP_Produce, 1,
            RD_KAFKA_RESP_ERR_UNKNOWN_PRODUCER_ID);

        TEST_CALL_ERR__(rd_kafka_producev(
            rk, RD_KAFKA_V_TOPIC("mytopic"), RD_KAFKA_V_PARTITION(0),
            RD_KAFKA_V_VALUE("hi", 2), RD_KAFKA_V_END));


        /* Commit the transaction, should fail */
        TIMING_START(&timing, "commit_transaction(-1)");
        error = rd_kafka_commit_transaction(rk, -1);
        TIMING_STOP(&timing);
        TEST_ASSERT(error != NULL, "Expected commit_transaction() to fail");

        TEST_SAY("commit_transaction() failed (expectedly): %s\n",
                 rd_kafka_error_string(error));

        TEST_ASSERT(!rd_kafka_error_is_fatal(error),
                    "Did not expect fatal error");
        TEST_ASSERT(rd_kafka_error_txn_requires_abort(error),
                    "Expected abortable error");
        rd_kafka_error_destroy(error);

        /* Abort the transaction, should fail with retriable (timeout) error */
        TIMING_START(&timing, "abort_transaction(100)");
        error = rd_kafka_abort_transaction(rk, 100);
        TIMING_STOP(&timing);
        TEST_ASSERT(error != NULL, "Expected abort_transaction() to fail");

        TEST_SAY("First abort_transaction() failed: %s\n",
                 rd_kafka_error_string(error));
        TEST_ASSERT(!rd_kafka_error_is_fatal(error),
                    "Did not expect fatal error");
        TEST_ASSERT(rd_kafka_error_is_retriable(error),
                    "Expected retriable error");
        rd_kafka_error_destroy(error);

        if (with_sleep)
                rd_sleep(12);

        /* Retry abort, should now finish. */
        TEST_SAY("Retrying abort\n");
        TIMING_START(&timing, "abort_transaction(-1)");
        TEST_CALL_ERROR__(rd_kafka_abort_transaction(rk, -1));
        TIMING_STOP(&timing);

        /* Run a new transaction without errors to verify that the
         * producer can recover. */
        TEST_CALL_ERROR__(rd_kafka_begin_transaction(rk));

        TEST_CALL_ERR__(rd_kafka_producev(
            rk, RD_KAFKA_V_TOPIC("mytopic"), RD_KAFKA_V_PARTITION(0),
            RD_KAFKA_V_VALUE("hi", 2), RD_KAFKA_V_END));

        TEST_CALL_ERROR__(rd_kafka_commit_transaction(rk, -1));

        /* All done */

        rd_kafka_destroy(rk);

        allowed_error = RD_KAFKA_RESP_ERR_NO_ERROR;

        SUB_TEST_PASS();
}



/**
 * @brief KIP-360: Test that fatal idempotence errors triggers abortable
 *        transaction errors, but let the broker-side bumping of the
 *        producer PID fail with a fencing error.
 *        Should raise a fatal error.
 *
 * @param error_code Which error code InitProducerIdRequest should fail with.
 *                   Either RD_KAFKA_RESP_ERR_INVALID_PRODUCER_EPOCH (older)
 *                   or RD_KAFKA_RESP_ERR_PRODUCER_FENCED (newer).
 */
static void do_test_txn_fenced_reinit(rd_kafka_resp_err_t error_code) {
        rd_kafka_t *rk;
        rd_kafka_mock_cluster_t *mcluster;
        rd_kafka_error_t *error;
        int32_t txn_coord = 2;
        const char *txnid = "myTxnId";
        char errstr[512];
        rd_kafka_resp_err_t fatal_err;

        SUB_TEST_QUICK("With error %s", rd_kafka_err2name(error_code));

        rk = create_txn_producer(&mcluster, txnid, 3, "batch.num.messages", "1",
                                 NULL);

        rd_kafka_mock_coordinator_set(mcluster, "transaction", txnid,
                                      txn_coord);

        test_curr->ignore_dr_err = rd_true;
        test_curr->is_fatal_cb   = error_is_fatal_cb;
        allowed_error            = RD_KAFKA_RESP_ERR__FENCED;

        TEST_CALL_ERROR__(rd_kafka_init_transactions(rk, -1));

        /*
         * Start a transaction
         */
        TEST_CALL_ERROR__(rd_kafka_begin_transaction(rk));


        /* Produce a message without error first */
        TEST_CALL_ERR__(rd_kafka_producev(
            rk, RD_KAFKA_V_TOPIC("mytopic"), RD_KAFKA_V_PARTITION(0),
            RD_KAFKA_V_VALUE("hi", 2), RD_KAFKA_V_END));

        test_flush(rk, -1);

        /* Fail the PID reinit */
        rd_kafka_mock_broker_push_request_error_rtts(
            mcluster, txn_coord, RD_KAFKAP_InitProducerId, 1, error_code, 0);

        /* Produce a message, let it fail with a fatal idempo error. */
        rd_kafka_mock_push_request_errors(
            mcluster, RD_KAFKAP_Produce, 1,
            RD_KAFKA_RESP_ERR_UNKNOWN_PRODUCER_ID);

        TEST_CALL_ERR__(rd_kafka_producev(
            rk, RD_KAFKA_V_TOPIC("mytopic"), RD_KAFKA_V_PARTITION(0),
            RD_KAFKA_V_VALUE("hi", 2), RD_KAFKA_V_END));

        test_flush(rk, -1);

        /* Abort the transaction, should fail with a fatal error */
        error = rd_kafka_abort_transaction(rk, -1);
        TEST_ASSERT(error != NULL, "Expected abort_transaction() to fail");

        TEST_SAY("abort_transaction() failed: %s\n",
                 rd_kafka_error_string(error));
        TEST_ASSERT(rd_kafka_error_is_fatal(error), "Expected a fatal error");
        rd_kafka_error_destroy(error);

        fatal_err = rd_kafka_fatal_error(rk, errstr, sizeof(errstr));
        TEST_ASSERT(fatal_err, "Expected a fatal error to have been raised");
        TEST_SAY("Fatal error: %s: %s\n", rd_kafka_err2name(fatal_err), errstr);

        /* All done */

        rd_kafka_destroy(rk);

        allowed_error = RD_KAFKA_RESP_ERR_NO_ERROR;

        SUB_TEST_PASS();
}


/**
 * @brief Test EndTxn errors.
 */
static void do_test_txn_endtxn_errors(void) {
        rd_kafka_t *rk                    = NULL;
        rd_kafka_mock_cluster_t *mcluster = NULL;
        rd_kafka_resp_err_t err;
        struct {
                size_t error_cnt;
                rd_kafka_resp_err_t errors[4];
                rd_kafka_resp_err_t exp_err;
                rd_bool_t exp_retriable;
                rd_bool_t exp_abortable;
                rd_bool_t exp_fatal;
                rd_bool_t exp_successful_abort;
        } scenario[] = {
            /* This list of errors is from the EndTxnResponse handler in
             * AK clients/.../TransactionManager.java */
            {
                /* #0 */
                2,
                {RD_KAFKA_RESP_ERR_COORDINATOR_NOT_AVAILABLE,
                 RD_KAFKA_RESP_ERR_COORDINATOR_NOT_AVAILABLE},
                /* Should auto-recover */
                RD_KAFKA_RESP_ERR_NO_ERROR,
            },
            {
                /* #1 */
                2,
                {RD_KAFKA_RESP_ERR_NOT_COORDINATOR,
                 RD_KAFKA_RESP_ERR_NOT_COORDINATOR},
                /* Should auto-recover */
                RD_KAFKA_RESP_ERR_NO_ERROR,
            },
            {
                /* #2 */
                1,
                {RD_KAFKA_RESP_ERR_COORDINATOR_LOAD_IN_PROGRESS},
                /* Should auto-recover */
                RD_KAFKA_RESP_ERR_NO_ERROR,
            },
            {
                /* #3 */
                3,
                {RD_KAFKA_RESP_ERR_CONCURRENT_TRANSACTIONS,
                 RD_KAFKA_RESP_ERR_CONCURRENT_TRANSACTIONS,
                 RD_KAFKA_RESP_ERR_CONCURRENT_TRANSACTIONS},
                /* Should auto-recover */
                RD_KAFKA_RESP_ERR_NO_ERROR,
            },
            {
                /* #4: the abort is auto-recovering thru epoch bump */
                1,
                {RD_KAFKA_RESP_ERR_UNKNOWN_PRODUCER_ID},
                RD_KAFKA_RESP_ERR_UNKNOWN_PRODUCER_ID,
                rd_false /* !retriable */,
                rd_true /* abortable */,
                rd_false /* !fatal */,
                rd_true /* successful abort */
            },
            {
                /* #5: the abort is auto-recovering thru epoch bump */
                1,
                {RD_KAFKA_RESP_ERR_INVALID_PRODUCER_ID_MAPPING},
                RD_KAFKA_RESP_ERR_INVALID_PRODUCER_ID_MAPPING,
                rd_false /* !retriable */,
                rd_true /* abortable */,
                rd_false /* !fatal */,
                rd_true /* successful abort */
            },
            {
                /* #6 */
                1,
                {RD_KAFKA_RESP_ERR_INVALID_PRODUCER_EPOCH},
                /* This error is normalized */
                RD_KAFKA_RESP_ERR__FENCED,
                rd_false /* !retriable */,
                rd_false /* !abortable */,
                rd_true /* fatal */
            },
            {
                /* #7 */
                1,
                {RD_KAFKA_RESP_ERR_PRODUCER_FENCED},
                /* This error is normalized */
                RD_KAFKA_RESP_ERR__FENCED,
                rd_false /* !retriable */,
                rd_false /* !abortable */,
                rd_true /* fatal */
            },
            {
                /* #8 */
                1,
                {RD_KAFKA_RESP_ERR_TRANSACTIONAL_ID_AUTHORIZATION_FAILED},
                RD_KAFKA_RESP_ERR_TRANSACTIONAL_ID_AUTHORIZATION_FAILED,
                rd_false /* !retriable */,
                rd_false /* !abortable */,
                rd_true /* fatal */
            },
            {
                /* #9 */
                1,
                {RD_KAFKA_RESP_ERR_GROUP_AUTHORIZATION_FAILED},
                RD_KAFKA_RESP_ERR_GROUP_AUTHORIZATION_FAILED,
                rd_false /* !retriable */,
                rd_true /* abortable */,
                rd_false /* !fatal */
            },
            {
                /* #10 */
                /* Any other error should raise a fatal error */
                1,
                {RD_KAFKA_RESP_ERR_INVALID_MSG_SIZE},
                RD_KAFKA_RESP_ERR_INVALID_MSG_SIZE,
                rd_false /* !retriable */,
                rd_true /* abortable */,
                rd_false /* !fatal */,
            },
            {
                /* #11 */
                1,
                {RD_KAFKA_RESP_ERR_PRODUCER_FENCED},
                /* This error is normalized */
                RD_KAFKA_RESP_ERR__FENCED,
                rd_false /* !retriable */,
                rd_false /* !abortable */,
                rd_true /* fatal */
            },
            {0},
        };
        int i;

        SUB_TEST_QUICK();

        for (i = 0; scenario[i].error_cnt > 0; i++) {
                int j;
                /* For each scenario, test:
                 *   commit_transaction()
                 *   flush() + commit_transaction()
                 *   abort_transaction()
                 *   flush() + abort_transaction()
                 */
                for (j = 0; j < (2 + 2); j++) {
                        rd_bool_t commit     = j < 2;
                        rd_bool_t with_flush = j & 1;
                        rd_bool_t exp_successful_abort =
                            !commit && scenario[i].exp_successful_abort;
                        const char *commit_str =
                            commit ? (with_flush ? "commit&flush" : "commit")
                                   : (with_flush ? "abort&flush" : "abort");
                        rd_kafka_topic_partition_list_t *offsets;
                        rd_kafka_consumer_group_metadata_t *cgmetadata;
                        rd_kafka_error_t *error;
                        test_timing_t t_call;

                        TEST_SAY("Testing scenario #%d %s with %" PRIusz
                                 " injected erorrs, expecting %s\n",
                                 i, commit_str, scenario[i].error_cnt,
                                 exp_successful_abort
                                     ? "successful abort"
                                     : rd_kafka_err2name(scenario[i].exp_err));

                        if (!rk) {
                                const char *txnid = "myTxnId";
                                rk = create_txn_producer(&mcluster, txnid, 3,
                                                         NULL);
                                TEST_CALL_ERROR__(
                                    rd_kafka_init_transactions(rk, 5000));
                        }

                        /*
                         * Start transaction
                         */
                        TEST_CALL_ERROR__(rd_kafka_begin_transaction(rk));

                        /* Transaction aborts will cause DR errors:
                         * ignore them. */
                        test_curr->ignore_dr_err = !commit;

                        /*
                         * Produce a message.
                         */
                        err = rd_kafka_producev(rk, RD_KAFKA_V_TOPIC("mytopic"),
                                                RD_KAFKA_V_VALUE("hi", 2),
                                                RD_KAFKA_V_END);
                        TEST_ASSERT(!err, "produce failed: %s",
                                    rd_kafka_err2str(err));

                        if (with_flush)
                                test_flush(rk, -1);

                        /*
                         * Send some arbitrary offsets.
                         */
                        offsets = rd_kafka_topic_partition_list_new(4);
                        rd_kafka_topic_partition_list_add(offsets, "srctopic4",
                                                          3)
                            ->offset = 12;
                        rd_kafka_topic_partition_list_add(offsets, "srctopic64",
                                                          60)
                            ->offset = 99999;

                        cgmetadata =
                            rd_kafka_consumer_group_metadata_new("mygroupid");

                        TEST_CALL_ERROR__(rd_kafka_send_offsets_to_transaction(
                            rk, offsets, cgmetadata, -1));

                        rd_kafka_consumer_group_metadata_destroy(cgmetadata);
                        rd_kafka_topic_partition_list_destroy(offsets);

                        /*
                         * Commit transaction, first with som failures,
                         * then succeed.
                         */
                        rd_kafka_mock_push_request_errors_array(
                            mcluster, RD_KAFKAP_EndTxn, scenario[i].error_cnt,
                            scenario[i].errors);

                        TIMING_START(&t_call, "%s", commit_str);
                        if (commit)
                                error = rd_kafka_commit_transaction(
                                    rk, tmout_multip(5000));
                        else
                                error = rd_kafka_abort_transaction(
                                    rk, tmout_multip(5000));
                        TIMING_STOP(&t_call);

                        if (error)
                                TEST_SAY(
                                    "Scenario #%d %s failed: %s: %s "
                                    "(retriable=%s, req_abort=%s, "
                                    "fatal=%s)\n",
                                    i, commit_str, rd_kafka_error_name(error),
                                    rd_kafka_error_string(error),
                                    RD_STR_ToF(
                                        rd_kafka_error_is_retriable(error)),
                                    RD_STR_ToF(
                                        rd_kafka_error_txn_requires_abort(
                                            error)),
                                    RD_STR_ToF(rd_kafka_error_is_fatal(error)));
                        else
                                TEST_SAY("Scenario #%d %s succeeded\n", i,
                                         commit_str);

                        if (!scenario[i].exp_err || exp_successful_abort) {
                                TEST_ASSERT(!error,
                                            "Expected #%d %s to succeed, "
                                            "got %s",
                                            i, commit_str,
                                            rd_kafka_error_string(error));
                                continue;
                        }


                        TEST_ASSERT(error != NULL, "Expected #%d %s to fail", i,
                                    commit_str);
                        TEST_ASSERT(scenario[i].exp_err ==
                                        rd_kafka_error_code(error),
                                    "Scenario #%d: expected %s, not %s", i,
                                    rd_kafka_err2name(scenario[i].exp_err),
                                    rd_kafka_error_name(error));
                        TEST_ASSERT(
                            scenario[i].exp_retriable ==
                                (rd_bool_t)rd_kafka_error_is_retriable(error),
                            "Scenario #%d: retriable mismatch", i);
                        TEST_ASSERT(
                            scenario[i].exp_abortable ==
                                (rd_bool_t)rd_kafka_error_txn_requires_abort(
                                    error),
                            "Scenario #%d: abortable mismatch", i);
                        TEST_ASSERT(
                            scenario[i].exp_fatal ==
                                (rd_bool_t)rd_kafka_error_is_fatal(error),
                            "Scenario #%d: fatal mismatch", i);

                        /* Handle errors according to the error flags */
                        if (rd_kafka_error_is_fatal(error)) {
                                TEST_SAY("Fatal error, destroying producer\n");
                                rd_kafka_error_destroy(error);
                                rd_kafka_destroy(rk);
                                rk = NULL; /* Will be re-created on the next
                                            * loop iteration. */

                        } else if (rd_kafka_error_txn_requires_abort(error)) {
                                rd_kafka_error_destroy(error);
                                TEST_SAY(
                                    "Abortable error, "
                                    "aborting transaction\n");
                                TEST_CALL_ERROR__(
                                    rd_kafka_abort_transaction(rk, -1));

                        } else if (rd_kafka_error_is_retriable(error)) {
                                rd_kafka_error_destroy(error);
                                TEST_SAY("Retriable error, retrying %s once\n",
                                         commit_str);
                                if (commit)
                                        TEST_CALL_ERROR__(
                                            rd_kafka_commit_transaction(rk,
                                                                        5000));
                                else
                                        TEST_CALL_ERROR__(
                                            rd_kafka_abort_transaction(rk,
                                                                       5000));
                        } else {
                                TEST_FAIL(
                                    "Scenario #%d %s: "
                                    "Permanent error without enough "
                                    "hints to proceed: %s\n",
                                    i, commit_str,
                                    rd_kafka_error_string(error));
                        }
                }
        }

        /* All done */
        if (rk)
                rd_kafka_destroy(rk);

        SUB_TEST_PASS();
}


/**
 * @brief Test that the commit/abort works properly with infinite timeout.
 */
static void do_test_txn_endtxn_infinite(void) {
        rd_kafka_t *rk;
        rd_kafka_mock_cluster_t *mcluster = NULL;
        const char *txnid                 = "myTxnId";
        int i;

        SUB_TEST_QUICK();

        rk = create_txn_producer(&mcluster, txnid, 3, NULL);

        TEST_CALL_ERROR__(rd_kafka_init_transactions(rk, 5000));

        for (i = 0; i < 2; i++) {
                rd_bool_t commit       = i == 0;
                const char *commit_str = commit ? "commit" : "abort";
                rd_kafka_error_t *error;
                test_timing_t t_call;

                /* Messages will fail on as the transaction fails,
                 * ignore the DR error */
                test_curr->ignore_dr_err = rd_true;

                TEST_CALL_ERROR__(rd_kafka_begin_transaction(rk));

                TEST_CALL_ERR__(rd_kafka_producev(
                    rk, RD_KAFKA_V_TOPIC("mytopic"), RD_KAFKA_V_VALUE("hi", 2),
                    RD_KAFKA_V_END));

                /*
                 * Commit/abort transaction, first with som retriable failures,
                 * then success.
                 */
                rd_kafka_mock_push_request_errors(
                    mcluster, RD_KAFKAP_EndTxn, 10,
                    RD_KAFKA_RESP_ERR_COORDINATOR_NOT_AVAILABLE,
                    RD_KAFKA_RESP_ERR_NOT_COORDINATOR,
                    RD_KAFKA_RESP_ERR_NOT_COORDINATOR,
                    RD_KAFKA_RESP_ERR_NOT_COORDINATOR,
                    RD_KAFKA_RESP_ERR_NOT_COORDINATOR,
                    RD_KAFKA_RESP_ERR_NOT_COORDINATOR,
                    RD_KAFKA_RESP_ERR_NOT_COORDINATOR,
                    RD_KAFKA_RESP_ERR_COORDINATOR_LOAD_IN_PROGRESS,
                    RD_KAFKA_RESP_ERR_NOT_COORDINATOR,
                    RD_KAFKA_RESP_ERR_NOT_COORDINATOR);

                rd_sleep(1);

                TIMING_START(&t_call, "%s_transaction()", commit_str);
                if (commit)
                        error = rd_kafka_commit_transaction(rk, -1);
                else
                        error = rd_kafka_abort_transaction(rk, -1);
                TIMING_STOP(&t_call);

                TEST_SAY("%s returned %s\n", commit_str,
                         error ? rd_kafka_error_string(error) : "success");

                TEST_ASSERT(!error, "Expected %s to succeed, got %s",
                            commit_str, rd_kafka_error_string(error));
        }

        /* All done */

        rd_kafka_destroy(rk);

        SUB_TEST_PASS();
}



/**
 * @brief Test that the commit/abort user timeout is honoured.
 */
static void do_test_txn_endtxn_timeout(void) {
        rd_kafka_t *rk;
        rd_kafka_mock_cluster_t *mcluster = NULL;
        const char *txnid                 = "myTxnId";
        int i;

        SUB_TEST_QUICK();

        rk = create_txn_producer(&mcluster, txnid, 3, NULL);

        TEST_CALL_ERROR__(rd_kafka_init_transactions(rk, 5000));

        for (i = 0; i < 2; i++) {
                rd_bool_t commit       = i == 0;
                const char *commit_str = commit ? "commit" : "abort";
                rd_kafka_error_t *error;
                test_timing_t t_call;

                /* Messages will fail as the transaction fails,
                 * ignore the DR error */
                test_curr->ignore_dr_err = rd_true;

                TEST_CALL_ERROR__(rd_kafka_begin_transaction(rk));

                TEST_CALL_ERR__(rd_kafka_producev(
                    rk, RD_KAFKA_V_TOPIC("mytopic"), RD_KAFKA_V_VALUE("hi", 2),
                    RD_KAFKA_V_END));

                /*
                 * Commit/abort transaction, first with some retriable failures
                 * whos retries exceed the user timeout.
                 */
                rd_kafka_mock_push_request_errors(
                    mcluster, RD_KAFKAP_EndTxn, 10,
                    RD_KAFKA_RESP_ERR_COORDINATOR_NOT_AVAILABLE,
                    RD_KAFKA_RESP_ERR_NOT_COORDINATOR,
                    RD_KAFKA_RESP_ERR_NOT_COORDINATOR,
                    RD_KAFKA_RESP_ERR_NOT_COORDINATOR,
                    RD_KAFKA_RESP_ERR_NOT_COORDINATOR,
                    RD_KAFKA_RESP_ERR_NOT_COORDINATOR,
                    RD_KAFKA_RESP_ERR_NOT_COORDINATOR,
                    RD_KAFKA_RESP_ERR_COORDINATOR_LOAD_IN_PROGRESS,
                    RD_KAFKA_RESP_ERR_NOT_COORDINATOR,
                    RD_KAFKA_RESP_ERR_NOT_COORDINATOR);

                rd_sleep(1);

                TIMING_START(&t_call, "%s_transaction()", commit_str);
                if (commit)
                        error = rd_kafka_commit_transaction(rk, 100);
                else
                        error = rd_kafka_abort_transaction(rk, 100);
                TIMING_STOP(&t_call);

                TEST_SAY_ERROR(error, "%s returned: ", commit_str);
                TEST_ASSERT(error != NULL, "Expected %s to fail", commit_str);
                TEST_ASSERT(
                    rd_kafka_error_code(error) == RD_KAFKA_RESP_ERR__TIMED_OUT,
                    "Expected %s to fail with timeout, not %s: %s", commit_str,
                    rd_kafka_error_name(error), rd_kafka_error_string(error));
                TEST_ASSERT(rd_kafka_error_is_retriable(error),
                            "%s failure should raise a retriable error",
                            commit_str);
                rd_kafka_error_destroy(error);

                /* Now call it again with an infinite timeout, should work. */
                TIMING_START(&t_call, "%s_transaction() nr 2", commit_str);
                if (commit)
                        TEST_CALL_ERROR__(rd_kafka_commit_transaction(rk, -1));
                else
                        TEST_CALL_ERROR__(rd_kafka_abort_transaction(rk, -1));
                TIMING_STOP(&t_call);
        }

        /* All done */

        rd_kafka_destroy(rk);

        SUB_TEST_PASS();
}



/**
 * @brief Test commit/abort inflight timeout behaviour, which should result
 *        in a retriable error.
 */
static void do_test_txn_endtxn_timeout_inflight(void) {
        rd_kafka_t *rk;
        rd_kafka_mock_cluster_t *mcluster = NULL;
        const char *txnid                 = "myTxnId";
        int32_t coord_id                  = 1;
        int i;

        SUB_TEST();

        allowed_error          = RD_KAFKA_RESP_ERR__TIMED_OUT;
        test_curr->is_fatal_cb = error_is_fatal_cb;

        rk = create_txn_producer(&mcluster, txnid, 1, "transaction.timeout.ms",
                                 "5000", NULL);

        TEST_CALL_ERROR__(rd_kafka_init_transactions(rk, -1));

        for (i = 0; i < 2; i++) {
                rd_bool_t commit       = i == 0;
                const char *commit_str = commit ? "commit" : "abort";
                rd_kafka_error_t *error;
                test_timing_t t_call;

                /* Messages will fail as the transaction fails,
                 * ignore the DR error */
                test_curr->ignore_dr_err = rd_true;

                TEST_CALL_ERROR__(rd_kafka_begin_transaction(rk));

                TEST_CALL_ERR__(rd_kafka_producev(
                    rk, RD_KAFKA_V_TOPIC("mytopic"), RD_KAFKA_V_VALUE("hi", 2),
                    RD_KAFKA_V_END));

                /* Let EndTxn & EndTxn retry timeout */
                rd_kafka_mock_broker_push_request_error_rtts(
                    mcluster, coord_id, RD_KAFKAP_EndTxn, 2,
                    RD_KAFKA_RESP_ERR_NO_ERROR, 10000,
                    RD_KAFKA_RESP_ERR_NO_ERROR, 10000);

                rd_sleep(1);

                TIMING_START(&t_call, "%s_transaction()", commit_str);
                if (commit)
                        error = rd_kafka_commit_transaction(rk, 4000);
                else
                        error = rd_kafka_abort_transaction(rk, 4000);
                TIMING_STOP(&t_call);

                TEST_SAY_ERROR(error, "%s returned: ", commit_str);
                TEST_ASSERT(error != NULL, "Expected %s to fail", commit_str);
                TEST_ASSERT(
                    rd_kafka_error_code(error) == RD_KAFKA_RESP_ERR__TIMED_OUT,
                    "Expected %s to fail with timeout, not %s: %s", commit_str,
                    rd_kafka_error_name(error), rd_kafka_error_string(error));
                TEST_ASSERT(rd_kafka_error_is_retriable(error),
                            "%s failure should raise a retriable error",
                            commit_str);
                rd_kafka_error_destroy(error);

                /* Now call it again with an infinite timeout, should work. */
                TIMING_START(&t_call, "%s_transaction() nr 2", commit_str);
                if (commit)
                        TEST_CALL_ERROR__(rd_kafka_commit_transaction(rk, -1));
                else
                        TEST_CALL_ERROR__(rd_kafka_abort_transaction(rk, -1));
                TIMING_STOP(&t_call);
        }

        /* All done */

        rd_kafka_destroy(rk);

        allowed_error          = RD_KAFKA_RESP_ERR_NO_ERROR;
        test_curr->is_fatal_cb = NULL;

        SUB_TEST_PASS();
}



/**
 * @brief Test that EndTxn is properly sent for aborted transactions
 *        even if AddOffsetsToTxnRequest was retried.
 *        This is a check for a txn_req_cnt bug.
 */
static void do_test_txn_req_cnt(void) {
        rd_kafka_t *rk;
        rd_kafka_mock_cluster_t *mcluster;
        rd_kafka_topic_partition_list_t *offsets;
        rd_kafka_consumer_group_metadata_t *cgmetadata;
        const char *txnid = "myTxnId";

        SUB_TEST_QUICK();

        rk = create_txn_producer(&mcluster, txnid, 3, NULL);

        /* Messages will fail on abort(), ignore the DR error */
        test_curr->ignore_dr_err = rd_true;

        TEST_CALL_ERROR__(rd_kafka_init_transactions(rk, 5000));

        TEST_CALL_ERROR__(rd_kafka_begin_transaction(rk));

        /*
         * Send some arbitrary offsets, first with some failures, then
         * succeed.
         */
        offsets = rd_kafka_topic_partition_list_new(2);
        rd_kafka_topic_partition_list_add(offsets, "srctopic4", 3)->offset = 12;
        rd_kafka_topic_partition_list_add(offsets, "srctopic64", 40)->offset =
            999999111;

        rd_kafka_mock_push_request_errors(mcluster, RD_KAFKAP_AddOffsetsToTxn,
                                          2,
                                          RD_KAFKA_RESP_ERR_REQUEST_TIMED_OUT,
                                          RD_KAFKA_RESP_ERR_NOT_COORDINATOR);

        rd_kafka_mock_push_request_errors(
            mcluster, RD_KAFKAP_TxnOffsetCommit, 2,
            RD_KAFKA_RESP_ERR_COORDINATOR_LOAD_IN_PROGRESS,
            RD_KAFKA_RESP_ERR_UNKNOWN_TOPIC_OR_PART);

        cgmetadata = rd_kafka_consumer_group_metadata_new("mygroupid");

        TEST_CALL_ERROR__(
            rd_kafka_send_offsets_to_transaction(rk, offsets, cgmetadata, -1));

        rd_kafka_consumer_group_metadata_destroy(cgmetadata);
        rd_kafka_topic_partition_list_destroy(offsets);

        TEST_CALL_ERROR__(rd_kafka_abort_transaction(rk, 5000));

        /* All done */

        rd_kafka_destroy(rk);

        SUB_TEST_PASS();
}


/**
 * @brief Test abortable errors using mock broker error injections
 *        and code coverage checks.
 */
static void do_test_txn_requires_abort_errors(void) {
        rd_kafka_t *rk;
        rd_kafka_mock_cluster_t *mcluster;
        rd_kafka_error_t *error;
        rd_kafka_resp_err_t err;
        rd_kafka_topic_partition_list_t *offsets;
        rd_kafka_consumer_group_metadata_t *cgmetadata;
        int r;

        SUB_TEST_QUICK();

        rk = create_txn_producer(&mcluster, "txnid", 3, NULL);

        test_curr->ignore_dr_err = rd_true;

        TEST_CALL_ERROR__(rd_kafka_init_transactions(rk, 5000));

        TEST_CALL_ERROR__(rd_kafka_begin_transaction(rk));

        /*
         * 1. Fail on produce
         */
        TEST_SAY("1. Fail on produce\n");

        rd_kafka_mock_push_request_errors(
            mcluster, RD_KAFKAP_Produce, 1,
            RD_KAFKA_RESP_ERR_TOPIC_AUTHORIZATION_FAILED);

        err = rd_kafka_producev(rk, RD_KAFKA_V_TOPIC("mytopic"),
                                RD_KAFKA_V_VALUE("hi", 2), RD_KAFKA_V_END);
        TEST_ASSERT(!err, "produce failed: %s", rd_kafka_err2str(err));

        /* Wait for messages to fail */
        test_flush(rk, 5000);

        /* Any other transactional API should now raise an error */
        offsets = rd_kafka_topic_partition_list_new(1);
        rd_kafka_topic_partition_list_add(offsets, "srctopic4", 3)->offset = 12;

        cgmetadata = rd_kafka_consumer_group_metadata_new("mygroupid");

        error =
            rd_kafka_send_offsets_to_transaction(rk, offsets, cgmetadata, -1);

        rd_kafka_consumer_group_metadata_destroy(cgmetadata);
        rd_kafka_topic_partition_list_destroy(offsets);
        TEST_ASSERT(error, "expected error");
        TEST_ASSERT(rd_kafka_error_txn_requires_abort(error),
                    "expected abortable error, not %s",
                    rd_kafka_error_string(error));
        TEST_SAY("Error %s: %s\n", rd_kafka_error_name(error),
                 rd_kafka_error_string(error));
        rd_kafka_error_destroy(error);

        TEST_CALL_ERROR__(rd_kafka_abort_transaction(rk, -1));

        /*
         * 2. Restart transaction and fail on AddPartitionsToTxn
         */
        TEST_SAY("2. Fail on AddPartitionsToTxn\n");

        /* First refresh proper Metadata to clear the topic's auth error,
         * otherwise the produce() below will fail immediately. */
        r = test_get_partition_count(rk, "mytopic", 5000);
        TEST_ASSERT(r > 0, "Expected topic %s to exist", "mytopic");

        TEST_CALL_ERROR__(rd_kafka_begin_transaction(rk));

        rd_kafka_mock_push_request_errors(
            mcluster, RD_KAFKAP_AddPartitionsToTxn, 1,
            RD_KAFKA_RESP_ERR_TOPIC_AUTHORIZATION_FAILED);

        err = rd_kafka_producev(rk, RD_KAFKA_V_TOPIC("mytopic"),
                                RD_KAFKA_V_VALUE("hi", 2), RD_KAFKA_V_END);
        TEST_ASSERT(!err, "produce failed: %s", rd_kafka_err2str(err));

        error = rd_kafka_commit_transaction(rk, 5000);
        TEST_ASSERT(error, "commit_transaction should have failed");
        TEST_SAY("commit_transaction() error %s: %s\n",
                 rd_kafka_error_name(error), rd_kafka_error_string(error));
        rd_kafka_error_destroy(error);

        TEST_CALL_ERROR__(rd_kafka_abort_transaction(rk, -1));

        /*
         * 3. Restart transaction and fail on AddOffsetsToTxn
         */
        TEST_SAY("3. Fail on AddOffsetsToTxn\n");

        TEST_CALL_ERROR__(rd_kafka_begin_transaction(rk));

        err = rd_kafka_producev(rk, RD_KAFKA_V_TOPIC("mytopic"),
                                RD_KAFKA_V_VALUE("hi", 2), RD_KAFKA_V_END);
        TEST_ASSERT(!err, "produce failed: %s", rd_kafka_err2str(err));

        rd_kafka_mock_push_request_errors(
            mcluster, RD_KAFKAP_AddOffsetsToTxn, 1,
            RD_KAFKA_RESP_ERR_GROUP_AUTHORIZATION_FAILED);

        offsets = rd_kafka_topic_partition_list_new(1);
        rd_kafka_topic_partition_list_add(offsets, "srctopic4", 3)->offset = 12;
        cgmetadata = rd_kafka_consumer_group_metadata_new("mygroupid");

        error =
            rd_kafka_send_offsets_to_transaction(rk, offsets, cgmetadata, -1);
        TEST_ASSERT(error, "Expected send_offsets..() to fail");
        TEST_ASSERT(rd_kafka_error_code(error) ==
                        RD_KAFKA_RESP_ERR_GROUP_AUTHORIZATION_FAILED,
                    "expected send_offsets_to_transaction() to fail with "
                    "group auth error: not %s",
                    rd_kafka_error_name(error));
        rd_kafka_error_destroy(error);

        rd_kafka_consumer_group_metadata_destroy(cgmetadata);
        rd_kafka_topic_partition_list_destroy(offsets);


        error = rd_kafka_commit_transaction(rk, 5000);
        TEST_ASSERT(error, "commit_transaction should have failed");
        rd_kafka_error_destroy(error);

        TEST_CALL_ERROR__(rd_kafka_abort_transaction(rk, -1));

        /* All done */

        rd_kafka_destroy(rk);

        SUB_TEST_PASS();
}


/**
 * @brief Test error handling and recover for when broker goes down during
 *        an ongoing transaction.
 */
static void do_test_txn_broker_down_in_txn(rd_bool_t down_coord) {
        rd_kafka_t *rk;
        rd_kafka_mock_cluster_t *mcluster;
        int32_t coord_id, leader_id, down_id;
        const char *down_what;
        rd_kafka_resp_err_t err;
        const char *topic            = "test";
        const char *transactional_id = "txnid";
        int msgcnt                   = 1000;
        int remains                  = 0;

        /* Assign coordinator and leader to two different brokers */
        coord_id  = 1;
        leader_id = 2;
        if (down_coord) {
                down_id   = coord_id;
                down_what = "coordinator";
        } else {
                down_id   = leader_id;
                down_what = "leader";
        }

        SUB_TEST_QUICK("Test %s down", down_what);

        rk = create_txn_producer(&mcluster, transactional_id, 3, NULL);

        /* Broker down is not a test-failing error */
        allowed_error          = RD_KAFKA_RESP_ERR__TRANSPORT;
        test_curr->is_fatal_cb = error_is_fatal_cb;

        err = rd_kafka_mock_topic_create(mcluster, topic, 1, 3);
        TEST_ASSERT(!err, "Failed to create topic: %s", rd_kafka_err2str(err));

        rd_kafka_mock_coordinator_set(mcluster, "transaction", transactional_id,
                                      coord_id);
        rd_kafka_mock_partition_set_leader(mcluster, topic, 0, leader_id);

        /* Start transactioning */
        TEST_SAY("Starting transaction\n");
        TEST_CALL_ERROR__(rd_kafka_init_transactions(rk, 5000));

        TEST_CALL_ERROR__(rd_kafka_begin_transaction(rk));

        test_produce_msgs2_nowait(rk, topic, 0, RD_KAFKA_PARTITION_UA, 0,
                                  msgcnt / 2, NULL, 0, &remains);

        TEST_SAY("Bringing down %s %" PRId32 "\n", down_what, down_id);
        rd_kafka_mock_broker_set_down(mcluster, down_id);

        rd_kafka_flush(rk, 3000);

        /* Produce remaining messages */
        test_produce_msgs2_nowait(rk, topic, 0, RD_KAFKA_PARTITION_UA,
                                  msgcnt / 2, msgcnt / 2, NULL, 0, &remains);

        rd_sleep(2);

        TEST_SAY("Bringing up %s %" PRId32 "\n", down_what, down_id);
        rd_kafka_mock_broker_set_up(mcluster, down_id);

        TEST_CALL_ERROR__(rd_kafka_commit_transaction(rk, -1));

        TEST_ASSERT(remains == 0, "%d message(s) were not produced\n", remains);

        rd_kafka_destroy(rk);

        test_curr->is_fatal_cb = NULL;

        SUB_TEST_PASS();
}



/**
 * @brief Advance the coord_id to the next broker.
 */
static void set_next_coord(rd_kafka_mock_cluster_t *mcluster,
                           const char *transactional_id,
                           int broker_cnt,
                           int32_t *coord_idp) {
        int32_t new_coord_id;

        new_coord_id = 1 + ((*coord_idp) % (broker_cnt));
        TEST_SAY("Changing transaction coordinator from %" PRId32 " to %" PRId32
                 "\n",
                 *coord_idp, new_coord_id);
        rd_kafka_mock_coordinator_set(mcluster, "transaction", transactional_id,
                                      new_coord_id);

        *coord_idp = new_coord_id;
}

/**
 * @brief Switch coordinator during a transaction.
 *
 */
static void do_test_txn_switch_coordinator(void) {
        rd_kafka_t *rk;
        rd_kafka_mock_cluster_t *mcluster;
        int32_t coord_id;
        const char *topic            = "test";
        const char *transactional_id = "txnid";
        const int broker_cnt         = 5;
        const int iterations         = 20;
        int i;

        test_timeout_set(iterations * 10);

        SUB_TEST("Test switching coordinators");

        rk = create_txn_producer(&mcluster, transactional_id, broker_cnt, NULL);

        coord_id = 1;
        rd_kafka_mock_coordinator_set(mcluster, "transaction", transactional_id,
                                      coord_id);

        /* Start transactioning */
        TEST_SAY("Starting transaction\n");
        TEST_CALL_ERROR__(rd_kafka_init_transactions(rk, 5000));

        for (i = 0; i < iterations; i++) {
                const int msgcnt = 100;
                int remains      = 0;

                set_next_coord(mcluster, transactional_id, broker_cnt,
                               &coord_id);

                TEST_CALL_ERROR__(rd_kafka_begin_transaction(rk));

                test_produce_msgs2(rk, topic, 0, RD_KAFKA_PARTITION_UA, 0,
                                   msgcnt / 2, NULL, 0);

                if (!(i % 3))
                        set_next_coord(mcluster, transactional_id, broker_cnt,
                                       &coord_id);

                /* Produce remaining messages */
                test_produce_msgs2_nowait(rk, topic, 0, RD_KAFKA_PARTITION_UA,
                                          msgcnt / 2, msgcnt / 2, NULL, 0,
                                          &remains);

                if ((i & 1) || !(i % 8))
                        set_next_coord(mcluster, transactional_id, broker_cnt,
                                       &coord_id);


                if (!(i % 5)) {
                        test_curr->ignore_dr_err = rd_false;
                        TEST_CALL_ERROR__(rd_kafka_commit_transaction(rk, -1));

                } else {
                        test_curr->ignore_dr_err = rd_true;
                        TEST_CALL_ERROR__(rd_kafka_abort_transaction(rk, -1));
                }
        }


        rd_kafka_destroy(rk);

        SUB_TEST_PASS();
}


/**
 * @brief Switch coordinator during a transaction when AddOffsetsToTxn
 *        are sent. #3571.
 */
static void do_test_txn_switch_coordinator_refresh(void) {
        rd_kafka_t *rk;
        rd_kafka_mock_cluster_t *mcluster;
        const char *topic            = "test";
        const char *transactional_id = "txnid";
        rd_kafka_topic_partition_list_t *offsets;
        rd_kafka_consumer_group_metadata_t *cgmetadata;

        SUB_TEST("Test switching coordinators (refresh)");

        rk = create_txn_producer(&mcluster, transactional_id, 3, NULL);

        rd_kafka_mock_coordinator_set(mcluster, "transaction", transactional_id,
                                      1);

        /* Start transactioning */
        TEST_SAY("Starting transaction\n");
        TEST_CALL_ERROR__(rd_kafka_init_transactions(rk, 5000));

        TEST_CALL_ERROR__(rd_kafka_begin_transaction(rk));

        /* Switch the coordinator so that AddOffsetsToTxnRequest
         * will respond with NOT_COORDINATOR. */
        TEST_SAY("Switching to coordinator 2\n");
        rd_kafka_mock_coordinator_set(mcluster, "transaction", transactional_id,
                                      2);

        /*
         * Send some arbitrary offsets.
         */
        offsets = rd_kafka_topic_partition_list_new(4);
        rd_kafka_topic_partition_list_add(offsets, "srctopic4", 3)->offset = 12;
        rd_kafka_topic_partition_list_add(offsets, "srctopic64", 29)->offset =
            99999;

        cgmetadata = rd_kafka_consumer_group_metadata_new("mygroupid");

        TEST_CALL_ERROR__(rd_kafka_send_offsets_to_transaction(
            rk, offsets, cgmetadata, 20 * 1000));

        rd_kafka_consumer_group_metadata_destroy(cgmetadata);
        rd_kafka_topic_partition_list_destroy(offsets);


        /* Produce some messages */
        test_produce_msgs2(rk, topic, 0, RD_KAFKA_PARTITION_UA, 0, 10, NULL, 0);

        /* And commit the transaction */
        TEST_CALL_ERROR__(rd_kafka_commit_transaction(rk, -1));

        rd_kafka_destroy(rk);

        SUB_TEST_PASS();
}


/**
 * @brief Test fatal error handling when transactions are not supported
 *        by the broker.
 */
static void do_test_txns_not_supported(void) {
        rd_kafka_t *rk;
        rd_kafka_conf_t *conf;
        rd_kafka_mock_cluster_t *mcluster;
        rd_kafka_error_t *error;
        rd_kafka_resp_err_t err;

        SUB_TEST_QUICK();

        test_conf_init(&conf, NULL, 10);

        test_conf_set(conf, "transactional.id", "myxnid");
        test_conf_set(conf, "bootstrap.servers", ",");
        rd_kafka_conf_set_dr_msg_cb(conf, test_dr_msg_cb);

        rk = test_create_handle(RD_KAFKA_PRODUCER, conf);

        /* Create mock cluster */
        mcluster = rd_kafka_mock_cluster_new(rk, 3);

        /* Disable InitProducerId */
        rd_kafka_mock_set_apiversion(mcluster, 22 /*InitProducerId*/, -1, -1);


        rd_kafka_brokers_add(rk, rd_kafka_mock_cluster_bootstraps(mcluster));



        error = rd_kafka_init_transactions(rk, 5 * 1000);
        TEST_SAY("init_transactions() returned %s: %s\n",
                 error ? rd_kafka_error_name(error) : "success",
                 error ? rd_kafka_error_string(error) : "success");

        TEST_ASSERT(error, "Expected init_transactions() to fail");
        TEST_ASSERT(rd_kafka_error_code(error) ==
                        RD_KAFKA_RESP_ERR__UNSUPPORTED_FEATURE,
                    "Expected init_transactions() to fail with %s, not %s: %s",
                    rd_kafka_err2name(RD_KAFKA_RESP_ERR__UNSUPPORTED_FEATURE),
                    rd_kafka_error_name(error), rd_kafka_error_string(error));
        rd_kafka_error_destroy(error);

        err = rd_kafka_producev(rk, RD_KAFKA_V_TOPIC("test"),
                                RD_KAFKA_V_KEY("test", 4), RD_KAFKA_V_END);
        TEST_ASSERT(err == RD_KAFKA_RESP_ERR__FATAL,
                    "Expected producev() to fail with %s, not %s",
                    rd_kafka_err2name(RD_KAFKA_RESP_ERR__FATAL),
                    rd_kafka_err2name(err));

        rd_kafka_mock_cluster_destroy(mcluster);

        rd_kafka_destroy(rk);

        SUB_TEST_PASS();
}


/**
 * @brief CONCURRENT_TRANSACTION on AddOffsets.. should be retried.
 */
static void do_test_txns_send_offsets_concurrent_is_retried(void) {
        rd_kafka_t *rk;
        rd_kafka_mock_cluster_t *mcluster;
        rd_kafka_resp_err_t err;
        rd_kafka_topic_partition_list_t *offsets;
        rd_kafka_consumer_group_metadata_t *cgmetadata;

        SUB_TEST_QUICK();

        rk = create_txn_producer(&mcluster, "txnid", 3, NULL);

        test_curr->ignore_dr_err = rd_true;

        TEST_CALL_ERROR__(rd_kafka_init_transactions(rk, 5000));

        TEST_CALL_ERROR__(rd_kafka_begin_transaction(rk));

        err = rd_kafka_producev(rk, RD_KAFKA_V_TOPIC("mytopic"),
                                RD_KAFKA_V_VALUE("hi", 2), RD_KAFKA_V_END);
        TEST_ASSERT(!err, "produce failed: %s", rd_kafka_err2str(err));

        /* Wait for messages to be delivered */
        test_flush(rk, 5000);


        /*
         * Have AddOffsetsToTxn fail but eventually succeed due to
         * infinite retries.
         */
        rd_kafka_mock_push_request_errors(
            mcluster, RD_KAFKAP_AddOffsetsToTxn,
            1 + 5, /* first request + some retries */
            RD_KAFKA_RESP_ERR_CONCURRENT_TRANSACTIONS,
            RD_KAFKA_RESP_ERR_CONCURRENT_TRANSACTIONS,
            RD_KAFKA_RESP_ERR_CONCURRENT_TRANSACTIONS,
            RD_KAFKA_RESP_ERR_CONCURRENT_TRANSACTIONS,
            RD_KAFKA_RESP_ERR_CONCURRENT_TRANSACTIONS,
            RD_KAFKA_RESP_ERR_CONCURRENT_TRANSACTIONS);

        offsets = rd_kafka_topic_partition_list_new(1);
        rd_kafka_topic_partition_list_add(offsets, "srctopic4", 3)->offset = 12;

        cgmetadata = rd_kafka_consumer_group_metadata_new("mygroupid");

        TEST_CALL_ERROR__(
            rd_kafka_send_offsets_to_transaction(rk, offsets, cgmetadata, -1));

        rd_kafka_consumer_group_metadata_destroy(cgmetadata);
        rd_kafka_topic_partition_list_destroy(offsets);

        TEST_CALL_ERROR__(rd_kafka_commit_transaction(rk, 5000));

        /* All done */

        rd_kafka_destroy(rk);

        SUB_TEST_PASS();
}


/**
 * @brief Verify that send_offsets_to_transaction() with no eligible offsets
 *        is handled properly - the call should succeed immediately and be
 *        repeatable.
 */
static void do_test_txns_send_offsets_non_eligible(void) {
        rd_kafka_t *rk;
        rd_kafka_mock_cluster_t *mcluster;
        rd_kafka_resp_err_t err;
        rd_kafka_topic_partition_list_t *offsets;
        rd_kafka_consumer_group_metadata_t *cgmetadata;

        SUB_TEST_QUICK();

        rk = create_txn_producer(&mcluster, "txnid", 3, NULL);

        test_curr->ignore_dr_err = rd_true;

        TEST_CALL_ERROR__(rd_kafka_init_transactions(rk, 5000));

        TEST_CALL_ERROR__(rd_kafka_begin_transaction(rk));

        err = rd_kafka_producev(rk, RD_KAFKA_V_TOPIC("mytopic"),
                                RD_KAFKA_V_VALUE("hi", 2), RD_KAFKA_V_END);
        TEST_ASSERT(!err, "produce failed: %s", rd_kafka_err2str(err));

        /* Wait for messages to be delivered */
        test_flush(rk, 5000);

        /* Empty offsets list */
        offsets = rd_kafka_topic_partition_list_new(0);

        cgmetadata = rd_kafka_consumer_group_metadata_new("mygroupid");

        TEST_CALL_ERROR__(
            rd_kafka_send_offsets_to_transaction(rk, offsets, cgmetadata, -1));

        /* Now call it again, should also succeed. */
        TEST_CALL_ERROR__(
            rd_kafka_send_offsets_to_transaction(rk, offsets, cgmetadata, -1));

        rd_kafka_consumer_group_metadata_destroy(cgmetadata);
        rd_kafka_topic_partition_list_destroy(offsets);

        TEST_CALL_ERROR__(rd_kafka_commit_transaction(rk, 5000));

        /* All done */

        rd_kafka_destroy(rk);

        SUB_TEST_PASS();
}


/**
 * @brief Verify that request timeouts don't cause crash (#2913).
 */
static void do_test_txns_no_timeout_crash(void) {
        rd_kafka_t *rk;
        rd_kafka_mock_cluster_t *mcluster;
        rd_kafka_error_t *error;
        rd_kafka_resp_err_t err;
        rd_kafka_topic_partition_list_t *offsets;
        rd_kafka_consumer_group_metadata_t *cgmetadata;

        SUB_TEST_QUICK();

        rk =
            create_txn_producer(&mcluster, "txnid", 3, "socket.timeout.ms",
                                "1000", "transaction.timeout.ms", "5000", NULL);

        TEST_CALL_ERROR__(rd_kafka_init_transactions(rk, 5000));

        TEST_CALL_ERROR__(rd_kafka_begin_transaction(rk));

        err = rd_kafka_producev(rk, RD_KAFKA_V_TOPIC("mytopic"),
                                RD_KAFKA_V_VALUE("hi", 2), RD_KAFKA_V_END);
        TEST_ASSERT(!err, "produce failed: %s", rd_kafka_err2str(err));

        test_flush(rk, -1);

        /* Delay all broker connections */
        if ((err = rd_kafka_mock_broker_set_rtt(mcluster, 1, 2000)) ||
            (err = rd_kafka_mock_broker_set_rtt(mcluster, 2, 2000)) ||
            (err = rd_kafka_mock_broker_set_rtt(mcluster, 3, 2000)))
                TEST_FAIL("Failed to set broker RTT: %s",
                          rd_kafka_err2str(err));

        /* send_offsets..() should now time out */
        offsets = rd_kafka_topic_partition_list_new(1);
        rd_kafka_topic_partition_list_add(offsets, "srctopic4", 3)->offset = 12;
        cgmetadata = rd_kafka_consumer_group_metadata_new("mygroupid");

        error =
            rd_kafka_send_offsets_to_transaction(rk, offsets, cgmetadata, -1);
        TEST_ASSERT(error, "Expected send_offsets..() to fail");
        TEST_SAY("send_offsets..() failed with %serror: %s\n",
                 rd_kafka_error_is_retriable(error) ? "retriable " : "",
                 rd_kafka_error_string(error));
        TEST_ASSERT(rd_kafka_error_code(error) == RD_KAFKA_RESP_ERR__TIMED_OUT,
                    "expected send_offsets_to_transaction() to fail with "
                    "timeout, not %s",
                    rd_kafka_error_name(error));
        TEST_ASSERT(rd_kafka_error_is_retriable(error),
                    "expected send_offsets_to_transaction() to fail with "
                    "a retriable error");
        rd_kafka_error_destroy(error);

        /* Reset delay and try again */
        if ((err = rd_kafka_mock_broker_set_rtt(mcluster, 1, 0)) ||
            (err = rd_kafka_mock_broker_set_rtt(mcluster, 2, 0)) ||
            (err = rd_kafka_mock_broker_set_rtt(mcluster, 3, 0)))
                TEST_FAIL("Failed to reset broker RTT: %s",
                          rd_kafka_err2str(err));

        TEST_SAY("Retrying send_offsets..()\n");
        error =
            rd_kafka_send_offsets_to_transaction(rk, offsets, cgmetadata, -1);
        TEST_ASSERT(!error, "Expected send_offsets..() to succeed, got: %s",
                    rd_kafka_error_string(error));

        rd_kafka_consumer_group_metadata_destroy(cgmetadata);
        rd_kafka_topic_partition_list_destroy(offsets);

        /* All done */
        rd_kafka_destroy(rk);

        SUB_TEST_PASS();
}


/**
 * @brief Test auth failure handling.
 */
static void do_test_txn_auth_failure(int16_t ApiKey,
                                     rd_kafka_resp_err_t ErrorCode) {
        rd_kafka_t *rk;
        rd_kafka_mock_cluster_t *mcluster;
        rd_kafka_error_t *error;

        SUB_TEST_QUICK("ApiKey=%s ErrorCode=%s", rd_kafka_ApiKey2str(ApiKey),
                       rd_kafka_err2name(ErrorCode));

        rk = create_txn_producer(&mcluster, "txnid", 3, NULL);

        rd_kafka_mock_push_request_errors(mcluster, ApiKey, 1, ErrorCode);

        error = rd_kafka_init_transactions(rk, 5000);
        TEST_ASSERT(error, "Expected init_transactions() to fail");

        TEST_SAY("init_transactions() failed: %s: %s\n",
                 rd_kafka_err2name(rd_kafka_error_code(error)),
                 rd_kafka_error_string(error));
        TEST_ASSERT(rd_kafka_error_code(error) == ErrorCode,
                    "Expected error %s, not %s", rd_kafka_err2name(ErrorCode),
                    rd_kafka_err2name(rd_kafka_error_code(error)));
        TEST_ASSERT(rd_kafka_error_is_fatal(error),
                    "Expected error to be fatal");
        TEST_ASSERT(!rd_kafka_error_is_retriable(error),
                    "Expected error to not be retriable");
        rd_kafka_error_destroy(error);

        /* All done */

        rd_kafka_destroy(rk);

        SUB_TEST_PASS();
}


/**
 * @brief Issue #3041: Commit fails due to message flush() taking too long,
 *        eventually resulting in an unabortable error and failure to
 *        re-init the transactional producer.
 */
static void do_test_txn_flush_timeout(void) {
        rd_kafka_t *rk;
        rd_kafka_mock_cluster_t *mcluster;
        rd_kafka_topic_partition_list_t *offsets;
        rd_kafka_consumer_group_metadata_t *cgmetadata;
        rd_kafka_error_t *error;
        const char *txnid      = "myTxnId";
        const char *topic      = "myTopic";
        const int32_t coord_id = 2;
        int msgcounter         = 0;
        rd_bool_t is_retry     = rd_false;

        SUB_TEST_QUICK();

        rk = create_txn_producer(&mcluster, txnid, 3, "message.timeout.ms",
                                 "10000", "transaction.timeout.ms", "10000",
                                 /* Speed up coordinator reconnect */
                                 "reconnect.backoff.max.ms", "1000", NULL);


        /* Broker down is not a test-failing error */
        test_curr->is_fatal_cb = error_is_fatal_cb;
        allowed_error          = RD_KAFKA_RESP_ERR__TRANSPORT;

        rd_kafka_mock_topic_create(mcluster, topic, 2, 3);

        /* Set coordinator so we can disconnect it later */
        rd_kafka_mock_coordinator_set(mcluster, "transaction", txnid, coord_id);

        /*
         * Init transactions
         */
        TEST_CALL_ERROR__(rd_kafka_init_transactions(rk, 5000));

retry:
        if (!is_retry) {
                /* First attempt should fail. */

                test_curr->ignore_dr_err = rd_true;
                test_curr->exp_dr_err    = RD_KAFKA_RESP_ERR__MSG_TIMED_OUT;

                /* Assign invalid partition leaders for some partitions so
                 * that messages will not be delivered. */
                rd_kafka_mock_partition_set_leader(mcluster, topic, 0, -1);
                rd_kafka_mock_partition_set_leader(mcluster, topic, 1, -1);

        } else {
                /* The retry should succeed */
                test_curr->ignore_dr_err = rd_false;
                test_curr->exp_dr_err    = is_retry
                                            ? RD_KAFKA_RESP_ERR_NO_ERROR
                                            : RD_KAFKA_RESP_ERR__MSG_TIMED_OUT;

                rd_kafka_mock_partition_set_leader(mcluster, topic, 0, 1);
                rd_kafka_mock_partition_set_leader(mcluster, topic, 1, 1);
        }


        /*
         * Start a transaction
         */
        TEST_CALL_ERROR__(rd_kafka_begin_transaction(rk));

        /*
         * Produce some messages to specific partitions and random.
         */
        test_produce_msgs2_nowait(rk, topic, 0, 0, 0, 100, NULL, 10,
                                  &msgcounter);
        test_produce_msgs2_nowait(rk, topic, 1, 0, 0, 100, NULL, 10,
                                  &msgcounter);
        test_produce_msgs2_nowait(rk, topic, RD_KAFKA_PARTITION_UA, 0, 0, 100,
                                  NULL, 10, &msgcounter);


        /*
         * Send some arbitrary offsets.
         */
        offsets = rd_kafka_topic_partition_list_new(4);
        rd_kafka_topic_partition_list_add(offsets, "srctopic4", 3)->offset = 12;
        rd_kafka_topic_partition_list_add(offsets, "srctopic64", 49)->offset =
            999999111;
        rd_kafka_topic_partition_list_add(offsets, "srctopic4", 0)->offset =
            999;
        rd_kafka_topic_partition_list_add(offsets, "srctopic64", 34)->offset =
            123456789;

        cgmetadata = rd_kafka_consumer_group_metadata_new("mygroupid");

        TEST_CALL_ERROR__(
            rd_kafka_send_offsets_to_transaction(rk, offsets, cgmetadata, -1));

        rd_kafka_consumer_group_metadata_destroy(cgmetadata);
        rd_kafka_topic_partition_list_destroy(offsets);

        rd_sleep(2);

        if (!is_retry) {
                /* Now disconnect the coordinator. */
                TEST_SAY("Disconnecting transaction coordinator %" PRId32 "\n",
                         coord_id);
                rd_kafka_mock_broker_set_down(mcluster, coord_id);
        }

        /*
         * Start committing.
         */
        error = rd_kafka_commit_transaction(rk, -1);

        if (!is_retry) {
                TEST_ASSERT(error != NULL, "Expected commit to fail");
                TEST_SAY("commit_transaction() failed (expectedly): %s\n",
                         rd_kafka_error_string(error));
                rd_kafka_error_destroy(error);

        } else {
                TEST_ASSERT(!error, "Expected commit to succeed, not: %s",
                            rd_kafka_error_string(error));
        }

        if (!is_retry) {
                /*
                 * Bring the coordinator back up.
                 */
                rd_kafka_mock_broker_set_up(mcluster, coord_id);
                rd_sleep(2);

                /*
                 * Abort, and try again, this time without error.
                 */
                TEST_SAY("Aborting and retrying\n");
                is_retry = rd_true;

                TEST_CALL_ERROR__(rd_kafka_abort_transaction(rk, 60000));
                goto retry;
        }

        /* All done */

        rd_kafka_destroy(rk);

        SUB_TEST_PASS();
}


/**
 * @brief ESC-4424: rko is reused in response handler after destroy in coord_req
 *        sender due to bad state.
 *
 * This is somewhat of a race condition so we need to perform a couple of
 * iterations before it hits, usually 2 or 3, so we try at least 15 times.
 */
static void do_test_txn_coord_req_destroy(void) {
        rd_kafka_t *rk;
        rd_kafka_mock_cluster_t *mcluster;
        int i;
        int errcnt = 0;

        SUB_TEST();

        rk = create_txn_producer(&mcluster, "txnid", 3, NULL);

        test_curr->ignore_dr_err = rd_true;

        TEST_CALL_ERROR__(rd_kafka_init_transactions(rk, 5000));

        for (i = 0; i < 15; i++) {
                rd_kafka_error_t *error;
                rd_kafka_resp_err_t err;
                rd_kafka_topic_partition_list_t *offsets;
                rd_kafka_consumer_group_metadata_t *cgmetadata;

                test_timeout_set(10);

                TEST_CALL_ERROR__(rd_kafka_begin_transaction(rk));

                /*
                 * Inject errors to trigger retries
                 */
                rd_kafka_mock_push_request_errors(
                    mcluster, RD_KAFKAP_AddPartitionsToTxn,
                    2, /* first request + number of internal retries */
                    RD_KAFKA_RESP_ERR_CONCURRENT_TRANSACTIONS,
                    RD_KAFKA_RESP_ERR_CONCURRENT_TRANSACTIONS);

                rd_kafka_mock_push_request_errors(
                    mcluster, RD_KAFKAP_AddOffsetsToTxn,
                    1, /* first request + number of internal retries */
                    RD_KAFKA_RESP_ERR_CONCURRENT_TRANSACTIONS);

                err = rd_kafka_producev(rk, RD_KAFKA_V_TOPIC("mytopic"),
                                        RD_KAFKA_V_VALUE("hi", 2),
                                        RD_KAFKA_V_END);
                TEST_ASSERT(!err, "produce failed: %s", rd_kafka_err2str(err));

                rd_kafka_mock_push_request_errors(
                    mcluster, RD_KAFKAP_Produce, 4,
                    RD_KAFKA_RESP_ERR_REQUEST_TIMED_OUT,
                    RD_KAFKA_RESP_ERR_REQUEST_TIMED_OUT,
                    RD_KAFKA_RESP_ERR_TOPIC_AUTHORIZATION_FAILED,
                    RD_KAFKA_RESP_ERR_TOPIC_AUTHORIZATION_FAILED);
                /* FIXME: When KIP-360 is supported, add this error:
                 *        RD_KAFKA_RESP_ERR_OUT_OF_ORDER_SEQUENCE_NUMBER */

                err = rd_kafka_producev(rk, RD_KAFKA_V_TOPIC("mytopic"),
                                        RD_KAFKA_V_VALUE("hi", 2),
                                        RD_KAFKA_V_END);
                TEST_ASSERT(!err, "produce failed: %s", rd_kafka_err2str(err));


                /*
                 * Send offsets to transaction
                 */

                offsets = rd_kafka_topic_partition_list_new(1);
                rd_kafka_topic_partition_list_add(offsets, "srctopic4", 3)
                    ->offset = 12;

                cgmetadata = rd_kafka_consumer_group_metadata_new("mygroupid");

                error = rd_kafka_send_offsets_to_transaction(rk, offsets,
                                                             cgmetadata, -1);

                TEST_SAY("send_offsets_to_transaction() #%d: %s\n", i,
                         rd_kafka_error_string(error));

                /* As we can't control the exact timing and sequence
                 * of requests this sometimes fails and sometimes succeeds,
                 * but we run the test enough times to trigger at least
                 * one failure. */
                if (error) {
                        TEST_SAY(
                            "send_offsets_to_transaction() #%d "
                            "failed (expectedly): %s\n",
                            i, rd_kafka_error_string(error));
                        TEST_ASSERT(rd_kafka_error_txn_requires_abort(error),
                                    "Expected abortable error for #%d", i);
                        rd_kafka_error_destroy(error);
                        errcnt++;
                }

                rd_kafka_consumer_group_metadata_destroy(cgmetadata);
                rd_kafka_topic_partition_list_destroy(offsets);

                /* Allow time for internal retries */
                rd_sleep(2);

                TEST_CALL_ERROR__(rd_kafka_abort_transaction(rk, 5000));
        }

        TEST_ASSERT(errcnt > 0,
                    "Expected at least one send_offets_to_transaction() "
                    "failure");

        /* All done */

        rd_kafka_destroy(rk);
}


static rd_atomic32_t multi_find_req_cnt;

static rd_kafka_resp_err_t
multi_find_on_response_received_cb(rd_kafka_t *rk,
                                   int sockfd,
                                   const char *brokername,
                                   int32_t brokerid,
                                   int16_t ApiKey,
                                   int16_t ApiVersion,
                                   int32_t CorrId,
                                   size_t size,
                                   int64_t rtt,
                                   rd_kafka_resp_err_t err,
                                   void *ic_opaque) {
        rd_kafka_mock_cluster_t *mcluster = rd_kafka_handle_mock_cluster(rk);
        rd_bool_t done = rd_atomic32_get(&multi_find_req_cnt) > 10000;

        if (ApiKey != RD_KAFKAP_AddOffsetsToTxn || done)
                return RD_KAFKA_RESP_ERR_NO_ERROR;

        TEST_SAY("on_response_received_cb: %s: %s: brokerid %" PRId32
                 ", ApiKey %hd, CorrId %d, rtt %.2fms, %s: %s\n",
                 rd_kafka_name(rk), brokername, brokerid, ApiKey, CorrId,
                 rtt != -1 ? (float)rtt / 1000.0 : 0.0,
                 done ? "already done" : "not done yet",
                 rd_kafka_err2name(err));


        if (rd_atomic32_add(&multi_find_req_cnt, 1) == 1) {
                /* Trigger a broker down/up event, which in turns
                 * triggers the coord_req_fsm(). */
                rd_kafka_mock_broker_set_down(mcluster, 2);
                rd_kafka_mock_broker_set_up(mcluster, 2);
                return RD_KAFKA_RESP_ERR_NO_ERROR;
        }

        /* Trigger a broker down/up event, which in turns
         * triggers the coord_req_fsm(). */
        rd_kafka_mock_broker_set_down(mcluster, 3);
        rd_kafka_mock_broker_set_up(mcluster, 3);

        /* Clear the downed broker's latency so that it reconnects
         * quickly, otherwise the ApiVersionRequest will be delayed and
         * this will in turn delay the -> UP transition that we need to
         * trigger the coord_reqs. */
        rd_kafka_mock_broker_set_rtt(mcluster, 3, 0);

        /* Only do this down/up once */
        rd_atomic32_add(&multi_find_req_cnt, 10000);

        return RD_KAFKA_RESP_ERR_NO_ERROR;
}


/**
 * @brief ESC-4444: multiple FindCoordinatorRequests are sent referencing
 *        the same coord_req_t, but the first one received will destroy
 *        the coord_req_t object and make the subsequent FindCoordingResponses
 *        reference a freed object.
 *
 * What we want to achieve is this sequence:
 *  1. AddOffsetsToTxnRequest + Response which..
 *  2. Triggers TxnOffsetCommitRequest, but the coordinator is not known, so..
 *  3. Triggers a FindCoordinatorRequest
 *  4. FindCoordinatorResponse from 3 is received ..
 *  5. A TxnOffsetCommitRequest is sent from coord_req_fsm().
 *  6. Another broker changing state to Up triggers coord reqs again, which..
 *  7. Triggers a second TxnOffsetCommitRequest from coord_req_fsm().
 *  7. FindCoordinatorResponse from 5 is received, references the destroyed rko
 *     and crashes.
 */
static void do_test_txn_coord_req_multi_find(void) {
        rd_kafka_t *rk;
        rd_kafka_mock_cluster_t *mcluster;
        rd_kafka_error_t *error;
        rd_kafka_resp_err_t err;
        rd_kafka_topic_partition_list_t *offsets;
        rd_kafka_consumer_group_metadata_t *cgmetadata;
        const char *txnid = "txnid", *groupid = "mygroupid", *topic = "mytopic";
        int i;

        SUB_TEST();

        rd_atomic32_init(&multi_find_req_cnt, 0);

        on_response_received_cb = multi_find_on_response_received_cb;
        rk                      = create_txn_producer(&mcluster, txnid, 3,
                                 /* Need connections to all brokers so we
                                  * can trigger coord_req_fsm events
                                  * by toggling connections. */
                                 "enable.sparse.connections", "false",
                                 /* Set up on_response_received interceptor */
                                 "on_response_received", "", NULL);

        /* Let broker 1 be both txn and group coordinator
         * so that the group coordinator connection is up when it is time
         * send the TxnOffsetCommitRequest. */
        rd_kafka_mock_coordinator_set(mcluster, "transaction", txnid, 1);
        rd_kafka_mock_coordinator_set(mcluster, "group", groupid, 1);

        /* Set broker 1, 2, and 3 as leaders for a partition each and
         * later produce to both partitions so we know there's a connection
         * to all brokers. */
        rd_kafka_mock_topic_create(mcluster, topic, 3, 1);
        rd_kafka_mock_partition_set_leader(mcluster, topic, 0, 1);
        rd_kafka_mock_partition_set_leader(mcluster, topic, 1, 2);
        rd_kafka_mock_partition_set_leader(mcluster, topic, 2, 3);

        /* Broker down is not a test-failing error */
        allowed_error          = RD_KAFKA_RESP_ERR__TRANSPORT;
        test_curr->is_fatal_cb = error_is_fatal_cb;

        TEST_CALL_ERROR__(rd_kafka_init_transactions(rk, 5000));

        TEST_CALL_ERROR__(rd_kafka_begin_transaction(rk));

        for (i = 0; i < 3; i++) {
                err = rd_kafka_producev(
                    rk, RD_KAFKA_V_TOPIC(topic), RD_KAFKA_V_PARTITION(i),
                    RD_KAFKA_V_VALUE("hi", 2), RD_KAFKA_V_END);
                TEST_ASSERT(!err, "produce failed: %s", rd_kafka_err2str(err));
        }

        test_flush(rk, 5000);

        /*
         * send_offsets_to_transaction() will query for the group coordinator,
         * we need to make those requests slow so that multiple requests are
         * sent.
         */
        for (i = 1; i <= 3; i++)
                rd_kafka_mock_broker_set_rtt(mcluster, (int32_t)i, 4000);

        /*
         * Send offsets to transaction
         */

        offsets = rd_kafka_topic_partition_list_new(1);
        rd_kafka_topic_partition_list_add(offsets, "srctopic4", 3)->offset = 12;

        cgmetadata = rd_kafka_consumer_group_metadata_new(groupid);

        error =
            rd_kafka_send_offsets_to_transaction(rk, offsets, cgmetadata, -1);

        TEST_SAY("send_offsets_to_transaction() %s\n",
                 rd_kafka_error_string(error));
        TEST_ASSERT(!error, "send_offsets_to_transaction() failed: %s",
                    rd_kafka_error_string(error));

        rd_kafka_consumer_group_metadata_destroy(cgmetadata);
        rd_kafka_topic_partition_list_destroy(offsets);

        /* Clear delay */
        for (i = 1; i <= 3; i++)
                rd_kafka_mock_broker_set_rtt(mcluster, (int32_t)i, 0);

        rd_sleep(5);

        TEST_CALL_ERROR__(rd_kafka_commit_transaction(rk, 5000));

        /* All done */

        TEST_ASSERT(rd_atomic32_get(&multi_find_req_cnt) > 10000,
                    "on_request_sent interceptor did not trigger properly");

        rd_kafka_destroy(rk);

        on_response_received_cb = NULL;

        SUB_TEST_PASS();
}


/**
 * @brief ESC-4410: adding producer partitions gradually will trigger multiple
 *        AddPartitionsToTxn requests. Due to a bug the third partition to be
 *        registered would hang in PEND_TXN state.
 *
 * Trigger this behaviour by having two outstanding AddPartitionsToTxn requests
 * at the same time, followed by a need for a third:
 *
 * 1. Set coordinator broker rtt high (to give us time to produce).
 * 2. Produce to partition 0, will trigger first AddPartitionsToTxn.
 * 3. Produce to partition 1, will trigger second AddPartitionsToTxn.
 * 4. Wait for second AddPartitionsToTxn response.
 * 5. Produce to partition 2, should trigger AddPartitionsToTxn, but bug
 *    causes it to be stale in pending state.
 */

static rd_atomic32_t multi_addparts_resp_cnt;
static rd_kafka_resp_err_t
multi_addparts_response_received_cb(rd_kafka_t *rk,
                                    int sockfd,
                                    const char *brokername,
                                    int32_t brokerid,
                                    int16_t ApiKey,
                                    int16_t ApiVersion,
                                    int32_t CorrId,
                                    size_t size,
                                    int64_t rtt,
                                    rd_kafka_resp_err_t err,
                                    void *ic_opaque) {

        if (ApiKey == RD_KAFKAP_AddPartitionsToTxn) {
                TEST_SAY("on_response_received_cb: %s: %s: brokerid %" PRId32
                         ", ApiKey %hd, CorrId %d, rtt %.2fms, count %" PRId32
                         ": %s\n",
                         rd_kafka_name(rk), brokername, brokerid, ApiKey,
                         CorrId, rtt != -1 ? (float)rtt / 1000.0 : 0.0,
                         rd_atomic32_get(&multi_addparts_resp_cnt),
                         rd_kafka_err2name(err));

                rd_atomic32_add(&multi_addparts_resp_cnt, 1);
        }

        return RD_KAFKA_RESP_ERR_NO_ERROR;
}


static void do_test_txn_addparts_req_multi(void) {
        rd_kafka_t *rk;
        rd_kafka_mock_cluster_t *mcluster;
        const char *txnid = "txnid", *topic = "mytopic";
        int32_t txn_coord = 2;

        SUB_TEST();

        rd_atomic32_init(&multi_addparts_resp_cnt, 0);

        on_response_received_cb = multi_addparts_response_received_cb;
        rk = create_txn_producer(&mcluster, txnid, 3, "linger.ms", "0",
                                 "message.timeout.ms", "9000",
                                 /* Set up on_response_received interceptor */
                                 "on_response_received", "", NULL);

        /* Let broker 1 be txn coordinator. */
        rd_kafka_mock_coordinator_set(mcluster, "transaction", txnid,
                                      txn_coord);

        rd_kafka_mock_topic_create(mcluster, topic, 3, 1);

        /* Set partition leaders to non-txn-coord broker so they wont
         * be affected by rtt delay */
        rd_kafka_mock_partition_set_leader(mcluster, topic, 0, 1);
        rd_kafka_mock_partition_set_leader(mcluster, topic, 1, 1);
        rd_kafka_mock_partition_set_leader(mcluster, topic, 2, 1);



        TEST_CALL_ERROR__(rd_kafka_init_transactions(rk, 5000));

        /*
         * Run one transaction first to let the client familiarize with
         * the topic, this avoids metadata lookups, etc, when the real
         * test is run.
         */
        TEST_SAY("Running seed transaction\n");
        TEST_CALL_ERROR__(rd_kafka_begin_transaction(rk));
        TEST_CALL_ERR__(rd_kafka_producev(rk, RD_KAFKA_V_TOPIC(topic),
                                          RD_KAFKA_V_VALUE("seed", 4),
                                          RD_KAFKA_V_END));
        TEST_CALL_ERROR__(rd_kafka_commit_transaction(rk, 5000));


        /*
         * Now perform test transaction with rtt delays
         */
        TEST_SAY("Running test transaction\n");

        TEST_CALL_ERROR__(rd_kafka_begin_transaction(rk));

        /* Reset counter */
        rd_atomic32_set(&multi_addparts_resp_cnt, 0);

        /* Add latency to txn coordinator so we can pace our produce() calls */
        rd_kafka_mock_broker_set_rtt(mcluster, txn_coord, 1000);

        /* Produce to partition 0 */
        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));

        rd_usleep(500 * 1000, NULL);

        /* Produce to partition 1 */
        TEST_CALL_ERR__(rd_kafka_producev(
            rk, RD_KAFKA_V_TOPIC(topic), RD_KAFKA_V_PARTITION(1),
            RD_KAFKA_V_VALUE("hi", 2), RD_KAFKA_V_END));

        TEST_SAY("Waiting for two AddPartitionsToTxnResponse\n");
        while (rd_atomic32_get(&multi_addparts_resp_cnt) < 2)
                rd_usleep(10 * 1000, NULL);

        TEST_SAY("%" PRId32 " AddPartitionsToTxnResponses seen\n",
                 rd_atomic32_get(&multi_addparts_resp_cnt));

        /* Produce to partition 2, this message will hang in
         * queue if the bug is not fixed. */
        TEST_CALL_ERR__(rd_kafka_producev(
            rk, RD_KAFKA_V_TOPIC(topic), RD_KAFKA_V_PARTITION(2),
            RD_KAFKA_V_VALUE("hi", 2), RD_KAFKA_V_END));

        /* Allow some extra time for things to settle before committing
         * transaction. */
        rd_usleep(1000 * 1000, NULL);

        TEST_CALL_ERROR__(rd_kafka_commit_transaction(rk, 10 * 1000));

        /* All done */
        rd_kafka_destroy(rk);

        on_response_received_cb = NULL;

        SUB_TEST_PASS();
}



/**
 * @brief Test handling of OffsetFetchRequest returning UNSTABLE_OFFSET_COMMIT.
 *
 * There are two things to test;
 *  - OffsetFetch triggered by committed() (and similar code paths)
 *  - OffsetFetch triggered by assign()
 */
static void do_test_unstable_offset_commit(void) {
        rd_kafka_t *rk, *c;
        rd_kafka_conf_t *c_conf;
        rd_kafka_mock_cluster_t *mcluster;
        rd_kafka_topic_partition_list_t *offsets;
        const char *topic              = "srctopic4";
        const int msgcnt               = 100;
        const int64_t offset_to_commit = msgcnt / 2;
        int i;
        int remains = 0;

        SUB_TEST_QUICK();

        rk = create_txn_producer(&mcluster, "txnid", 3, NULL);

        test_conf_init(&c_conf, NULL, 0);
        test_conf_set(c_conf, "security.protocol", "PLAINTEXT");
        test_conf_set(c_conf, "bootstrap.servers",
                      rd_kafka_mock_cluster_bootstraps(mcluster));
        test_conf_set(c_conf, "enable.partition.eof", "true");
        test_conf_set(c_conf, "auto.offset.reset", "error");
        c = test_create_consumer("mygroup", NULL, c_conf, NULL);

        rd_kafka_mock_topic_create(mcluster, topic, 2, 3);

        /* Produce some messages to the topic so that the consumer has
         * something to read. */
        TEST_CALL_ERROR__(rd_kafka_init_transactions(rk, -1));
        TEST_CALL_ERROR__(rd_kafka_begin_transaction(rk));
        test_produce_msgs2_nowait(rk, topic, 0, 0, 0, msgcnt, NULL, 0,
                                  &remains);
        TEST_CALL_ERROR__(rd_kafka_commit_transaction(rk, -1));


        /* Commit offset */
        offsets = rd_kafka_topic_partition_list_new(1);
        rd_kafka_topic_partition_list_add(offsets, topic, 0)->offset =
            offset_to_commit;
        TEST_CALL_ERR__(rd_kafka_commit(c, offsets, 0 /*sync*/));
        rd_kafka_topic_partition_list_destroy(offsets);

        /* Retrieve offsets by calling committed().
         *
         * Have OffsetFetch fail and retry, on the first iteration
         * the API timeout is higher than the amount of time the retries will
         * take and thus succeed, and on the second iteration the timeout
         * will be lower and thus fail. */
        for (i = 0; i < 2; i++) {
                rd_kafka_resp_err_t err;
                rd_kafka_resp_err_t exp_err =
                    i == 0 ? RD_KAFKA_RESP_ERR_NO_ERROR
                           : RD_KAFKA_RESP_ERR__TIMED_OUT;
                int timeout_ms = exp_err ? 200 : 5 * 1000;

                rd_kafka_mock_push_request_errors(
                    mcluster, RD_KAFKAP_OffsetFetch,
                    1 + 5, /* first request + some retries */
                    RD_KAFKA_RESP_ERR_UNSTABLE_OFFSET_COMMIT,
                    RD_KAFKA_RESP_ERR_UNSTABLE_OFFSET_COMMIT,
                    RD_KAFKA_RESP_ERR_UNSTABLE_OFFSET_COMMIT,
                    RD_KAFKA_RESP_ERR_UNSTABLE_OFFSET_COMMIT,
                    RD_KAFKA_RESP_ERR_UNSTABLE_OFFSET_COMMIT,
                    RD_KAFKA_RESP_ERR_UNSTABLE_OFFSET_COMMIT);

                offsets = rd_kafka_topic_partition_list_new(1);
                rd_kafka_topic_partition_list_add(offsets, topic, 0);

                err = rd_kafka_committed(c, offsets, timeout_ms);

                TEST_SAY("#%d: committed() returned %s (expected %s)\n", i,
                         rd_kafka_err2name(err), rd_kafka_err2name(exp_err));

                TEST_ASSERT(err == exp_err,
                            "#%d: Expected committed() to return %s, not %s", i,
                            rd_kafka_err2name(exp_err), rd_kafka_err2name(err));
                TEST_ASSERT(offsets->cnt == 1,
                            "Expected 1 committed offset, not %d",
                            offsets->cnt);
                if (!exp_err)
                        TEST_ASSERT(offsets->elems[0].offset ==
                                        offset_to_commit,
                                    "Expected committed offset %" PRId64
                                    ", "
                                    "not %" PRId64,
                                    offset_to_commit, offsets->elems[0].offset);
                else
                        TEST_ASSERT(offsets->elems[0].offset < 0,
                                    "Expected no committed offset, "
                                    "not %" PRId64,
                                    offsets->elems[0].offset);

                rd_kafka_topic_partition_list_destroy(offsets);
        }

        TEST_SAY("Phase 2: OffsetFetch lookup through assignment\n");
        offsets = rd_kafka_topic_partition_list_new(1);
        rd_kafka_topic_partition_list_add(offsets, topic, 0)->offset =
            RD_KAFKA_OFFSET_STORED;

        rd_kafka_mock_push_request_errors(
            mcluster, RD_KAFKAP_OffsetFetch,
            1 + 5, /* first request + some retries */
            RD_KAFKA_RESP_ERR_UNSTABLE_OFFSET_COMMIT,
            RD_KAFKA_RESP_ERR_UNSTABLE_OFFSET_COMMIT,
            RD_KAFKA_RESP_ERR_UNSTABLE_OFFSET_COMMIT,
            RD_KAFKA_RESP_ERR_UNSTABLE_OFFSET_COMMIT,
            RD_KAFKA_RESP_ERR_UNSTABLE_OFFSET_COMMIT,
            RD_KAFKA_RESP_ERR_UNSTABLE_OFFSET_COMMIT);

        test_consumer_incremental_assign("assign", c, offsets);
        rd_kafka_topic_partition_list_destroy(offsets);

        test_consumer_poll_exact("consume", c, 0, 1 /*eof*/, 0, msgcnt / 2,
                                 rd_true /*exact counts*/, NULL);

        /* All done */
        rd_kafka_destroy(c);
        rd_kafka_destroy(rk);

        SUB_TEST_PASS();
}


/**
 * @brief If a message times out locally before being attempted to send
 *        and commit_transaction() is called, the transaction must not succeed.
 *        https://github.com/confluentinc/confluent-kafka-dotnet/issues/1568
 */
static void do_test_commit_after_msg_timeout(void) {
        rd_kafka_t *rk;
        rd_kafka_mock_cluster_t *mcluster;
        int32_t coord_id, leader_id;
        rd_kafka_resp_err_t err;
        rd_kafka_error_t *error;
        const char *topic            = "test";
        const char *transactional_id = "txnid";
        int remains                  = 0;

        SUB_TEST_QUICK();

        /* Assign coordinator and leader to two different brokers */
        coord_id  = 1;
        leader_id = 2;

        rk = create_txn_producer(&mcluster, transactional_id, 3,
                                 "message.timeout.ms", "5000",
                                 "transaction.timeout.ms", "10000", NULL);

        /* Broker down is not a test-failing error */
        allowed_error          = RD_KAFKA_RESP_ERR__TRANSPORT;
        test_curr->is_fatal_cb = error_is_fatal_cb;
        test_curr->exp_dr_err  = RD_KAFKA_RESP_ERR__MSG_TIMED_OUT;

        err = rd_kafka_mock_topic_create(mcluster, topic, 1, 3);
        TEST_ASSERT(!err, "Failed to create topic: %s", rd_kafka_err2str(err));

        rd_kafka_mock_coordinator_set(mcluster, "transaction", transactional_id,
                                      coord_id);
        rd_kafka_mock_partition_set_leader(mcluster, topic, 0, leader_id);

        /* Start transactioning */
        TEST_SAY("Starting transaction\n");
        TEST_CALL_ERROR__(rd_kafka_init_transactions(rk, -1));

        TEST_CALL_ERROR__(rd_kafka_begin_transaction(rk));

        TEST_SAY("Bringing down %" PRId32 "\n", leader_id);
        rd_kafka_mock_broker_set_down(mcluster, leader_id);
        rd_kafka_mock_broker_set_down(mcluster, coord_id);

        test_produce_msgs2_nowait(rk, topic, 0, 0, 0, 1, NULL, 0, &remains);

        error = rd_kafka_commit_transaction(rk, -1);
        TEST_ASSERT(error != NULL, "expected commit_transaciton() to fail");
        TEST_SAY_ERROR(error, "commit_transaction() failed (as expected): ");
        TEST_ASSERT(rd_kafka_error_txn_requires_abort(error),
                    "Expected txn_requires_abort error");
        rd_kafka_error_destroy(error);

        /* Bring the brokers up so the abort can complete */
        rd_kafka_mock_broker_set_up(mcluster, coord_id);
        rd_kafka_mock_broker_set_up(mcluster, leader_id);

        TEST_SAY("Aborting transaction\n");
        TEST_CALL_ERROR__(rd_kafka_abort_transaction(rk, -1));

        TEST_ASSERT(remains == 0, "%d message(s) were not flushed\n", remains);

        TEST_SAY("Attempting second transaction, which should succeed\n");
        test_curr->is_fatal_cb = error_is_fatal_cb;
        test_curr->exp_dr_err  = RD_KAFKA_RESP_ERR_NO_ERROR;

        TEST_CALL_ERROR__(rd_kafka_begin_transaction(rk));
        test_produce_msgs2_nowait(rk, topic, 0, 0, 0, 1, NULL, 0, &remains);

        TEST_CALL_ERROR__(rd_kafka_commit_transaction(rk, -1));

        TEST_ASSERT(remains == 0, "%d message(s) were not produced\n", remains);

        rd_kafka_destroy(rk);

        allowed_error          = RD_KAFKA_RESP_ERR_NO_ERROR;
        test_curr->is_fatal_cb = NULL;

        SUB_TEST_PASS();
}


/**
 * @brief #3575: Verify that OUT_OF_ORDER_SEQ does not trigger an epoch bump
 *        during an ongoing transaction.
 *        The transaction should instead enter the abortable state.
 */
static void do_test_out_of_order_seq(void) {
        rd_kafka_t *rk;
        rd_kafka_mock_cluster_t *mcluster;
        rd_kafka_error_t *error;
        int32_t txn_coord = 1, leader = 2;
        const char *txnid = "myTxnId";
        test_timing_t timing;
        rd_kafka_resp_err_t err;

        SUB_TEST_QUICK();

        rk = create_txn_producer(&mcluster, txnid, 3, "batch.num.messages", "1",
                                 NULL);

        rd_kafka_mock_coordinator_set(mcluster, "transaction", txnid,
                                      txn_coord);

        rd_kafka_mock_partition_set_leader(mcluster, "mytopic", 0, leader);

        test_curr->ignore_dr_err = rd_true;
        test_curr->is_fatal_cb   = NULL;

        TEST_CALL_ERROR__(rd_kafka_init_transactions(rk, -1));

        /*
         * Start a transaction
         */
        TEST_CALL_ERROR__(rd_kafka_begin_transaction(rk));



        /* Produce one seeding message first to get the leader up and running */
        TEST_CALL_ERR__(rd_kafka_producev(
            rk, RD_KAFKA_V_TOPIC("mytopic"), RD_KAFKA_V_PARTITION(0),
            RD_KAFKA_V_VALUE("hi", 2), RD_KAFKA_V_END));
        test_flush(rk, -1);

        /* Let partition leader have a latency of 2 seconds
         * so that we can have multiple messages in-flight. */
        rd_kafka_mock_broker_set_rtt(mcluster, leader, 2 * 1000);

        /* Produce a message, let it fail with with different errors,
         * ending with OUT_OF_ORDER which previously triggered an
         * Epoch bump. */
        rd_kafka_mock_push_request_errors(
            mcluster, RD_KAFKAP_Produce, 3,
            RD_KAFKA_RESP_ERR_NOT_LEADER_FOR_PARTITION,
            RD_KAFKA_RESP_ERR_NOT_LEADER_FOR_PARTITION,
            RD_KAFKA_RESP_ERR_OUT_OF_ORDER_SEQUENCE_NUMBER);

        /* Produce three messages that will be delayed
         * and have errors injected.*/
        TEST_CALL_ERR__(rd_kafka_producev(
            rk, RD_KAFKA_V_TOPIC("mytopic"), RD_KAFKA_V_PARTITION(0),
            RD_KAFKA_V_VALUE("hi", 2), RD_KAFKA_V_END));
        TEST_CALL_ERR__(rd_kafka_producev(
            rk, RD_KAFKA_V_TOPIC("mytopic"), RD_KAFKA_V_PARTITION(0),
            RD_KAFKA_V_VALUE("hi", 2), RD_KAFKA_V_END));
        TEST_CALL_ERR__(rd_kafka_producev(
            rk, RD_KAFKA_V_TOPIC("mytopic"), RD_KAFKA_V_PARTITION(0),
            RD_KAFKA_V_VALUE("hi", 2), RD_KAFKA_V_END));

        /* Now sleep a short while so that the messages are processed
         * by the broker and errors are returned. */
        TEST_SAY("Sleeping..\n");
        rd_sleep(5);

        rd_kafka_mock_broker_set_rtt(mcluster, leader, 0);

        /* Produce a fifth message, should fail with ERR__STATE since
         * the transaction should have entered the abortable state. */
        err = rd_kafka_producev(rk, RD_KAFKA_V_TOPIC("mytopic"),
                                RD_KAFKA_V_PARTITION(0),
                                RD_KAFKA_V_VALUE("hi", 2), RD_KAFKA_V_END);
        TEST_ASSERT(err == RD_KAFKA_RESP_ERR__STATE,
                    "Expected produce() to fail with ERR__STATE, not %s",
                    rd_kafka_err2name(err));
        TEST_SAY("produce() failed as expected: %s\n", rd_kafka_err2str(err));

        /* Commit the transaction, should fail with abortable error. */
        TIMING_START(&timing, "commit_transaction(-1)");
        error = rd_kafka_commit_transaction(rk, -1);
        TIMING_STOP(&timing);
        TEST_ASSERT(error != NULL, "Expected commit_transaction() to fail");

        TEST_SAY("commit_transaction() failed (expectedly): %s\n",
                 rd_kafka_error_string(error));

        TEST_ASSERT(!rd_kafka_error_is_fatal(error),
                    "Did not expect fatal error");
        TEST_ASSERT(rd_kafka_error_txn_requires_abort(error),
                    "Expected abortable error");
        rd_kafka_error_destroy(error);

        /* Abort the transaction */
        TEST_CALL_ERROR__(rd_kafka_abort_transaction(rk, -1));

        /* Run a new transaction without errors to verify that the
         * producer can recover. */
        TEST_CALL_ERROR__(rd_kafka_begin_transaction(rk));

        TEST_CALL_ERR__(rd_kafka_producev(
            rk, RD_KAFKA_V_TOPIC("mytopic"), RD_KAFKA_V_PARTITION(0),
            RD_KAFKA_V_VALUE("hi", 2), RD_KAFKA_V_END));

        TEST_CALL_ERROR__(rd_kafka_commit_transaction(rk, -1));

        rd_kafka_destroy(rk);

        SUB_TEST_PASS();
}


/**
 * @brief Verify lossless delivery if topic disappears from Metadata for awhile.
 *
 * If a topic is removed from metadata inbetween transactions, the producer
 * will remove its partition state for the topic's partitions.
 * If later the same topic comes back (same topic instance, not a new creation)
 * then the producer must restore the previously used msgid/BaseSequence
 * in case the same Epoch is still used, or messages will be silently lost
 * as they would seem like legit duplicates to the broker.
 *
 * Reproduction:
 *   1. produce msgs to topic, commit transaction.
 *   2. remove topic from metadata
 *   3. make sure client updates its metadata, which removes the partition
 *      objects.
 *   4. restore the topic in metadata
 *   5. produce new msgs to topic, commit transaction.
 *   6. consume topic. All messages should be accounted for.
 */
static void do_test_topic_disappears_for_awhile(void) {
        rd_kafka_t *rk, *c;
        rd_kafka_conf_t *c_conf;
        rd_kafka_mock_cluster_t *mcluster;
        const char *topic = "mytopic";
        const char *txnid = "myTxnId";
        test_timing_t timing;
        int i;
        int msgcnt              = 0;
        const int partition_cnt = 10;

        SUB_TEST_QUICK();

        rk = create_txn_producer(
            &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);

        TEST_CALL_ERROR__(rd_kafka_init_transactions(rk, -1));

        for (i = 0; i < 2; i++) {
                int cnt                = 3 * 2 * partition_cnt;
                rd_bool_t remove_topic = (i % 2) == 0;
                /*
                 * Start a transaction
                 */
                TEST_CALL_ERROR__(rd_kafka_begin_transaction(rk));


                while (cnt-- >= 0) {
                        TEST_CALL_ERR__(rd_kafka_producev(
                            rk, RD_KAFKA_V_TOPIC(topic),
                            RD_KAFKA_V_PARTITION(cnt % partition_cnt),
                            RD_KAFKA_V_VALUE("hi", 2), RD_KAFKA_V_END));
                        msgcnt++;
                }

                /* Commit the transaction */
                TIMING_START(&timing, "commit_transaction(-1)");
                TEST_CALL_ERROR__(rd_kafka_commit_transaction(rk, -1));
                TIMING_STOP(&timing);



                if (remove_topic) {
                        /* Make it seem the topic is removed, refresh metadata,
                         * and then make the topic available again. */
                        const rd_kafka_metadata_t *md;

                        TEST_SAY("Marking topic as non-existent\n");

                        rd_kafka_mock_topic_set_error(
                            mcluster, topic,
                            RD_KAFKA_RESP_ERR_UNKNOWN_TOPIC_OR_PART);

                        TEST_CALL_ERR__(rd_kafka_metadata(rk, 0, NULL, &md,
                                                          tmout_multip(5000)));

                        rd_kafka_metadata_destroy(md);

                        rd_sleep(2);

                        TEST_SAY("Bringing topic back to life\n");
                        rd_kafka_mock_topic_set_error(
                            mcluster, topic, RD_KAFKA_RESP_ERR_NO_ERROR);
                }
        }

        TEST_SAY("Verifying messages by consumtion\n");
        test_conf_init(&c_conf, NULL, 0);
        test_conf_set(c_conf, "security.protocol", "PLAINTEXT");
        test_conf_set(c_conf, "bootstrap.servers",
                      rd_kafka_mock_cluster_bootstraps(mcluster));
        test_conf_set(c_conf, "enable.partition.eof", "true");
        test_conf_set(c_conf, "auto.offset.reset", "earliest");
        c = test_create_consumer("mygroup", NULL, c_conf, NULL);

        test_consumer_subscribe(c, topic);
        test_consumer_poll_exact("consume", c, 0, partition_cnt, 0, msgcnt,
                                 rd_true /*exact*/, NULL);
        rd_kafka_destroy(c);


        rd_kafka_destroy(rk);

        SUB_TEST_PASS();
}


/**
 * @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.
 *
 * There are two variants to this test based on switch_coord:
 *  - True - Switches the coordinator during the downtime.
 *           The client should detect this and send the request to the
 *           new coordinator.
 *  - False - The coordinator remains on the down broker. Client will reconnect
 *            when down broker comes up again.
 */
struct some_state {
        rd_kafka_mock_cluster_t *mcluster;
        rd_bool_t switch_coord;
        int32_t broker_id;
        const char *grpid;
};

static int delayed_up_cb(void *arg) {
        struct some_state *state = arg;
        rd_sleep(3);
        if (state->switch_coord) {
                TEST_SAY("Switching group coordinator to %" PRId32 "\n",
                         state->broker_id);
                rd_kafka_mock_coordinator_set(state->mcluster, "group",
                                              state->grpid, state->broker_id);
        } else {
                TEST_SAY("Bringing up group coordinator %" PRId32 "..\n",
                         state->broker_id);
                rd_kafka_mock_broker_set_up(state->mcluster, state->broker_id);
        }
        return 0;
}

static void do_test_disconnected_group_coord(rd_bool_t switch_coord) {
        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;
        struct some_state state = RD_ZERO_INIT;
        test_timing_t timing;
        thrd_t thrd;
        int ret;

        SUB_TEST_QUICK("switch_coord=%s", RD_STR_ToF(switch_coord));

        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 & backup coord if switch_coord=true */
        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. */
        state.switch_coord = switch_coord;
        state.mcluster     = mcluster;
        state.grpid        = grpid;
        state.broker_id    = switch_coord ? 3 : 2;
        if (thrd_create(&thrd, delayed_up_cb, &state) != thrd_success)
                TEST_FAIL("Failed to create thread");

        TEST_SAY("Calling send_offsets_to_transaction()\n");
        offsets = rd_kafka_topic_partition_list_new(1);
        rd_kafka_topic_partition_list_add(offsets, "srctopic4", 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();
}


/**
 * @brief Test that a NULL coordinator is not fatal when
 * the transactional producer reconnects to the txn coordinator
 * and the first thing it does is a FindCoordinatorRequest that
 * fails with COORDINATOR_NOT_AVAILABLE, setting coordinator to NULL.
 */
static void do_test_txn_coordinator_null_not_fatal(void) {
        rd_kafka_t *rk;
        rd_kafka_mock_cluster_t *mcluster;
        rd_kafka_error_t *error;
        rd_kafka_resp_err_t err;
        int32_t coord_id             = 1;
        const char *topic            = "test";
        const char *transactional_id = "txnid";
        int msgcnt                   = 1;
        int remains                  = 0;

        SUB_TEST_QUICK();

        /* Broker down is not a test-failing error */
        allowed_error          = RD_KAFKA_RESP_ERR__TRANSPORT;
        test_curr->is_fatal_cb = error_is_fatal_cb;
        test_curr->exp_dr_err  = RD_KAFKA_RESP_ERR__MSG_TIMED_OUT;

        /* One second is the minimum transaction timeout */
        rk = create_txn_producer(&mcluster, transactional_id, 1,
                                 "transaction.timeout.ms", "1000", NULL);

        err = rd_kafka_mock_topic_create(mcluster, topic, 1, 1);
        TEST_ASSERT(!err, "Failed to create topic: %s", rd_kafka_err2str(err));

        rd_kafka_mock_coordinator_set(mcluster, "transaction", transactional_id,
                                      coord_id);
        rd_kafka_mock_partition_set_leader(mcluster, topic, 0, coord_id);

        /* Start transactioning */
        TEST_SAY("Starting transaction\n");
        TEST_CALL_ERROR__(rd_kafka_init_transactions(rk, 5000));
        TEST_CALL_ERROR__(rd_kafka_begin_transaction(rk));

        /* Makes the produce request timeout. */
        rd_kafka_mock_broker_push_request_error_rtts(
            mcluster, coord_id, RD_KAFKAP_Produce, 1,
            RD_KAFKA_RESP_ERR_NO_ERROR, 3000);

        test_produce_msgs2_nowait(rk, topic, 0, RD_KAFKA_PARTITION_UA, 0,
                                  msgcnt, NULL, 0, &remains);

        /* This value is linked to transaction.timeout.ms, needs enough time
         * so the message times out and a DrainBump sequence is started. */
        rd_kafka_flush(rk, 1000);

        /* To trigger the error the COORDINATOR_NOT_AVAILABLE response
         * must come AFTER idempotent state has changed to WaitTransport
         * but BEFORE it changes to WaitPID. To make it more likely
         * rd_kafka_txn_coord_timer_start timeout can be changed to 5 ms
         * in rd_kafka_txn_coord_query, when unable to query for
         * transaction coordinator.
         */
        rd_kafka_mock_broker_push_request_error_rtts(
            mcluster, coord_id, RD_KAFKAP_FindCoordinator, 1,
            RD_KAFKA_RESP_ERR_COORDINATOR_NOT_AVAILABLE, 10);

        /* Coordinator down starts the FindCoordinatorRequest loop. */
        TEST_SAY("Bringing down coordinator %" PRId32 "\n", coord_id);
        rd_kafka_mock_broker_set_down(mcluster, coord_id);

        /* Coordinator down for some time. */
        rd_usleep(100 * 1000, NULL);

        /* When it comes up, the error is triggered, if the preconditions
         * happen. */
        TEST_SAY("Bringing up coordinator %" PRId32 "\n", coord_id);
        rd_kafka_mock_broker_set_up(mcluster, coord_id);

        /* Make sure DRs are received */
        rd_kafka_flush(rk, 1000);

        error = rd_kafka_commit_transaction(rk, -1);

        TEST_ASSERT(remains == 0, "%d message(s) were not produced\n", remains);
        TEST_ASSERT(error != NULL, "Expected commit_transaction() to fail");
        TEST_SAY("commit_transaction() failed (expectedly): %s\n",
                 rd_kafka_error_string(error));
        rd_kafka_error_destroy(error);

        /* Needs to wait some time before closing to make sure it doesn't go
         * into TERMINATING state before error is triggered. */
        rd_usleep(1000 * 1000, NULL);
        rd_kafka_destroy(rk);

        allowed_error          = RD_KAFKA_RESP_ERR_NO_ERROR;
        test_curr->exp_dr_err  = RD_KAFKA_RESP_ERR_NO_ERROR;
        test_curr->is_fatal_cb = NULL;

        SUB_TEST_PASS();
}



/**
 * @brief Simple test to make sure the init_transactions() timeout is honoured
 *        and also not infinite.
 */
static void do_test_txn_resumable_init(void) {
        rd_kafka_t *rk;
        const char *transactional_id = "txnid";
        rd_kafka_error_t *error;
        test_timing_t duration;

        SUB_TEST();

        rd_kafka_conf_t *conf;

        test_conf_init(&conf, NULL, 20);
        test_conf_set(conf, "bootstrap.servers", "");
        test_conf_set(conf, "transactional.id", transactional_id);
        test_conf_set(conf, "transaction.timeout.ms", "4000");

        rk = test_create_handle(RD_KAFKA_PRODUCER, conf);

        /* First make sure a lower timeout is honoured. */
        TIMING_START(&duration, "init_transactions(1000)");
        error = rd_kafka_init_transactions(rk, 1000);
        TIMING_STOP(&duration);

        if (error)
                TEST_SAY("First init_transactions failed (as expected): %s\n",
                         rd_kafka_error_string(error));
        TEST_ASSERT(rd_kafka_error_code(error) == RD_KAFKA_RESP_ERR__TIMED_OUT,
                    "Expected _TIMED_OUT, not %s",
                    error ? rd_kafka_error_string(error) : "success");
        rd_kafka_error_destroy(error);

        TIMING_ASSERT(&duration, 900, 1500);

        TEST_SAY(
            "Performing second init_transactions() call now with an "
            "infinite timeout: "
            "should time out in 2 x transaction.timeout.ms\n");

        TIMING_START(&duration, "init_transactions(infinite)");
        error = rd_kafka_init_transactions(rk, -1);
        TIMING_STOP(&duration);

        if (error)
                TEST_SAY("Second init_transactions failed (as expected): %s\n",
                         rd_kafka_error_string(error));
        TEST_ASSERT(rd_kafka_error_code(error) == RD_KAFKA_RESP_ERR__TIMED_OUT,
                    "Expected _TIMED_OUT, not %s",
                    error ? rd_kafka_error_string(error) : "success");
        rd_kafka_error_destroy(error);

        TIMING_ASSERT(&duration, 2 * 4000 - 500, 2 * 4000 + 500);

        rd_kafka_destroy(rk);

        SUB_TEST_PASS();
}


/**
 * @brief Retries a transaction call until it succeeds or returns a
 *        non-retriable error - which will cause the test to fail.
 *
 * @param intermed_calls Is a block of code that will be called after each
 *                       retriable failure of \p call.
 */
#define RETRY_TXN_CALL__(call, intermed_calls)                                 \
        do {                                                                   \
                rd_kafka_error_t *_error = call;                               \
                if (!_error)                                                   \
                        break;                                                 \
                TEST_SAY_ERROR(_error, "%s: ", "" #call);                      \
                TEST_ASSERT(rd_kafka_error_is_retriable(_error),               \
                            "Expected retriable error");                       \
                TEST_SAY("%s failed, retrying in 1 second\n", "" #call);       \
                rd_kafka_error_destroy(_error);                                \
                intermed_calls;                                                \
                rd_sleep(1);                                                   \
        } while (1)

/**
 * @brief Call \p call and expect it to fail with \p exp_err_code.
 */
#define TXN_CALL_EXPECT_ERROR__(call, exp_err_code)                            \
        do {                                                                   \
                rd_kafka_error_t *_error = call;                               \
                TEST_ASSERT(_error != NULL,                                    \
                            "%s: Expected %s error, got success", "" #call,    \
                            rd_kafka_err2name(exp_err_code));                  \
                TEST_SAY_ERROR(_error, "%s: ", "" #call);                      \
                TEST_ASSERT(rd_kafka_error_code(_error) == exp_err_code,       \
                            "%s: Expected %s error, got %s", "" #call,         \
                            rd_kafka_err2name(exp_err_code),                   \
                            rd_kafka_error_name(_error));                      \
                rd_kafka_error_destroy(_error);                                \
        } while (0)


/**
 * @brief Simple test to make sure short API timeouts can be safely resumed
 *        by calling the same API again.
 *
 * @param do_commit Commit transaction if true, else abort transaction.
 */
static void do_test_txn_resumable_calls_timeout(rd_bool_t do_commit) {
        rd_kafka_t *rk;
        rd_kafka_mock_cluster_t *mcluster;
        rd_kafka_resp_err_t err;
        rd_kafka_topic_partition_list_t *offsets;
        rd_kafka_consumer_group_metadata_t *cgmetadata;
        int32_t coord_id             = 1;
        const char *topic            = "test";
        const char *transactional_id = "txnid";
        int msgcnt                   = 1;
        int remains                  = 0;

        SUB_TEST("%s_transaction", do_commit ? "commit" : "abort");

        rk = create_txn_producer(&mcluster, transactional_id, 1, NULL);

        err = rd_kafka_mock_topic_create(mcluster, topic, 1, 1);
        TEST_ASSERT(!err, "Failed to create topic: %s", rd_kafka_err2str(err));

        rd_kafka_mock_coordinator_set(mcluster, "transaction", transactional_id,
                                      coord_id);
        rd_kafka_mock_partition_set_leader(mcluster, topic, 0, coord_id);

        TEST_SAY("Starting transaction\n");
        TEST_SAY("Delaying first two InitProducerIdRequests by 500ms\n");
        rd_kafka_mock_broker_push_request_error_rtts(
            mcluster, coord_id, RD_KAFKAP_InitProducerId, 2,
            RD_KAFKA_RESP_ERR_NO_ERROR, 500, RD_KAFKA_RESP_ERR_NO_ERROR, 500);

        RETRY_TXN_CALL__(
            rd_kafka_init_transactions(rk, 100),
            TXN_CALL_EXPECT_ERROR__(rd_kafka_abort_transaction(rk, -1),
                                    RD_KAFKA_RESP_ERR__CONFLICT));

        RETRY_TXN_CALL__(rd_kafka_begin_transaction(rk), /*none*/);


        TEST_SAY("Delaying ProduceRequests by 3000ms\n");
        rd_kafka_mock_broker_push_request_error_rtts(
            mcluster, coord_id, RD_KAFKAP_Produce, 1,
            RD_KAFKA_RESP_ERR_NO_ERROR, 3000);

        test_produce_msgs2_nowait(rk, topic, 0, RD_KAFKA_PARTITION_UA, 0,
                                  msgcnt, NULL, 0, &remains);


        TEST_SAY("Delaying SendOffsetsToTransaction by 400ms\n");
        rd_kafka_mock_broker_push_request_error_rtts(
            mcluster, coord_id, RD_KAFKAP_AddOffsetsToTxn, 1,
            RD_KAFKA_RESP_ERR_NO_ERROR, 400);
        offsets = rd_kafka_topic_partition_list_new(1);
        rd_kafka_topic_partition_list_add(offsets, "srctopic4", 0)->offset = 12;
        cgmetadata = rd_kafka_consumer_group_metadata_new("mygroupid");

        /* This is not a resumable call on timeout */
        TEST_CALL_ERROR__(
            rd_kafka_send_offsets_to_transaction(rk, offsets, cgmetadata, -1));

        rd_kafka_consumer_group_metadata_destroy(cgmetadata);
        rd_kafka_topic_partition_list_destroy(offsets);


        TEST_SAY("Delaying EndTxnRequests by 1200ms\n");
        rd_kafka_mock_broker_push_request_error_rtts(
            mcluster, coord_id, RD_KAFKAP_EndTxn, 1, RD_KAFKA_RESP_ERR_NO_ERROR,
            1200);

        /* Committing/aborting the transaction will also be delayed by the
         * previous accumulated remaining delays. */

        if (do_commit) {
                TEST_SAY("Committing transaction\n");

                RETRY_TXN_CALL__(
                    rd_kafka_commit_transaction(rk, 100),
                    TXN_CALL_EXPECT_ERROR__(rd_kafka_abort_transaction(rk, -1),
                                            RD_KAFKA_RESP_ERR__CONFLICT));
        } else {
                TEST_SAY("Aborting transaction\n");

                RETRY_TXN_CALL__(
                    rd_kafka_abort_transaction(rk, 100),
                    TXN_CALL_EXPECT_ERROR__(rd_kafka_commit_transaction(rk, -1),
                                            RD_KAFKA_RESP_ERR__CONFLICT));
        }

        rd_kafka_destroy(rk);

        SUB_TEST_PASS();
}


/**
 * @brief Verify that resuming timed out calls that after the timeout, but
 *        before the resuming call, would error out.
 */
static void do_test_txn_resumable_calls_timeout_error(rd_bool_t do_commit) {
        rd_kafka_t *rk;
        rd_kafka_mock_cluster_t *mcluster;
        rd_kafka_resp_err_t err;
        int32_t coord_id             = 1;
        const char *topic            = "test";
        const char *transactional_id = "txnid";
        int msgcnt                   = 1;
        int remains                  = 0;
        rd_kafka_error_t *error;

        SUB_TEST_QUICK("%s_transaction", do_commit ? "commit" : "abort");

        rk = create_txn_producer(&mcluster, transactional_id, 1, NULL);

        err = rd_kafka_mock_topic_create(mcluster, topic, 1, 1);
        TEST_ASSERT(!err, "Failed to create topic: %s", rd_kafka_err2str(err));

        rd_kafka_mock_coordinator_set(mcluster, "transaction", transactional_id,
                                      coord_id);
        rd_kafka_mock_partition_set_leader(mcluster, topic, 0, coord_id);

        TEST_SAY("Starting transaction\n");

        TEST_CALL_ERROR__(rd_kafka_init_transactions(rk, -1));

        TEST_CALL_ERROR__(rd_kafka_begin_transaction(rk));

        test_produce_msgs2_nowait(rk, topic, 0, RD_KAFKA_PARTITION_UA, 0,
                                  msgcnt, NULL, 0, &remains);


        TEST_SAY("Fail EndTxn fatally after 2000ms\n");
        rd_kafka_mock_broker_push_request_error_rtts(
            mcluster, coord_id, RD_KAFKAP_EndTxn, 1,
            RD_KAFKA_RESP_ERR_INVALID_TXN_STATE, 2000);

        if (do_commit) {
                TEST_SAY("Committing transaction\n");

                TXN_CALL_EXPECT_ERROR__(rd_kafka_commit_transaction(rk, 500),
                                        RD_KAFKA_RESP_ERR__TIMED_OUT);

                /* Sleep so that the background EndTxn fails locally and sets
                 * an error result. */
                rd_sleep(3);

                error = rd_kafka_commit_transaction(rk, -1);

        } else {
                TEST_SAY("Aborting transaction\n");

                TXN_CALL_EXPECT_ERROR__(rd_kafka_commit_transaction(rk, 500),
                                        RD_KAFKA_RESP_ERR__TIMED_OUT);

                /* Sleep so that the background EndTxn fails locally and sets
                 * an error result. */
                rd_sleep(3);

                error = rd_kafka_commit_transaction(rk, -1);
        }

        TEST_ASSERT(error != NULL && rd_kafka_error_is_fatal(error),
                    "Expected fatal error, not %s",
                    rd_kafka_error_string(error));
        TEST_ASSERT(rd_kafka_error_code(error) ==
                        RD_KAFKA_RESP_ERR_INVALID_TXN_STATE,
                    "Expected error INVALID_TXN_STATE, got %s",
                    rd_kafka_error_name(error));
        rd_kafka_error_destroy(error);

        rd_kafka_destroy(rk);

        SUB_TEST_PASS();
}


/**
 * @brief Concurrent transaction API calls are not permitted.
 *        This test makes sure they're properly enforced.
 *
 * For each transactional API, call it with a 5s timeout, and during that time
 * from another thread call transactional APIs, one by one, and verify that
 * we get an ERR__CONFLICT error back in the second thread.
 *
 * We use a mutex for synchronization, the main thread will hold the lock
 * when not calling an API but release it just prior to calling.
 * The other thread will acquire the lock, sleep, and hold the lock while
 * calling the concurrent API that should fail immediately, releasing the lock
 * when done.
 *
 */

struct _txn_concurrent_state {
        const char *api;
        mtx_t lock;
        rd_kafka_t *rk;
        struct test *test;
};

static int txn_concurrent_thread_main(void *arg) {
        struct _txn_concurrent_state *state = arg;
        static const char *apis[]           = {
            "init_transactions",           "begin_transaction",
            "send_offsets_to_transaction", "commit_transaction",
            "abort_transaction",           NULL};
        rd_kafka_t *rk       = state->rk;
        const char *main_api = NULL;
        int i;

        /* Update TLS variable so TEST_..() macros work */
        test_curr = state->test;

        while (1) {
                const char *api         = NULL;
                const int timeout_ms    = 10000;
                rd_kafka_error_t *error = NULL;
                rd_kafka_resp_err_t exp_err;
                test_timing_t duration;

                /* Wait for other thread's txn call to start, then sleep a bit
                 * to increase the chance of that call has really begun. */
                mtx_lock(&state->lock);

                if (state->api && state->api == main_api) {
                        /* Main thread is still blocking on the last API call */
                        TEST_SAY("Waiting for main thread to finish %s()\n",
                                 main_api);
                        mtx_unlock(&state->lock);
                        rd_sleep(1);
                        continue;
                } else if (!(main_api = state->api)) {
                        mtx_unlock(&state->lock);
                        break;
                }

                rd_sleep(1);

                for (i = 0; (api = apis[i]) != NULL; i++) {
                        TEST_SAY(
                            "Triggering concurrent %s() call while "
                            "main is in %s() call\n",
                            api, main_api);
                        TIMING_START(&duration, "%s", api);

                        if (!strcmp(api, "init_transactions"))
                                error =
                                    rd_kafka_init_transactions(rk, timeout_ms);
                        else if (!strcmp(api, "begin_transaction"))
                                error = rd_kafka_begin_transaction(rk);
                        else if (!strcmp(api, "send_offsets_to_transaction")) {
                                rd_kafka_topic_partition_list_t *offsets =
                                    rd_kafka_topic_partition_list_new(1);
                                rd_kafka_consumer_group_metadata_t *cgmetadata =
                                    rd_kafka_consumer_group_metadata_new(
                                        "mygroupid");
                                rd_kafka_topic_partition_list_add(
                                    offsets, "srctopic4", 0)
                                    ->offset = 12;

                                error = rd_kafka_send_offsets_to_transaction(
                                    rk, offsets, cgmetadata, -1);
                                rd_kafka_consumer_group_metadata_destroy(
                                    cgmetadata);
                                rd_kafka_topic_partition_list_destroy(offsets);
                        } else if (!strcmp(api, "commit_transaction"))
                                error =
                                    rd_kafka_commit_transaction(rk, timeout_ms);
                        else if (!strcmp(api, "abort_transaction"))
                                error =
                                    rd_kafka_abort_transaction(rk, timeout_ms);
                        else
                                TEST_FAIL("Unknown API: %s", api);

                        TIMING_STOP(&duration);

                        TEST_SAY_ERROR(error, "Conflicting %s() call: ", api);
                        TEST_ASSERT(error,
                                    "Expected conflicting %s() call to fail",
                                    api);

                        exp_err = !strcmp(api, main_api)
                                      ? RD_KAFKA_RESP_ERR__PREV_IN_PROGRESS
                                      : RD_KAFKA_RESP_ERR__CONFLICT;

                        TEST_ASSERT(rd_kafka_error_code(error) == exp_err,

                                    "Conflicting %s(): Expected %s, not %s",
                                    api, rd_kafka_err2str(exp_err),
                                    rd_kafka_error_name(error));
                        TEST_ASSERT(
                            rd_kafka_error_is_retriable(error),
                            "Conflicting %s(): Expected retriable error", api);
                        rd_kafka_error_destroy(error);
                        /* These calls should fail immediately */
                        TIMING_ASSERT(&duration, 0, 100);
                }

                mtx_unlock(&state->lock);
        }

        return 0;
}

static void do_test_txn_concurrent_operations(rd_bool_t do_commit) {
        rd_kafka_t *rk;
        rd_kafka_mock_cluster_t *mcluster;
        int32_t coord_id = 1;
        rd_kafka_resp_err_t err;
        const char *topic            = "test";
        const char *transactional_id = "txnid";
        int remains                  = 0;
        thrd_t thrd;
        struct _txn_concurrent_state state = RD_ZERO_INIT;
        rd_kafka_topic_partition_list_t *offsets;
        rd_kafka_consumer_group_metadata_t *cgmetadata;

        SUB_TEST("%s", do_commit ? "commit" : "abort");

        test_timeout_set(90);

        /* We need to override the value of socket.connection.setup.timeout.ms
         * to be at least 2*RTT of the mock broker. This is because the first
         * ApiVersion request will fail, since we make the request with v3, and
         * the mock broker's MaxVersion is 2, so the request is retried with v0.
         * We use the value 3*RTT to add some buffer.
         */
        rk = create_txn_producer(&mcluster, transactional_id, 1,
                                 "socket.connection.setup.timeout.ms", "15000",
                                 NULL);

        /* Set broker RTT to 3.5s so that the background thread has ample
         * time to call its conflicting APIs.
         * This value must be less than socket.connection.setup.timeout.ms/2. */
        rd_kafka_mock_broker_set_rtt(mcluster, coord_id, 3500);

        err = rd_kafka_mock_topic_create(mcluster, topic, 1, 1);
        TEST_ASSERT(!err, "Failed to create topic: %s", rd_kafka_err2str(err));

        /* Set up shared state between us and the concurrent thread */
        mtx_init(&state.lock, mtx_plain);
        state.test = test_curr;
        state.rk   = rk;

        /* We release the lock only while calling the TXN API */
        mtx_lock(&state.lock);

        /* Spin up concurrent thread */
        if (thrd_create(&thrd, txn_concurrent_thread_main, (void *)&state) !=
            thrd_success)
                TEST_FAIL("Failed to create thread");

#define _start_call(callname)                                                  \
        do {                                                                   \
                state.api = callname;                                          \
                mtx_unlock(&state.lock);                                       \
        } while (0)
#define _end_call() mtx_lock(&state.lock)

        _start_call("init_transactions");
        TEST_CALL_ERROR__(rd_kafka_init_transactions(rk, -1));
        _end_call();

        /* This call doesn't block, so can't really be tested concurrently. */
        TEST_CALL_ERROR__(rd_kafka_begin_transaction(rk));

        test_produce_msgs2_nowait(rk, topic, 0, RD_KAFKA_PARTITION_UA, 0, 10,
                                  NULL, 0, &remains);

        _start_call("send_offsets_to_transaction");
        offsets = rd_kafka_topic_partition_list_new(1);
        rd_kafka_topic_partition_list_add(offsets, "srctopic4", 0)->offset = 12;
        cgmetadata = rd_kafka_consumer_group_metadata_new("mygroupid");

        TEST_CALL_ERROR__(
            rd_kafka_send_offsets_to_transaction(rk, offsets, cgmetadata, -1));
        rd_kafka_consumer_group_metadata_destroy(cgmetadata);
        rd_kafka_topic_partition_list_destroy(offsets);
        _end_call();

        if (do_commit) {
                _start_call("commit_transaction");
                TEST_CALL_ERROR__(rd_kafka_commit_transaction(rk, -1));
                _end_call();
        } else {
                _start_call("abort_transaction");
                TEST_CALL_ERROR__(rd_kafka_abort_transaction(rk, -1));
                _end_call();
        }

        /* Signal completion to background thread */
        state.api = NULL;

        mtx_unlock(&state.lock);

        thrd_join(thrd, NULL);

        rd_kafka_destroy(rk);

        mtx_destroy(&state.lock);

        SUB_TEST_PASS();
}


/**
 * @brief KIP-360: Test that fatal idempotence errors triggers abortable
 *        transaction errors, but let the broker-side abort of the
 *        transaction fail with a fencing error.
 *        Should raise a fatal error.
 *
 * @param error_code Which error code EndTxn should fail with.
 *                   Either RD_KAFKA_RESP_ERR_INVALID_PRODUCER_EPOCH (older)
 *                   or RD_KAFKA_RESP_ERR_PRODUCER_FENCED (newer).
 */
static void do_test_txn_fenced_abort(rd_kafka_resp_err_t error_code) {
        rd_kafka_t *rk;
        rd_kafka_mock_cluster_t *mcluster;
        rd_kafka_error_t *error;
        int32_t txn_coord = 2;
        const char *txnid = "myTxnId";
        char errstr[512];
        rd_kafka_resp_err_t fatal_err;
        size_t errors_cnt;

        SUB_TEST_QUICK("With error %s", rd_kafka_err2name(error_code));

        rk = create_txn_producer(&mcluster, txnid, 3, "batch.num.messages", "1",
                                 NULL);

        rd_kafka_mock_coordinator_set(mcluster, "transaction", txnid,
                                      txn_coord);

        test_curr->ignore_dr_err = rd_true;
        test_curr->is_fatal_cb   = error_is_fatal_cb;
        allowed_error            = RD_KAFKA_RESP_ERR__FENCED;

        TEST_CALL_ERROR__(rd_kafka_init_transactions(rk, -1));

        /*
         * Start a transaction
         */
        TEST_CALL_ERROR__(rd_kafka_begin_transaction(rk));


        /* Produce a message without error first */
        TEST_CALL_ERR__(rd_kafka_producev(
            rk, RD_KAFKA_V_TOPIC("mytopic"), RD_KAFKA_V_PARTITION(0),
            RD_KAFKA_V_VALUE("hi", 2), RD_KAFKA_V_END));

        test_flush(rk, -1);

        /* Fail abort transaction  */
        rd_kafka_mock_broker_push_request_error_rtts(
            mcluster, txn_coord, RD_KAFKAP_EndTxn, 1, error_code, 0);

        /* Fail the PID reinit */
        rd_kafka_mock_broker_push_request_error_rtts(
            mcluster, txn_coord, RD_KAFKAP_InitProducerId, 1, error_code, 0);

        /* Produce a message, let it fail with a fatal idempo error. */
        rd_kafka_mock_push_request_errors(
            mcluster, RD_KAFKAP_Produce, 1,
            RD_KAFKA_RESP_ERR_UNKNOWN_PRODUCER_ID);

        TEST_CALL_ERR__(rd_kafka_producev(
            rk, RD_KAFKA_V_TOPIC("mytopic"), RD_KAFKA_V_PARTITION(0),
            RD_KAFKA_V_VALUE("hi", 2), RD_KAFKA_V_END));

        test_flush(rk, -1);

        /* Abort the transaction, should fail with a fatal error */
        error = rd_kafka_abort_transaction(rk, -1);
        TEST_ASSERT(error != NULL, "Expected abort_transaction() to fail");

        TEST_SAY_ERROR(error, "abort_transaction() failed: ");
        TEST_ASSERT(rd_kafka_error_is_fatal(error), "Expected a fatal error");
        rd_kafka_error_destroy(error);

        fatal_err = rd_kafka_fatal_error(rk, errstr, sizeof(errstr));
        TEST_ASSERT(fatal_err, "Expected a fatal error to have been raised");
        TEST_SAY("Fatal error: %s: %s\n", rd_kafka_err2name(fatal_err), errstr);

        /* Verify that the producer sent the expected number of EndTxn requests
         * by inspecting the mock broker error stack,
         * which should now be empty. */
        if (rd_kafka_mock_broker_error_stack_cnt(
                mcluster, txn_coord, RD_KAFKAP_EndTxn, &errors_cnt)) {
                TEST_FAIL(
                    "Broker error count should succeed for API %s"
                    " on broker %" PRId32,
                    rd_kafka_ApiKey2str(RD_KAFKAP_EndTxn), txn_coord);
        }
        /* Checks all the  RD_KAFKAP_EndTxn responses have been consumed */
        TEST_ASSERT(errors_cnt == 0,
                    "Expected error count 0 for API %s, found %zu",
                    rd_kafka_ApiKey2str(RD_KAFKAP_EndTxn), errors_cnt);

        if (rd_kafka_mock_broker_error_stack_cnt(
                mcluster, txn_coord, RD_KAFKAP_InitProducerId, &errors_cnt)) {
                TEST_FAIL(
                    "Broker error count should succeed for API %s"
                    " on broker %" PRId32,
                    rd_kafka_ApiKey2str(RD_KAFKAP_InitProducerId), txn_coord);
        }
        /* Checks none of the RD_KAFKAP_InitProducerId responses have been
         * consumed
         */
        TEST_ASSERT(errors_cnt == 1,
                    "Expected error count 1 for API %s, found %zu",
                    rd_kafka_ApiKey2str(RD_KAFKAP_InitProducerId), errors_cnt);

        /* All done */
        rd_kafka_destroy(rk);

        allowed_error = RD_KAFKA_RESP_ERR_NO_ERROR;

        SUB_TEST_PASS();
}


/**
 * @brief Test that the TxnOffsetCommit op doesn't retry without waiting
 * if the coordinator is found but not available, causing too frequent retries.
 */
static void
do_test_txn_offset_commit_doesnt_retry_too_quickly(rd_bool_t times_out) {
        rd_kafka_t *rk;
        rd_kafka_mock_cluster_t *mcluster;
        rd_kafka_resp_err_t err;
        rd_kafka_topic_partition_list_t *offsets;
        rd_kafka_consumer_group_metadata_t *cgmetadata;
        rd_kafka_error_t *error;
        int timeout;

        SUB_TEST_QUICK("times_out=%s", RD_STR_ToF(times_out));

        rk = create_txn_producer(&mcluster, "txnid", 3, NULL);

        test_curr->ignore_dr_err = rd_true;

        TEST_CALL_ERROR__(rd_kafka_init_transactions(rk, 5000));

        TEST_CALL_ERROR__(rd_kafka_begin_transaction(rk));

        err = rd_kafka_producev(rk, RD_KAFKA_V_TOPIC("mytopic"),
                                RD_KAFKA_V_VALUE("hi", 2), RD_KAFKA_V_END);
        TEST_ASSERT(!err, "produce failed: %s", rd_kafka_err2str(err));

        /* Wait for messages to be delivered */
        test_flush(rk, 5000);

        /*
         * Fail TxnOffsetCommit with COORDINATOR_NOT_AVAILABLE
         * repeatedly.
         */
        rd_kafka_mock_push_request_errors(
            mcluster, RD_KAFKAP_TxnOffsetCommit, 4,
            RD_KAFKA_RESP_ERR_COORDINATOR_NOT_AVAILABLE,
            RD_KAFKA_RESP_ERR_COORDINATOR_NOT_AVAILABLE,
            RD_KAFKA_RESP_ERR_COORDINATOR_NOT_AVAILABLE,
            RD_KAFKA_RESP_ERR_COORDINATOR_NOT_AVAILABLE);

        offsets = rd_kafka_topic_partition_list_new(1);
        rd_kafka_topic_partition_list_add(offsets, "srctopic4", 3)->offset = 1;

        cgmetadata = rd_kafka_consumer_group_metadata_new("mygroupid");

        /* The retry delay is 500ms, with 4 retries it should take at least
         * 2000ms for this call to succeed. */
        timeout = times_out ? 500 : 4000;
        error   = rd_kafka_send_offsets_to_transaction(rk, offsets, cgmetadata,
                                                     timeout);
        rd_kafka_consumer_group_metadata_destroy(cgmetadata);
        rd_kafka_topic_partition_list_destroy(offsets);

        if (times_out) {
                TEST_ASSERT(rd_kafka_error_code(error) ==
                                RD_KAFKA_RESP_ERR_COORDINATOR_NOT_AVAILABLE,
                            "expected %s, got: %s",
                            rd_kafka_err2name(
                                RD_KAFKA_RESP_ERR_COORDINATOR_NOT_AVAILABLE),
                            rd_kafka_err2str(rd_kafka_error_code(error)));
        } else {
                TEST_ASSERT(rd_kafka_error_code(error) ==
                                RD_KAFKA_RESP_ERR_NO_ERROR,
                            "expected \"Success\", found: %s",
                            rd_kafka_err2str(rd_kafka_error_code(error)));
        }
        rd_kafka_error_destroy(error);

        /* All done */
        rd_kafka_destroy(rk);

        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");
                return 0;
        }

        do_test_txn_recoverable_errors();

        do_test_txn_fatal_idempo_errors();

        do_test_txn_fenced_reinit(RD_KAFKA_RESP_ERR_INVALID_PRODUCER_EPOCH);
        do_test_txn_fenced_reinit(RD_KAFKA_RESP_ERR_PRODUCER_FENCED);

        do_test_txn_req_cnt();

        do_test_txn_requires_abort_errors();

        do_test_txn_slow_reinit(rd_false);
        do_test_txn_slow_reinit(rd_true);

        /* Just do a subset of tests in quick mode */
        if (test_quick)
                return 0;

        do_test_txn_endtxn_errors();

        do_test_txn_endtxn_infinite();

        do_test_txn_endtxn_timeout();

        do_test_txn_endtxn_timeout_inflight();

        /* Bring down the coordinator */
        do_test_txn_broker_down_in_txn(rd_true);

        /* Bring down partition leader */
        do_test_txn_broker_down_in_txn(rd_false);

        do_test_txns_not_supported();

        do_test_txns_send_offsets_concurrent_is_retried();

        do_test_txns_send_offsets_non_eligible();

        do_test_txn_coord_req_destroy();

        do_test_txn_coord_req_multi_find();

        do_test_txn_addparts_req_multi();

        do_test_txns_no_timeout_crash();

        do_test_txn_auth_failure(
            RD_KAFKAP_InitProducerId,
            RD_KAFKA_RESP_ERR_CLUSTER_AUTHORIZATION_FAILED);

        do_test_txn_auth_failure(
            RD_KAFKAP_FindCoordinator,
            RD_KAFKA_RESP_ERR_CLUSTER_AUTHORIZATION_FAILED);

        do_test_txn_flush_timeout();

        do_test_unstable_offset_commit();

        do_test_commit_after_msg_timeout();

        do_test_txn_switch_coordinator();

        do_test_txn_switch_coordinator_refresh();

        do_test_out_of_order_seq();

        do_test_topic_disappears_for_awhile();

        do_test_disconnected_group_coord(rd_false);

        do_test_disconnected_group_coord(rd_true);

        do_test_txn_coordinator_null_not_fatal();

        do_test_txn_resumable_calls_timeout(rd_true);

        do_test_txn_resumable_calls_timeout(rd_false);

        do_test_txn_resumable_calls_timeout_error(rd_true);

        do_test_txn_resumable_calls_timeout_error(rd_false);
        do_test_txn_resumable_init();

        do_test_txn_concurrent_operations(rd_true /*commit*/);

        do_test_txn_concurrent_operations(rd_false /*abort*/);

        do_test_txn_fenced_abort(RD_KAFKA_RESP_ERR_INVALID_PRODUCER_EPOCH);

        do_test_txn_fenced_abort(RD_KAFKA_RESP_ERR_PRODUCER_FENCED);

        do_test_txn_offset_commit_doesnt_retry_too_quickly(rd_true);

        do_test_txn_offset_commit_doesnt_retry_too_quickly(rd_false);

        return 0;
}