Skip to content
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

Jump to bottom

[improve][broker] Add error logs and retry if the broker failed to register itself for metadata node deletion #23390

Closed
wants to merge 2 commits into from
Closed

[improve][broker] Add error logs and retry if the broker failed to register itself for metadata node deletion #23390

Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
b98f7ff
[improve][broker] Add error logs if the broker failed to register its…
BewareMyPower Oct 2, 2024
f0b236f
Add the retry logic
BewareMyPower Oct 2, 2024
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
14 changes: 13 additions & 1 deletion ...ker/src/main/java/org/apache/pulsar/broker/loadbalance/extensions/BrokerRegistryImpl.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -220,7 +220,7 @@ private void handleMetadataStoreNotification(Notification t) {
// is expired. In this case, we should register again.
final var brokerId = t.getPath().substring(LOADBALANCE_BROKERS_ROOT.length() + 1);
if (t.getType() == NotificationType.Deleted && getBrokerId().equals(brokerId)) {
registerAsync();
registerWithRetry();
}
if (listeners.isEmpty()) {
return;
Expand All @@ -235,6 +235,18 @@ private void handleMetadataStoreNotification(Notification t) {
}
}

private void registerWithRetry() {
registerAsync().exceptionallyAsync(e -> {
log.error("[{}] Failed to register self to {} (state: {}), retry registering", getBrokerId(),
brokerIdKeyPath, state.get(), e);
// Keep retrying registering itself with a fixed delay time because all lookup operations rely on the
// successful registering.
// Don't use the load manager executor because that thread might be blocked by some lookup operations.
pulsar.getExecutor().schedule(this::registerWithRetry, 100, TimeUnit.MILLISECONDS);
return null;
});
}

@VisibleForTesting
protected static boolean isVerifiedNotification(Notification t) {
return t.getPath().startsWith(LOADBALANCE_BROKERS_ROOT + "/")
Expand Down
Loading