/* * librdkafka - Apache Kafka C library * * Copyright (c) 2019-2022, 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 /** * @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; }