Skip to content

Report ALL_BROKERS_DOWN only once per outage with metadata.recovery.strategy=rebootstrap #5547

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
Marko Lahma (lahma) wants to merge 1 commit into
base: master
Choose a base branch
from
+353 −8
Open

Report ALL_BROKERS_DOWN only once per outage with metadata.recovery.strategy=rebootstrap #5547

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
25 changes: 25 additions & 0 deletions CHANGELOG.md
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -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:
Expand Down
1 change: 1 addition & 0 deletions src/rdkafka.c
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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);
Expand Down
48 changes: 40 additions & 8 deletions src/rdkafka_broker.c
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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) &&
Expand All @@ -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
Expand Down
6 changes: 6 additions & 0 deletions src/rdkafka_int.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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. */
Expand Down
Loading