diff --git a/CHANGELOG.md b/CHANGELOG.md index e06aa0af3a..4d517c8014 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,3 +1,28 @@ +# librdkafka v2.15.1 + +librdkafka v2.15.1 is a maintenance release: + +- Fix to report the `RD_KAFKA_RESP_ERR__ALL_BROKERS_DOWN` error only once per + outage instead of on every re-bootstrap cycle while all brokers stay + unreachable (#5546). + + +## Fixes + +### General fixes + +- Issues: #5546. + The `RD_KAFKA_RESP_ERR__ALL_BROKERS_DOWN` error was reported on every + re-bootstrap cycle (roughly every reconnect interval) for the whole duration + of an outage in which all brokers are unreachable, instead of only once per + outage, because the re-bootstrap sequence resets the per-broker + down-reported state. The error is now reported once per outage and armed + again as soon as any broker connection comes up. Re-bootstrap sequences + continue on every cycle as before. + Happening since 2.11.1 (#5126). + + + # librdkafka v2.15.0 librdkafka v2.15.0 is a feature release: diff --git a/src/rdkafka.c b/src/rdkafka.c index a8691f0a3a..fc9df8cf52 100644 --- a/src/rdkafka.c +++ b/src/rdkafka.c @@ -2610,6 +2610,7 @@ rd_kafka_t *rd_kafka_new(rd_kafka_type_t type, rd_atomic32_init(&rk->rk_logical_broker_cnt, 0); rd_atomic32_init(&rk->rk_broker_up_cnt, 0); rd_atomic32_init(&rk->rk_broker_down_cnt, 0); + rd_atomic32_init(&rk->rk_all_brokers_down_reported, 0); rd_atomic32_init(&rk->rk_rebootstrap_in_progress, 0); rk->rk_rep = rd_kafka_q_new(rk); diff --git a/src/rdkafka_broker.c b/src/rdkafka_broker.c index d8e5c48e81..0e565da57f 100644 --- a/src/rdkafka_broker.c +++ b/src/rdkafka_broker.c @@ -372,20 +372,46 @@ void rd_kafka_broker_set_state(rd_kafka_broker_t *rkb, int state) { rd_atomic32_set(&rkb->rkb_down_reported, 1) == 0) { /* Propagate ALL_BROKERS_DOWN event if all brokers are - * now down, unless we're terminating. */ + * now down, unless we're terminating. + * The error is reported only once per outage: the + * re-bootstrap sequence resets each broker's down reported + * state to retry all brokers before reaching this condition + * again, so the error would otherwise be reported on every + * re-bootstrap cycle for the same continuing outage. + * It's armed again when any broker connection comes up. */ if (rd_atomic32_add(&rkb->rkb_rk->rk_broker_down_cnt, 1) == rd_atomic32_get(&rkb->rkb_rk->rk_broker_cnt) - rd_atomic32_get( &rkb->rkb_rk->rk_logical_broker_cnt) && !rd_kafka_terminating(rkb->rkb_rk)) { rd_kafka_rebootstrap(rkb->rkb_rk); - rd_kafka_op_err( - rkb->rkb_rk, RD_KAFKA_RESP_ERR__ALL_BROKERS_DOWN, - "%i/%i brokers are down", - rd_atomic32_get(&rkb->rkb_rk->rk_broker_down_cnt), - rd_atomic32_get(&rkb->rkb_rk->rk_broker_cnt) - - rd_atomic32_get( - &rkb->rkb_rk->rk_logical_broker_cnt)); + if (rd_atomic32_set( + &rkb->rkb_rk->rk_all_brokers_down_reported, + 1) == 0) + rd_kafka_op_err( + rkb->rkb_rk, + RD_KAFKA_RESP_ERR__ALL_BROKERS_DOWN, + "%i/%i brokers are down", + rd_atomic32_get( + &rkb->rkb_rk->rk_broker_down_cnt), + rd_atomic32_get( + &rkb->rkb_rk->rk_broker_cnt) - + rd_atomic32_get( + &rkb->rkb_rk + ->rk_logical_broker_cnt)); + else + rd_kafka_dbg( + rkb->rkb_rk, BROKER, "ALLDOWN", + "%i/%i brokers are down: " + "ALL_BROKERS_DOWN already reported " + "for the current outage", + rd_atomic32_get( + &rkb->rkb_rk->rk_broker_down_cnt), + rd_atomic32_get( + &rkb->rkb_rk->rk_broker_cnt) - + rd_atomic32_get( + &rkb->rkb_rk + ->rk_logical_broker_cnt)); } } else if (rd_kafka_broker_state_is_up(state) && @@ -402,6 +428,12 @@ void rd_kafka_broker_set_state(rd_kafka_broker_t *rkb, int state) { rd_atomic32_add(&rkb->rkb_rk->rk_broker_up_cnt, 1); + /* The outage is over: arm the + * ALL_BROKERS_DOWN report again. */ + rd_atomic32_set( + &rkb->rkb_rk->rk_all_brokers_down_reported, + 0); + /* If at least one broker connects we reset * the down counter to try again with rest of * brokers. Otherwise, a single broker diff --git a/src/rdkafka_int.h b/src/rdkafka_int.h index e2ed5d1dc6..f6f8b4e3df 100644 --- a/src/rdkafka_int.h +++ b/src/rdkafka_int.h @@ -306,6 +306,12 @@ struct rd_kafka_s { * that have had at least one connection attempt * and are configured or learned. */ rd_atomic32_t rk_broker_down_cnt; + /** Set to 1 when ERR__ALL_BROKERS_DOWN has been reported for the + * current outage. Prevents re-reporting the error on every + * re-bootstrap cycle while all brokers remain down, given the + * re-bootstrap sequence resets each broker's down reported state. + * Reset to 0 when any non-logical broker enters the UP state. */ + rd_atomic32_t rk_all_brokers_down_reported; /** Set to 1 when there's a re-bootstrap in progress. * Set to 0 when the re-bootstrap is done. * Accessed from the main thread and the broker threads. */ diff --git a/tests/0152-rebootstrap.c b/tests/0152-rebootstrap.c index e9c9776de5..2e3a0d07b1 100644 --- a/tests/0152-rebootstrap.c +++ b/tests/0152-rebootstrap.c @@ -28,6 +28,8 @@ #include "test.h" +#include "../src/rdkafka_protocol.h" + /** * @brief Verify the case where there are no bootstrap servers * and the client is re-bootstrapped after brokers were added @@ -50,10 +52,289 @@ do_test_rebootstrap_local_no_bootstrap_servers(rd_kafka_type_t rk_type) { SUB_TEST_PASS(); } +static rd_atomic32_t all_brokers_down_cnt; +static rd_atomic32_t rebootstrap_sequence_cnt; +static rd_atomic32_t connect_attempt_cnt; + +/** + * @brief Error callback counting ERR__ALL_BROKERS_DOWN events. + */ +static void sustained_outage_error_cb(rd_kafka_t *rk, + int err, + const char *reason, + void *opaque) { + if ((rd_kafka_resp_err_t)err == RD_KAFKA_RESP_ERR__ALL_BROKERS_DOWN) + rd_atomic32_add(&all_brokers_down_cnt, 1); + TEST_SAY("error_cb: %s: %s\n", + rd_kafka_err2name((rd_kafka_resp_err_t)err), reason); +} + +/** + * @brief Log callback counting re-bootstrap sequences and + * connection attempts. + */ +static void sustained_outage_log_cb(const rd_kafka_t *rk, + int level, + const char *fac, + const char *buf) { + if (strstr(buf, "Starting re-bootstrap sequence")) + rd_atomic32_add(&rebootstrap_sequence_cnt, 1); + if (strstr(buf, "broker in state ") && strstr(buf, "connecting")) + rd_atomic32_add(&connect_attempt_cnt, 1); +} + +/** + * @brief Make the next \p cnt connections fail during the ApiVersion + * handshake: the mock broker closes the connection when returning + * RD_KAFKA_RESP_ERR__TRANSPORT. + */ +static void push_ApiVersion_transport_errors(rd_kafka_mock_cluster_t *mcluster, + size_t cnt) { + rd_kafka_resp_err_t errs[512]; + size_t i; + + TEST_ASSERT(cnt <= RD_ARRAY_SIZE(errs), "cnt too large"); + for (i = 0; i < cnt; i++) + errs[i] = RD_KAFKA_RESP_ERR__TRANSPORT; + rd_kafka_mock_push_request_errors_array(mcluster, RD_KAFKAP_ApiVersion, + cnt, errs); +} + +/** + * @brief Create a consumer connecting to \p bootstraps with the outage + * counters reset and the counting error and log callbacks installed. + */ +static rd_kafka_t * +sustained_outage_client_new(const char *bootstraps, + const char *reconnect_backoff_ms, + const char *reconnect_backoff_max_ms, + test_conf_log_interceptor_t **interceptorp) { + rd_kafka_conf_t *conf; + const char *debug_contexts[2] = {"broker", NULL}; + + rd_atomic32_init(&all_brokers_down_cnt, 0); + rd_atomic32_init(&rebootstrap_sequence_cnt, 0); + rd_atomic32_init(&connect_attempt_cnt, 0); + + test_conf_init(&conf, NULL, 60); + test_conf_set(conf, "bootstrap.servers", bootstraps); + test_conf_set(conf, "reconnect.backoff.ms", reconnect_backoff_ms); + test_conf_set(conf, "reconnect.backoff.max.ms", + reconnect_backoff_max_ms); + test_conf_set(conf, "socket.connection.setup.timeout.ms", "2000"); + test_conf_set(conf, "group.id", "0152-outage"); + rd_kafka_conf_set_error_cb(conf, sustained_outage_error_cb); + *interceptorp = test_conf_set_log_interceptor( + conf, sustained_outage_log_cb, debug_contexts); + + return test_create_handle(RD_KAFKA_CONSUMER, conf); +} + +/** + * @brief Poll \p rk for \p duration_ms milliseconds. + */ +static void poll_for(rd_kafka_t *rk, int duration_ms) { + rd_ts_t ts_end = test_clock() + (rd_ts_t)duration_ms * 1000; + while (test_clock() < ts_end) + rd_kafka_poll(rk, 100); +} + +/** + * @brief Poll \p rk until \p cnt reaches \p expected, + * failing after \p timeout_ms. + */ +static void poll_until_cnt(rd_kafka_t *rk, + rd_atomic32_t *cnt, + int32_t expected, + int timeout_ms, + const char *what) { + rd_ts_t abs_timeout = test_clock() + (rd_ts_t)timeout_ms * 1000; + while (rd_atomic32_get(cnt) < expected) { + TEST_ASSERT(test_clock() < abs_timeout, + "Timed out waiting for %s to reach %d, got %d", + what, expected, rd_atomic32_get(cnt)); + rd_kafka_poll(rk, 100); + } +} + +/** + * @brief Sustained outage: every connection attempt fails during ApiVersion + * negotiation (the peer closes the connection), for longer than + * several re-bootstrap cycles. + * ERR__ALL_BROKERS_DOWN must be reported only once for the whole + * outage while the re-bootstrap sequences keep cycling. + */ +static void do_test_sustained_outage_single_all_brokers_down(void) { + rd_kafka_mock_cluster_t *mcluster; + const char *bootstraps; + rd_kafka_t *rk; + test_conf_log_interceptor_t *interceptor; + int32_t errors, rebootstraps; + + SUB_TEST(); + + mcluster = test_mock_cluster_new(3, &bootstraps); + push_ApiVersion_transport_errors(mcluster, 500); + + rk = + sustained_outage_client_new(bootstraps, "100", "500", &interceptor); + + poll_for(rk, 10000); + + errors = rd_atomic32_get(&all_brokers_down_cnt); + rebootstraps = rd_atomic32_get(&rebootstrap_sequence_cnt); + TEST_SAY("Got %d ALL_BROKERS_DOWN errors, %d re-bootstrap sequences\n", + errors, rebootstraps); + TEST_ASSERT(errors == 1, + "Expected exactly 1 ALL_BROKERS_DOWN error " + "during a sustained outage, got %d", + errors); + TEST_ASSERT(rebootstraps >= 2, + "Expected re-bootstrap to keep cycling " + "(>= 2 sequences), got %d", + rebootstraps); + + rd_kafka_destroy(rk); + test_mock_cluster_destroy(mcluster); + rd_free(interceptor); + SUB_TEST_PASS(); +} + +/** + * @brief ERR__ALL_BROKERS_DOWN is reported once per distinct outage: + * after the cluster recovers and all connections are lost again + * a second error must be reported, exactly once, too. + */ +static void do_test_second_outage_second_error(void) { + rd_kafka_mock_cluster_t *mcluster; + const char *bootstraps; + rd_kafka_t *rk; + test_conf_log_interceptor_t *interceptor; + const struct rd_kafka_metadata *md; + rd_ts_t abs_timeout; + int32_t i, errors; + rd_bool_t recovered = rd_false; + + SUB_TEST(); + + mcluster = test_mock_cluster_new(3, &bootstraps); + push_ApiVersion_transport_errors(mcluster, 500); + + rk = + sustained_outage_client_new(bootstraps, "100", "500", &interceptor); + + TEST_SAY("Outage #1: awaiting the first ALL_BROKERS_DOWN\n"); + poll_until_cnt(rk, &all_brokers_down_cnt, 1, 15000, + "ALL_BROKERS_DOWN errors"); + poll_for(rk, 3000); + errors = rd_atomic32_get(&all_brokers_down_cnt); + TEST_ASSERT(errors == 1, + "Expected exactly 1 ALL_BROKERS_DOWN error " + "after the first outage, got %d", + errors); + + TEST_SAY("Healing the cluster\n"); + rd_kafka_mock_clear_request_errors(mcluster, RD_KAFKAP_ApiVersion); + abs_timeout = test_clock() + 15 * 1000000; + while (!recovered) { + TEST_ASSERT(test_clock() < abs_timeout, + "Timed out waiting for cluster recovery"); + if (rd_kafka_metadata(rk, 0, NULL, &md, 1000) == + RD_KAFKA_RESP_ERR_NO_ERROR) { + rd_kafka_metadata_destroy(md); + recovered = rd_true; + } else { + rd_kafka_poll(rk, 100); + } + } + + TEST_SAY("Outage #2: dropping all connections\n"); + push_ApiVersion_transport_errors(mcluster, 500); + for (i = 1; i <= 3; i++) { + TEST_CALL_ERR__(rd_kafka_mock_broker_set_down(mcluster, i)); + TEST_CALL_ERR__(rd_kafka_mock_broker_set_up(mcluster, i)); + } + + poll_until_cnt(rk, &all_brokers_down_cnt, 2, 15000, + "ALL_BROKERS_DOWN errors"); + poll_for(rk, 3000); + errors = rd_atomic32_get(&all_brokers_down_cnt); + TEST_ASSERT(errors == 2, + "Expected exactly 2 ALL_BROKERS_DOWN errors " + "after the second outage, got %d", + errors); + + rd_kafka_destroy(rk); + test_mock_cluster_destroy(mcluster); + rd_free(interceptor); + SUB_TEST_PASS(); +} + +/** + * @brief The reconnect backoff must keep growing during a sustained outage + * until capped by reconnect.backoff.max.ms, pacing the connection + * attempts: neither the re-bootstrap sequences nor the periodic + * cluster connection maintenance may reset it. + * Each connection attempt fails during ApiVersion negotiation and + * doubles the reconnect backoff (once more when the broker address + * list is exhausted), so with reconnect.backoff.ms=1000 and + * reconnect.backoff.max.ms=5000 only attempts at roughly 0, 1.5-3, + * 4.5-8 and 8.25-13 s fit in a 10 s window. If the backoff were + * reset on each cycle there would be an attempt roughly every + * 0.5-1 s. + */ +static void do_test_backoff_grows_during_outage(void) { + rd_kafka_mock_cluster_t *mcluster; + const char *bootstraps; + rd_kafka_t *rk; + test_conf_log_interceptor_t *interceptor; + int32_t connects, errors; + + SUB_TEST(); + + mcluster = test_mock_cluster_new(1, &bootstraps); + push_ApiVersion_transport_errors(mcluster, 500); + + rk = sustained_outage_client_new(bootstraps, "1000", "5000", + &interceptor); + + poll_for(rk, 10000); + + connects = rd_atomic32_get(&connect_attempt_cnt); + errors = rd_atomic32_get(&all_brokers_down_cnt); + TEST_SAY("Got %d connection attempts, %d ALL_BROKERS_DOWN errors\n", + connects, errors); + TEST_ASSERT(connects >= 3 && connects <= 6, + "Expected 3..6 connection attempts in 10 s with a " + "growing reconnect backoff, got %d", + connects); + TEST_ASSERT(errors == 1, + "Expected exactly 1 ALL_BROKERS_DOWN error, got %d", + errors); + + rd_kafka_destroy(rk); + test_mock_cluster_destroy(mcluster); + rd_free(interceptor); + SUB_TEST_PASS(); +} + int main_0152_rebootstrap_local(int argc, char **argv) { do_test_rebootstrap_local_no_bootstrap_servers(RD_KAFKA_PRODUCER); do_test_rebootstrap_local_no_bootstrap_servers(RD_KAFKA_CONSUMER); + if (test_needs_auth()) { + TEST_SAY( + "Skipping mock cluster subtests: " + "mock cluster does not support SSL/SASL\n"); + return 0; + } + + do_test_sustained_outage_single_all_brokers_down(); + + do_test_second_outage_second_error(); + + do_test_backoff_grows_during_outage(); + return 0; }