You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by "mattisonchao (via GitHub)" <gi...@apache.org> on 2023/06/29 02:39:05 UTC
[GitHub] [pulsar] mattisonchao opened a new pull request, #20642: [fix][meta] Bookie Info lost by notification race condition.
mattisonchao opened a new pull request, #20642:
URL: https://github.com/apache/pulsar/pull/20642
### Motivation
https://github.com/apache/pulsar/issues/20643
### Investigating
#### Why is the exception `Bookie handle is not available`?
We found this problem in the broker's continued to report `Bookie handle is not available` exception longer than 6 hours. The call chain should be as follows:
1. Bk client is trying to [obtain](https://github.com/apache/bookkeeper/blob/23c83a341f352a59081eef1010c50f19d85292ca/bookkeeper-server/src/main/java/org/apache/bookkeeper/proto/PerChannelBookieClientPool.java#L41C9-L41C9) a new channel for connection.
2. The channel was disconnected, So it is trying to [connect](https://github.com/apache/bookkeeper/blob/23c83a341f352a59081eef1010c50f19d85292ca/bookkeeper-server/src/main/java/org/apache/bookkeeper/proto/PerChannelBookieClient.java#L643) again.
3. Bk Client is trying to [get the bookie](https://github.com/apache/bookkeeper/blob/23c83a341f352a59081eef1010c50f19d85292ca/bookkeeper-server/src/main/java/org/apache/bookkeeper/proto/PerChannelBookieClient.java#L539) info by bookie id.
4. Bk client uses pulsar implemented `PulsarRegistrationClient` to [get bookie info](https://github.com/apache/pulsar/blob/master/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/bookkeeper/PulsarRegistrationClient.java#L224), but the local cache doesn't have it, and then throws the `BKBookieHandleNotAvailableException`
#### Why did cache lose Bookie's info?
The following is an analysis that has yet to be rigorously tested. (Since this will consume a relatively long time)
1. Check out our [implementation](https://github.com/apache/pulsar/blob/46b6dcd9571262f72f963a755c8a51fb4f66c81e/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/bookkeeper/PulsarRegistrationClient.java#L174-L192). We will update the local cache async/sync(Depending on the internal metadata cache) and invalidate the local cache synchronously.
2. In Bookie `ZKRegistrationManager`'s [implementation](https://github.com/apache/bookkeeper/blob/23c83a341f352a59081eef1010c50f19d85292ca/bookkeeper-server/src/main/java/org/apache/bookkeeper/discover/ZKRegistrationManager.java#L287-L313), transit bookie mode from writable to read-only should create a read-only bookie node and then delete the writable bookie node.
3. writable -> read-only notification should be as follow:
- CREATED [read-only path]:bookieId
- DELETE [writable path]: bookieId
4. Consider that metadata has a cache that invokes the CREATEED notification [logic](https://github.com/apache/pulsar/blob/46b6dcd9571262f72f963a755c8a51fb4f66c81e/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/bookkeeper/PulsarRegistrationClient.java#L177) synchronously. So, we may create this bookie info first and then delete it. **(bookie info lost here)**
### Modifications
- Make the notification process sequentially.
- Support bookie mode in the local cache.
### Verifying this change
- [x] Make sure that the change passes the CI checks.
### Documentation
<!-- DO NOT REMOVE THIS SECTION. CHECK THE PROPER BOX ONLY. -->
- [ ] `doc` <!-- Your PR contains doc changes. -->
- [ ] `doc-required` <!-- Your PR changes impact docs and you will update later -->
- [x] `doc-not-needed` <!-- Your PR changes do not impact docs -->
- [ ] `doc-complete` <!-- Docs have been already added -->
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] mattisonchao commented on pull request #20642: [fix][meta] Bookie Info lost by notification race condition.
Posted by "mattisonchao (via GitHub)" <gi...@apache.org>.
mattisonchao commented on PR #20642:
URL: https://github.com/apache/pulsar/pull/20642#issuecomment-1609184004
@hangc0276 test has been added.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] codelipenghui commented on a diff in pull request #20642: [fix][meta] Bookie Info lost by notification race condition.
Posted by "codelipenghui (via GitHub)" <gi...@apache.org>.
codelipenghui commented on code in PR #20642:
URL: https://github.com/apache/pulsar/pull/20642#discussion_r1243457522
##########
pulsar-metadata/src/main/java/org/apache/pulsar/metadata/bookkeeper/PulsarRegistrationClient.java:
##########
@@ -153,42 +178,57 @@ public void unwatchReadOnlyBookies(RegistrationListener registrationListener) {
readOnlyBookiesWatchers.remove(registrationListener);
}
- private void handleDeletedBookieNode(Notification n) {
- if (n.getType() == NotificationType.Deleted) {
- BookieId bookieId = stripBookieIdFromPath(n.getPath());
- if (bookieId != null) {
- log.info("Bookie {} disappeared", bookieId);
- bookieServiceInfoCache.remove(bookieId);
- }
+ /**
+ * This method will receive metadata store notifications and then update the
+ * local cache in background sequentially.
+ */
+ private void updatedBookies(Notification n) {
+ // make the notification callback run sequential in background.
+ final String path = n.getPath();
+ if (!path.startsWith(bookieReadonlyRegistrationPath) || !path.startsWith(bookieRegistrationPath)) {
+ // ignore unknown path
+ return;
}
- }
-
- private void handleUpdatedBookieNode(Notification n) {
- BookieId bookieId = stripBookieIdFromPath(n.getPath());
+ final BookieId bookieId = stripBookieIdFromPath(n.getPath());
if (bookieId != null) {
- log.info("Bookie {} info updated", bookieId);
- readBookieServiceInfoAsync(bookieId);
+ log.info("Bookie {} do {}. path: {}", bookieId, n.getType(), n.getPath());
}
- }
-
- private void updatedBookies(Notification n) {
- if (n.getType() == NotificationType.Created || n.getType() == NotificationType.Deleted) {
- if (n.getPath().startsWith(bookieReadonlyRegistrationPath)) {
- getReadOnlyBookies().thenAccept(bookies -> {
- readOnlyBookiesWatchers.forEach(w -> executor.execute(() -> w.onBookiesChanged(bookies)));
- });
- handleDeletedBookieNode(n);
- } else if (n.getPath().startsWith(bookieRegistrationPath)) {
- getWritableBookies().thenAccept(bookies ->
- writableBookiesWatchers.forEach(w -> executor.execute(() -> w.onBookiesChanged(bookies))));
- handleDeletedBookieNode(n);
- }
- } else if (n.getType() == NotificationType.Modified) {
- if (n.getPath().startsWith(bookieReadonlyRegistrationPath)
- || n.getPath().startsWith(bookieRegistrationPath)) {
- handleUpdatedBookieNode(n);
+ sequencer.sequential(() -> {
+ switch (n.getType()) {
+ case Created:
+ if (path.equals(bookieReadonlyRegistrationPath)) {
+ return getReadOnlyBookies().thenAccept(bookies ->
+ readOnlyBookiesWatchers.forEach(w ->
+ executor.execute(() -> w.onBookiesChanged(bookies))));
+ }
+ return getWritableBookies().thenAccept(bookies ->
+ writableBookiesWatchers.forEach(w ->
+ executor.execute(() -> w.onBookiesChanged(bookies))));
+ case Modified:
+ if (bookieId == null) {
+ return completedFuture(null);
+ }
+ if (path.equals(bookieReadonlyRegistrationPath)) {
+ return readBookieInfoAsReadonlyBookie(bookieId).thenApply(__ -> null);
+ }
+ return readBookieServiceInfoAsync(bookieId).thenApply(__ -> null);
Review Comment:
Ok, got it.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] mattisonchao commented on a diff in pull request #20642: [fix][meta] Bookie Info lost by notification race condition.
Posted by "mattisonchao (via GitHub)" <gi...@apache.org>.
mattisonchao commented on code in PR #20642:
URL: https://github.com/apache/pulsar/pull/20642#discussion_r1241116178
##########
pulsar-metadata/src/main/java/org/apache/pulsar/metadata/bookkeeper/PulsarRegistrationClient.java:
##########
@@ -153,42 +173,61 @@ public void unwatchReadOnlyBookies(RegistrationListener registrationListener) {
readOnlyBookiesWatchers.remove(registrationListener);
}
- private void handleDeletedBookieNode(Notification n) {
- if (n.getType() == NotificationType.Deleted) {
- BookieId bookieId = stripBookieIdFromPath(n.getPath());
- if (bookieId != null) {
- log.info("Bookie {} disappeared", bookieId);
- bookieServiceInfoCache.remove(bookieId);
- }
+ /**
+ * This method will receive metadata store notifications and then update the
+ * local cache in background sequentially.
+ */
+ private void updatedBookies(Notification n) {
+ final BookieMode mode;
+ if (n.getPath().startsWith(bookieReadonlyRegistrationPath)) {
+ mode = BookieMode.READ_ONLY;
+ } else if (n.getPath().startsWith(bookieRegistrationPath)) {
+ mode = BookieMode.READ_WRITE;
+ } else {
+ // unknown notification, should not happen.
+ return;
Review Comment:
Got rid of `BookieMode` already :)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] liangyepianzhou commented on pull request #20642: [fix][meta] Bookie Info lost by notification race condition.
Posted by "liangyepianzhou (via GitHub)" <gi...@apache.org>.
liangyepianzhou commented on PR #20642:
URL: https://github.com/apache/pulsar/pull/20642#issuecomment-1626524140
@hangc0276 @mattisonchao The zookeeper version quoted in this P person is 3.8.1(master) instead of 3.6.4 (branch-2.10).
We should not upgrade the version in branch 2.10, according to https://github.com/apache/pulsar/pull/19425.
Could you please help take a look at whether we need to cherry-pick this PR to branch 2.10?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] github-actions[bot] commented on pull request #20642: [fix][metadata] Bookie Info lost by notifier race condition.
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #20642:
URL: https://github.com/apache/pulsar/pull/20642#issuecomment-1605918138
@mattisonchao Please add the following content to your PR description and select a checkbox:
```
- [ ] `doc` <!-- Your PR contains doc changes -->
- [ ] `doc-required` <!-- Your PR changes impact docs and you will update later -->
- [ ] `doc-not-needed` <!-- Your PR changes do not impact docs -->
- [ ] `doc-complete` <!-- Docs have been already added -->
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] mattisonchao commented on a diff in pull request #20642: [fix][meta] Bookie Info lost by notification race condition.
Posted by "mattisonchao (via GitHub)" <gi...@apache.org>.
mattisonchao commented on code in PR #20642:
URL: https://github.com/apache/pulsar/pull/20642#discussion_r1243230366
##########
pulsar-metadata/src/main/java/org/apache/pulsar/metadata/bookkeeper/PulsarRegistrationClient.java:
##########
@@ -227,35 +267,37 @@ public CompletableFuture<Versioned<BookieServiceInfo>> getBookieServiceInfo(Book
// this is because there are a few cases in which some operations on the main thread
// wait for the result. This is due to the fact that resolving the address of a bookie
// is needed in many code paths.
- Versioned<BookieServiceInfo> resultFromCache = bookieServiceInfoCache.get(bookieId);
+ Versioned<BookieServiceInfo> info;
+ if ((info = writableBookieInfo.get(bookieId)) == null) {
+ info = readOnlyBookieInfo.get(bookieId);
+ }
if (log.isDebugEnabled()) {
- log.debug("getBookieServiceInfo {} -> {}", bookieId, resultFromCache);
+ log.debug("getBookieServiceInfo {} -> {}", bookieId, info);
}
- if (resultFromCache != null) {
- return CompletableFuture.completedFuture(resultFromCache);
+ if (info != null) {
+ return completedFuture(info);
} else {
return FutureUtils.exception(new BKException.BKBookieHandleNotAvailableException());
}
}
- public CompletableFuture<Void> readBookieServiceInfoAsync(BookieId bookieId) {
+ public CompletableFuture<Boolean> readBookieServiceInfoAsync(BookieId bookieId) {
String asWritable = bookieRegistrationPath + "/" + bookieId;
return bookieServiceInfoMetadataCache.get(asWritable)
- .thenCompose((Optional<BookieServiceInfo> getResult) -> {
- if (getResult.isPresent()) {
- Versioned<BookieServiceInfo> res =
- new Versioned<>(getResult.get(), new LongVersion(-1));
- log.info("Update BookieInfoCache (writable bookie) {} -> {}", bookieId, getResult.get());
- bookieServiceInfoCache.put(bookieId, res);
- return CompletableFuture.completedFuture(null);
- } else {
- return readBookieInfoAsReadonlyBookie(bookieId);
- }
- }
- );
+ .thenApply((Optional<BookieServiceInfo> getResult) -> {
+ if (getResult.isPresent()) {
+ Versioned<BookieServiceInfo> res =
+ new Versioned<>(getResult.get(), new LongVersion(-1));
Review Comment:
I put the version information in it. Because bk defined the interface as follows.
```java
default CompletableFuture<Versioned<BookieServiceInfo>> getBookieServiceInfo(BookieId bookieId)
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] JooHyukKim commented on a diff in pull request #20642: [fix][meta] Bookie Info lost by notification race condition.
Posted by "JooHyukKim (via GitHub)" <gi...@apache.org>.
JooHyukKim commented on code in PR #20642:
URL: https://github.com/apache/pulsar/pull/20642#discussion_r1241099760
##########
pulsar-metadata/src/main/java/org/apache/pulsar/metadata/bookkeeper/PulsarRegistrationClient.java:
##########
@@ -153,42 +173,61 @@ public void unwatchReadOnlyBookies(RegistrationListener registrationListener) {
readOnlyBookiesWatchers.remove(registrationListener);
}
- private void handleDeletedBookieNode(Notification n) {
- if (n.getType() == NotificationType.Deleted) {
- BookieId bookieId = stripBookieIdFromPath(n.getPath());
- if (bookieId != null) {
- log.info("Bookie {} disappeared", bookieId);
- bookieServiceInfoCache.remove(bookieId);
- }
+ /**
+ * This method will receive metadata store notifications and then update the
+ * local cache in background sequentially.
+ */
+ private void updatedBookies(Notification n) {
+ final BookieMode mode;
+ if (n.getPath().startsWith(bookieReadonlyRegistrationPath)) {
+ mode = BookieMode.READ_ONLY;
+ } else if (n.getPath().startsWith(bookieRegistrationPath)) {
+ mode = BookieMode.READ_WRITE;
+ } else {
+ // unknown notification, should not happen.
+ return;
Review Comment:
Would there be any alternative location to do this parsing and `BookieMode` enum, like `RegistrationClient` interface? I was wondering maybe this `if-else` block can at least be extracted as a method or become a Value Object.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] mattisonchao merged pull request #20642: [fix][meta] Bookie Info lost by notification race condition.
Posted by "mattisonchao (via GitHub)" <gi...@apache.org>.
mattisonchao merged PR #20642:
URL: https://github.com/apache/pulsar/pull/20642
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] mattisonchao commented on a diff in pull request #20642: [fix][meta] Bookie Info lost by notification race condition.
Posted by "mattisonchao (via GitHub)" <gi...@apache.org>.
mattisonchao commented on code in PR #20642:
URL: https://github.com/apache/pulsar/pull/20642#discussion_r1241116178
##########
pulsar-metadata/src/main/java/org/apache/pulsar/metadata/bookkeeper/PulsarRegistrationClient.java:
##########
@@ -153,42 +173,61 @@ public void unwatchReadOnlyBookies(RegistrationListener registrationListener) {
readOnlyBookiesWatchers.remove(registrationListener);
}
- private void handleDeletedBookieNode(Notification n) {
- if (n.getType() == NotificationType.Deleted) {
- BookieId bookieId = stripBookieIdFromPath(n.getPath());
- if (bookieId != null) {
- log.info("Bookie {} disappeared", bookieId);
- bookieServiceInfoCache.remove(bookieId);
- }
+ /**
+ * This method will receive metadata store notifications and then update the
+ * local cache in background sequentially.
+ */
+ private void updatedBookies(Notification n) {
+ final BookieMode mode;
+ if (n.getPath().startsWith(bookieReadonlyRegistrationPath)) {
+ mode = BookieMode.READ_ONLY;
+ } else if (n.getPath().startsWith(bookieRegistrationPath)) {
+ mode = BookieMode.READ_WRITE;
+ } else {
+ // unknown notification, should not happen.
+ return;
Review Comment:
Get ride of `BookieMode` already :)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] eolivelli commented on a diff in pull request #20642: [fix][meta] Bookie Info lost by notification race condition.
Posted by "eolivelli (via GitHub)" <gi...@apache.org>.
eolivelli commented on code in PR #20642:
URL: https://github.com/apache/pulsar/pull/20642#discussion_r1241950327
##########
pulsar-metadata/src/main/java/org/apache/pulsar/metadata/bookkeeper/PulsarRegistrationClient.java:
##########
@@ -153,42 +176,56 @@ public void unwatchReadOnlyBookies(RegistrationListener registrationListener) {
readOnlyBookiesWatchers.remove(registrationListener);
}
- private void handleDeletedBookieNode(Notification n) {
- if (n.getType() == NotificationType.Deleted) {
- BookieId bookieId = stripBookieIdFromPath(n.getPath());
- if (bookieId != null) {
- log.info("Bookie {} disappeared", bookieId);
- bookieServiceInfoCache.remove(bookieId);
- }
- }
- }
-
- private void handleUpdatedBookieNode(Notification n) {
- BookieId bookieId = stripBookieIdFromPath(n.getPath());
- if (bookieId != null) {
- log.info("Bookie {} info updated", bookieId);
- readBookieServiceInfoAsync(bookieId);
- }
- }
-
+ /**
+ * This method will receive metadata store notifications and then update the
+ * local cache in background sequentially.
+ */
private void updatedBookies(Notification n) {
- if (n.getType() == NotificationType.Created || n.getType() == NotificationType.Deleted) {
- if (n.getPath().startsWith(bookieReadonlyRegistrationPath)) {
- getReadOnlyBookies().thenAccept(bookies -> {
- readOnlyBookiesWatchers.forEach(w -> executor.execute(() -> w.onBookiesChanged(bookies)));
- });
- handleDeletedBookieNode(n);
- } else if (n.getPath().startsWith(bookieRegistrationPath)) {
- getWritableBookies().thenAccept(bookies ->
- writableBookiesWatchers.forEach(w -> executor.execute(() -> w.onBookiesChanged(bookies))));
- handleDeletedBookieNode(n);
- }
- } else if (n.getType() == NotificationType.Modified) {
- if (n.getPath().startsWith(bookieReadonlyRegistrationPath)
- || n.getPath().startsWith(bookieRegistrationPath)) {
- handleUpdatedBookieNode(n);
- }
+ // make the notification callback run sequential in background.
+ final String path = n.getPath();
+ if (!path.startsWith(bookieReadonlyRegistrationPath) || !path.startsWith(bookieRegistrationPath)) {
+ // ignore unknown path
+ return;
}
+ final BookieId bookieId = stripBookieIdFromPath(n.getPath());
+ log.info("Bookie {} do {}. path: {}", bookieId, n.getType(), n.getPath());
Review Comment:
we should log this only if bookieId is available
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] coderzc commented on a diff in pull request #20642: [fix][meta] Bookie Info lost by notification race condition.
Posted by "coderzc (via GitHub)" <gi...@apache.org>.
coderzc commented on code in PR #20642:
URL: https://github.com/apache/pulsar/pull/20642#discussion_r1243418768
##########
pulsar-metadata/src/main/java/org/apache/pulsar/metadata/bookkeeper/PulsarRegistrationClient.java:
##########
@@ -227,46 +276,48 @@ public CompletableFuture<Versioned<BookieServiceInfo>> getBookieServiceInfo(Book
// this is because there are a few cases in which some operations on the main thread
// wait for the result. This is due to the fact that resolving the address of a bookie
// is needed in many code paths.
- Versioned<BookieServiceInfo> resultFromCache = bookieServiceInfoCache.get(bookieId);
+ Versioned<BookieServiceInfo> info;
+ if ((info = writableBookieInfo.get(bookieId)) == null) {
+ info = readOnlyBookieInfo.get(bookieId);
+ }
if (log.isDebugEnabled()) {
- log.debug("getBookieServiceInfo {} -> {}", bookieId, resultFromCache);
+ log.debug("getBookieServiceInfo {} -> {}", bookieId, info);
}
- if (resultFromCache != null) {
- return CompletableFuture.completedFuture(resultFromCache);
+ if (info != null) {
+ return completedFuture(info);
} else {
return FutureUtils.exception(new BKException.BKBookieHandleNotAvailableException());
}
}
- public CompletableFuture<Void> readBookieServiceInfoAsync(BookieId bookieId) {
- String asWritable = bookieRegistrationPath + "/" + bookieId;
- return bookieServiceInfoMetadataCache.get(asWritable)
- .thenCompose((Optional<BookieServiceInfo> getResult) -> {
- if (getResult.isPresent()) {
- Versioned<BookieServiceInfo> res =
- new Versioned<>(getResult.get(), new LongVersion(-1));
- log.info("Update BookieInfoCache (writable bookie) {} -> {}", bookieId, getResult.get());
- bookieServiceInfoCache.put(bookieId, res);
- return CompletableFuture.completedFuture(null);
- } else {
- return readBookieInfoAsReadonlyBookie(bookieId);
- }
- }
- );
+ public CompletableFuture<Optional<CacheGetResult<BookieServiceInfo>>> readBookieServiceInfoAsync(
Review Comment:
```suggestion
public CompletableFuture<Optional<CacheGetResult<BookieServiceInfo>>> readBookieInfoAsWritableBookie(
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] mattisonchao commented on a diff in pull request #20642: [fix][meta] Bookie Info lost by notification race condition.
Posted by "mattisonchao (via GitHub)" <gi...@apache.org>.
mattisonchao commented on code in PR #20642:
URL: https://github.com/apache/pulsar/pull/20642#discussion_r1243484097
##########
pulsar-metadata/src/main/java/org/apache/pulsar/metadata/bookkeeper/PulsarRegistrationClient.java:
##########
@@ -227,46 +276,48 @@ public CompletableFuture<Versioned<BookieServiceInfo>> getBookieServiceInfo(Book
// this is because there are a few cases in which some operations on the main thread
// wait for the result. This is due to the fact that resolving the address of a bookie
// is needed in many code paths.
- Versioned<BookieServiceInfo> resultFromCache = bookieServiceInfoCache.get(bookieId);
+ Versioned<BookieServiceInfo> info;
+ if ((info = writableBookieInfo.get(bookieId)) == null) {
+ info = readOnlyBookieInfo.get(bookieId);
+ }
if (log.isDebugEnabled()) {
- log.debug("getBookieServiceInfo {} -> {}", bookieId, resultFromCache);
+ log.debug("getBookieServiceInfo {} -> {}", bookieId, info);
}
- if (resultFromCache != null) {
- return CompletableFuture.completedFuture(resultFromCache);
+ if (info != null) {
+ return completedFuture(info);
} else {
return FutureUtils.exception(new BKException.BKBookieHandleNotAvailableException());
}
}
- public CompletableFuture<Void> readBookieServiceInfoAsync(BookieId bookieId) {
- String asWritable = bookieRegistrationPath + "/" + bookieId;
- return bookieServiceInfoMetadataCache.get(asWritable)
- .thenCompose((Optional<BookieServiceInfo> getResult) -> {
- if (getResult.isPresent()) {
- Versioned<BookieServiceInfo> res =
- new Versioned<>(getResult.get(), new LongVersion(-1));
- log.info("Update BookieInfoCache (writable bookie) {} -> {}", bookieId, getResult.get());
- bookieServiceInfoCache.put(bookieId, res);
- return CompletableFuture.completedFuture(null);
- } else {
- return readBookieInfoAsReadonlyBookie(bookieId);
- }
- }
- );
+ public CompletableFuture<Optional<CacheGetResult<BookieServiceInfo>>> readBookieServiceInfoAsync(
Review Comment:
fixed.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] mattisonchao commented on a diff in pull request #20642: [fix][meta] Bookie Info lost by notification race condition.
Posted by "mattisonchao (via GitHub)" <gi...@apache.org>.
mattisonchao commented on code in PR #20642:
URL: https://github.com/apache/pulsar/pull/20642#discussion_r1241969377
##########
pulsar-metadata/src/main/java/org/apache/pulsar/metadata/bookkeeper/PulsarRegistrationClient.java:
##########
@@ -153,42 +176,56 @@ public void unwatchReadOnlyBookies(RegistrationListener registrationListener) {
readOnlyBookiesWatchers.remove(registrationListener);
}
- private void handleDeletedBookieNode(Notification n) {
- if (n.getType() == NotificationType.Deleted) {
- BookieId bookieId = stripBookieIdFromPath(n.getPath());
- if (bookieId != null) {
- log.info("Bookie {} disappeared", bookieId);
- bookieServiceInfoCache.remove(bookieId);
- }
- }
- }
-
- private void handleUpdatedBookieNode(Notification n) {
- BookieId bookieId = stripBookieIdFromPath(n.getPath());
- if (bookieId != null) {
- log.info("Bookie {} info updated", bookieId);
- readBookieServiceInfoAsync(bookieId);
- }
- }
-
+ /**
+ * This method will receive metadata store notifications and then update the
+ * local cache in background sequentially.
+ */
private void updatedBookies(Notification n) {
- if (n.getType() == NotificationType.Created || n.getType() == NotificationType.Deleted) {
- if (n.getPath().startsWith(bookieReadonlyRegistrationPath)) {
- getReadOnlyBookies().thenAccept(bookies -> {
- readOnlyBookiesWatchers.forEach(w -> executor.execute(() -> w.onBookiesChanged(bookies)));
- });
- handleDeletedBookieNode(n);
- } else if (n.getPath().startsWith(bookieRegistrationPath)) {
- getWritableBookies().thenAccept(bookies ->
- writableBookiesWatchers.forEach(w -> executor.execute(() -> w.onBookiesChanged(bookies))));
- handleDeletedBookieNode(n);
- }
- } else if (n.getType() == NotificationType.Modified) {
- if (n.getPath().startsWith(bookieReadonlyRegistrationPath)
- || n.getPath().startsWith(bookieRegistrationPath)) {
- handleUpdatedBookieNode(n);
- }
+ // make the notification callback run sequential in background.
+ final String path = n.getPath();
+ if (!path.startsWith(bookieReadonlyRegistrationPath) || !path.startsWith(bookieRegistrationPath)) {
+ // ignore unknown path
+ return;
}
+ final BookieId bookieId = stripBookieIdFromPath(n.getPath());
+ log.info("Bookie {} do {}. path: {}", bookieId, n.getType(), n.getPath());
Review Comment:
fixed.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] mattisonchao closed pull request #20642: [fix][meta] Bookie Info lost by notification race condition.
Posted by "mattisonchao (via GitHub)" <gi...@apache.org>.
mattisonchao closed pull request #20642: [fix][meta] Bookie Info lost by notification race condition.
URL: https://github.com/apache/pulsar/pull/20642
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] tisonkun commented on pull request #20642: [fix][meta] Bookie Info lost by notification race condition.
Posted by "tisonkun (via GitHub)" <gi...@apache.org>.
tisonkun commented on PR #20642:
URL: https://github.com/apache/pulsar/pull/20642#issuecomment-1613986545
Great.
@mattisonchao one suggestion is that you can link to a permalink instead of `blob/main` where the line number will quickly be invalid.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] codelipenghui commented on a diff in pull request #20642: [fix][meta] Bookie Info lost by notification race condition.
Posted by "codelipenghui (via GitHub)" <gi...@apache.org>.
codelipenghui commented on code in PR #20642:
URL: https://github.com/apache/pulsar/pull/20642#discussion_r1243457958
##########
pulsar-metadata/src/main/java/org/apache/pulsar/metadata/bookkeeper/PulsarRegistrationClient.java:
##########
@@ -227,35 +267,37 @@ public CompletableFuture<Versioned<BookieServiceInfo>> getBookieServiceInfo(Book
// this is because there are a few cases in which some operations on the main thread
// wait for the result. This is due to the fact that resolving the address of a bookie
// is needed in many code paths.
- Versioned<BookieServiceInfo> resultFromCache = bookieServiceInfoCache.get(bookieId);
+ Versioned<BookieServiceInfo> info;
+ if ((info = writableBookieInfo.get(bookieId)) == null) {
+ info = readOnlyBookieInfo.get(bookieId);
+ }
if (log.isDebugEnabled()) {
- log.debug("getBookieServiceInfo {} -> {}", bookieId, resultFromCache);
+ log.debug("getBookieServiceInfo {} -> {}", bookieId, info);
}
- if (resultFromCache != null) {
- return CompletableFuture.completedFuture(resultFromCache);
+ if (info != null) {
+ return completedFuture(info);
} else {
return FutureUtils.exception(new BKException.BKBookieHandleNotAvailableException());
}
}
- public CompletableFuture<Void> readBookieServiceInfoAsync(BookieId bookieId) {
+ public CompletableFuture<Boolean> readBookieServiceInfoAsync(BookieId bookieId) {
String asWritable = bookieRegistrationPath + "/" + bookieId;
return bookieServiceInfoMetadataCache.get(asWritable)
- .thenCompose((Optional<BookieServiceInfo> getResult) -> {
- if (getResult.isPresent()) {
- Versioned<BookieServiceInfo> res =
- new Versioned<>(getResult.get(), new LongVersion(-1));
- log.info("Update BookieInfoCache (writable bookie) {} -> {}", bookieId, getResult.get());
- bookieServiceInfoCache.put(bookieId, res);
- return CompletableFuture.completedFuture(null);
- } else {
- return readBookieInfoAsReadonlyBookie(bookieId);
- }
- }
- );
+ .thenApply((Optional<BookieServiceInfo> getResult) -> {
+ if (getResult.isPresent()) {
+ Versioned<BookieServiceInfo> res =
+ new Versioned<>(getResult.get(), new LongVersion(-1));
Review Comment:
Oh, I see.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] codelipenghui commented on a diff in pull request #20642: [fix][meta] Bookie Info lost by notification race condition.
Posted by "codelipenghui (via GitHub)" <gi...@apache.org>.
codelipenghui commented on code in PR #20642:
URL: https://github.com/apache/pulsar/pull/20642#discussion_r1243173969
##########
pulsar-metadata/src/main/java/org/apache/pulsar/metadata/bookkeeper/PulsarRegistrationClient.java:
##########
@@ -153,42 +178,57 @@ public void unwatchReadOnlyBookies(RegistrationListener registrationListener) {
readOnlyBookiesWatchers.remove(registrationListener);
}
- private void handleDeletedBookieNode(Notification n) {
- if (n.getType() == NotificationType.Deleted) {
- BookieId bookieId = stripBookieIdFromPath(n.getPath());
- if (bookieId != null) {
- log.info("Bookie {} disappeared", bookieId);
- bookieServiceInfoCache.remove(bookieId);
- }
+ /**
+ * This method will receive metadata store notifications and then update the
+ * local cache in background sequentially.
+ */
+ private void updatedBookies(Notification n) {
+ // make the notification callback run sequential in background.
+ final String path = n.getPath();
+ if (!path.startsWith(bookieReadonlyRegistrationPath) || !path.startsWith(bookieRegistrationPath)) {
Review Comment:
It should be `&&` here? Otherwise, we will miss the update from the `bookieRegistrationPath` because `!path.startsWith(bookieReadonlyRegistrationPath)` will be true.
##########
pulsar-metadata/src/main/java/org/apache/pulsar/metadata/bookkeeper/PulsarRegistrationClient.java:
##########
@@ -91,38 +96,58 @@ public void close() {
@Override
public CompletableFuture<Versioned<Set<BookieId>>> getWritableBookies() {
- return getChildren(bookieRegistrationPath);
+ return getBookiesThenFreshCache(bookieRegistrationPath);
}
@Override
public CompletableFuture<Versioned<Set<BookieId>>> getAllBookies() {
// this method is meant to return all the known bookies, even the bookies
// that are not in a running state
- return getChildren(bookieAllRegistrationPath);
+ return getBookiesThenFreshCache(bookieAllRegistrationPath);
}
@Override
public CompletableFuture<Versioned<Set<BookieId>>> getReadOnlyBookies() {
- return getChildren(bookieReadonlyRegistrationPath);
+ return getBookiesThenFreshCache(bookieReadonlyRegistrationPath);
}
- private CompletableFuture<Versioned<Set<BookieId>>> getChildren(String path) {
+ /**
+ * @throws IllegalArgumentException if parameter path is null or empty.
+ */
+ private CompletableFuture<Versioned<Set<BookieId>>> getBookiesThenFreshCache(String path) {
+ if (path == null || path.isEmpty()) {
+ return failedFuture(
+ new IllegalArgumentException(String.format("parameter [path] can not be null or empty.")));
+ }
return store.getChildren(path)
.thenComposeAsync(children -> {
- Set<BookieId> bookieIds = PulsarRegistrationClient.convertToBookieAddresses(children);
- List<CompletableFuture<?>> bookieInfoUpdated =
- new ArrayList<>(bookieIds.size());
+ final Set<BookieId> bookieIds = PulsarRegistrationClient.convertToBookieAddresses(children);
+ final List<CompletableFuture<?>> bookieInfoUpdated = new ArrayList<>(bookieIds.size());
for (BookieId id : bookieIds) {
// update the cache for new bookies
- if (!bookieServiceInfoCache.containsKey(id)) {
+ if (path.equals(bookieRegistrationPath) && writableBookieInfo.get(id) == null) {
bookieInfoUpdated.add(readBookieServiceInfoAsync(id));
+ continue;
+ }
+ if (path.equals(bookieReadonlyRegistrationPath) && readOnlyBookieInfo.get(id) == null) {
+ bookieInfoUpdated.add(readBookieInfoAsReadonlyBookie(id));
+ continue;
+ }
+ if (path.equals(bookieAllRegistrationPath)) {
+ if (writableBookieInfo.get(id) != null || readOnlyBookieInfo != null) {
Review Comment:
```suggestion
if (writableBookieInfo.get(id) != null || readOnlyBookieInfo.get(id) != null) {
```
##########
pulsar-metadata/src/main/java/org/apache/pulsar/metadata/bookkeeper/PulsarRegistrationClient.java:
##########
@@ -91,38 +96,58 @@ public void close() {
@Override
public CompletableFuture<Versioned<Set<BookieId>>> getWritableBookies() {
- return getChildren(bookieRegistrationPath);
+ return getBookiesThenFreshCache(bookieRegistrationPath);
}
@Override
public CompletableFuture<Versioned<Set<BookieId>>> getAllBookies() {
// this method is meant to return all the known bookies, even the bookies
// that are not in a running state
- return getChildren(bookieAllRegistrationPath);
+ return getBookiesThenFreshCache(bookieAllRegistrationPath);
}
@Override
public CompletableFuture<Versioned<Set<BookieId>>> getReadOnlyBookies() {
- return getChildren(bookieReadonlyRegistrationPath);
+ return getBookiesThenFreshCache(bookieReadonlyRegistrationPath);
}
- private CompletableFuture<Versioned<Set<BookieId>>> getChildren(String path) {
+ /**
+ * @throws IllegalArgumentException if parameter path is null or empty.
+ */
+ private CompletableFuture<Versioned<Set<BookieId>>> getBookiesThenFreshCache(String path) {
+ if (path == null || path.isEmpty()) {
+ return failedFuture(
+ new IllegalArgumentException(String.format("parameter [path] can not be null or empty.")));
+ }
return store.getChildren(path)
.thenComposeAsync(children -> {
- Set<BookieId> bookieIds = PulsarRegistrationClient.convertToBookieAddresses(children);
- List<CompletableFuture<?>> bookieInfoUpdated =
- new ArrayList<>(bookieIds.size());
+ final Set<BookieId> bookieIds = PulsarRegistrationClient.convertToBookieAddresses(children);
+ final List<CompletableFuture<?>> bookieInfoUpdated = new ArrayList<>(bookieIds.size());
for (BookieId id : bookieIds) {
// update the cache for new bookies
- if (!bookieServiceInfoCache.containsKey(id)) {
+ if (path.equals(bookieRegistrationPath) && writableBookieInfo.get(id) == null) {
bookieInfoUpdated.add(readBookieServiceInfoAsync(id));
+ continue;
+ }
+ if (path.equals(bookieReadonlyRegistrationPath) && readOnlyBookieInfo.get(id) == null) {
+ bookieInfoUpdated.add(readBookieInfoAsReadonlyBookie(id));
+ continue;
+ }
+ if (path.equals(bookieAllRegistrationPath)) {
+ if (writableBookieInfo.get(id) != null || readOnlyBookieInfo != null) {
+ // jump to next bookie id
+ continue;
+ }
+ bookieInfoUpdated.add(readBookieServiceInfoAsync(id) // check writable first
+ .thenCompose(updated ->
+ updated ? completedFuture(null)
+ : readBookieInfoAsReadonlyBookie(id))); // check read-only then
Review Comment:
It will be easier to read the code by changing the return type to `Optional<BookieServiceInfo>` for methods `readBookieServiceInfoAsync` and `readBookieInfoAsReadonlyBookie`. So that we can check if the BookieServiceInfo is present or not here. It's a little confusing to use a BOOL value `updated` here.
##########
pulsar-metadata/src/main/java/org/apache/pulsar/metadata/bookkeeper/PulsarRegistrationClient.java:
##########
@@ -153,42 +178,57 @@ public void unwatchReadOnlyBookies(RegistrationListener registrationListener) {
readOnlyBookiesWatchers.remove(registrationListener);
}
- private void handleDeletedBookieNode(Notification n) {
- if (n.getType() == NotificationType.Deleted) {
- BookieId bookieId = stripBookieIdFromPath(n.getPath());
- if (bookieId != null) {
- log.info("Bookie {} disappeared", bookieId);
- bookieServiceInfoCache.remove(bookieId);
- }
+ /**
+ * This method will receive metadata store notifications and then update the
+ * local cache in background sequentially.
+ */
+ private void updatedBookies(Notification n) {
+ // make the notification callback run sequential in background.
+ final String path = n.getPath();
+ if (!path.startsWith(bookieReadonlyRegistrationPath) || !path.startsWith(bookieRegistrationPath)) {
+ // ignore unknown path
+ return;
}
- }
-
- private void handleUpdatedBookieNode(Notification n) {
- BookieId bookieId = stripBookieIdFromPath(n.getPath());
+ final BookieId bookieId = stripBookieIdFromPath(n.getPath());
if (bookieId != null) {
- log.info("Bookie {} info updated", bookieId);
- readBookieServiceInfoAsync(bookieId);
+ log.info("Bookie {} do {}. path: {}", bookieId, n.getType(), n.getPath());
}
- }
-
- private void updatedBookies(Notification n) {
- if (n.getType() == NotificationType.Created || n.getType() == NotificationType.Deleted) {
- if (n.getPath().startsWith(bookieReadonlyRegistrationPath)) {
- getReadOnlyBookies().thenAccept(bookies -> {
- readOnlyBookiesWatchers.forEach(w -> executor.execute(() -> w.onBookiesChanged(bookies)));
- });
- handleDeletedBookieNode(n);
- } else if (n.getPath().startsWith(bookieRegistrationPath)) {
- getWritableBookies().thenAccept(bookies ->
- writableBookiesWatchers.forEach(w -> executor.execute(() -> w.onBookiesChanged(bookies))));
- handleDeletedBookieNode(n);
- }
- } else if (n.getType() == NotificationType.Modified) {
- if (n.getPath().startsWith(bookieReadonlyRegistrationPath)
- || n.getPath().startsWith(bookieRegistrationPath)) {
- handleUpdatedBookieNode(n);
+ sequencer.sequential(() -> {
+ switch (n.getType()) {
+ case Created:
+ if (path.equals(bookieReadonlyRegistrationPath)) {
Review Comment:
It should be `.startsWith`, not `.equals`?
##########
pulsar-metadata/src/main/java/org/apache/pulsar/metadata/bookkeeper/PulsarRegistrationClient.java:
##########
@@ -227,35 +267,37 @@ public CompletableFuture<Versioned<BookieServiceInfo>> getBookieServiceInfo(Book
// this is because there are a few cases in which some operations on the main thread
// wait for the result. This is due to the fact that resolving the address of a bookie
// is needed in many code paths.
- Versioned<BookieServiceInfo> resultFromCache = bookieServiceInfoCache.get(bookieId);
+ Versioned<BookieServiceInfo> info;
+ if ((info = writableBookieInfo.get(bookieId)) == null) {
+ info = readOnlyBookieInfo.get(bookieId);
+ }
if (log.isDebugEnabled()) {
- log.debug("getBookieServiceInfo {} -> {}", bookieId, resultFromCache);
+ log.debug("getBookieServiceInfo {} -> {}", bookieId, info);
}
- if (resultFromCache != null) {
- return CompletableFuture.completedFuture(resultFromCache);
+ if (info != null) {
+ return completedFuture(info);
} else {
return FutureUtils.exception(new BKException.BKBookieHandleNotAvailableException());
}
}
- public CompletableFuture<Void> readBookieServiceInfoAsync(BookieId bookieId) {
+ public CompletableFuture<Boolean> readBookieServiceInfoAsync(BookieId bookieId) {
String asWritable = bookieRegistrationPath + "/" + bookieId;
return bookieServiceInfoMetadataCache.get(asWritable)
- .thenCompose((Optional<BookieServiceInfo> getResult) -> {
- if (getResult.isPresent()) {
- Versioned<BookieServiceInfo> res =
- new Versioned<>(getResult.get(), new LongVersion(-1));
- log.info("Update BookieInfoCache (writable bookie) {} -> {}", bookieId, getResult.get());
- bookieServiceInfoCache.put(bookieId, res);
- return CompletableFuture.completedFuture(null);
- } else {
- return readBookieInfoAsReadonlyBookie(bookieId);
- }
- }
- );
+ .thenApply((Optional<BookieServiceInfo> getResult) -> {
+ if (getResult.isPresent()) {
+ Versioned<BookieServiceInfo> res =
+ new Versioned<>(getResult.get(), new LongVersion(-1));
Review Comment:
Can we remove the Versioned wrapper? I will always be `-1`, right?
##########
pulsar-metadata/src/main/java/org/apache/pulsar/metadata/bookkeeper/PulsarRegistrationClient.java:
##########
@@ -153,42 +178,57 @@ public void unwatchReadOnlyBookies(RegistrationListener registrationListener) {
readOnlyBookiesWatchers.remove(registrationListener);
}
- private void handleDeletedBookieNode(Notification n) {
- if (n.getType() == NotificationType.Deleted) {
- BookieId bookieId = stripBookieIdFromPath(n.getPath());
- if (bookieId != null) {
- log.info("Bookie {} disappeared", bookieId);
- bookieServiceInfoCache.remove(bookieId);
- }
+ /**
+ * This method will receive metadata store notifications and then update the
+ * local cache in background sequentially.
+ */
+ private void updatedBookies(Notification n) {
+ // make the notification callback run sequential in background.
+ final String path = n.getPath();
+ if (!path.startsWith(bookieReadonlyRegistrationPath) || !path.startsWith(bookieRegistrationPath)) {
+ // ignore unknown path
+ return;
}
- }
-
- private void handleUpdatedBookieNode(Notification n) {
- BookieId bookieId = stripBookieIdFromPath(n.getPath());
+ final BookieId bookieId = stripBookieIdFromPath(n.getPath());
if (bookieId != null) {
- log.info("Bookie {} info updated", bookieId);
- readBookieServiceInfoAsync(bookieId);
+ log.info("Bookie {} do {}. path: {}", bookieId, n.getType(), n.getPath());
}
- }
-
- private void updatedBookies(Notification n) {
- if (n.getType() == NotificationType.Created || n.getType() == NotificationType.Deleted) {
- if (n.getPath().startsWith(bookieReadonlyRegistrationPath)) {
- getReadOnlyBookies().thenAccept(bookies -> {
- readOnlyBookiesWatchers.forEach(w -> executor.execute(() -> w.onBookiesChanged(bookies)));
- });
- handleDeletedBookieNode(n);
- } else if (n.getPath().startsWith(bookieRegistrationPath)) {
- getWritableBookies().thenAccept(bookies ->
- writableBookiesWatchers.forEach(w -> executor.execute(() -> w.onBookiesChanged(bookies))));
- handleDeletedBookieNode(n);
- }
- } else if (n.getType() == NotificationType.Modified) {
- if (n.getPath().startsWith(bookieReadonlyRegistrationPath)
- || n.getPath().startsWith(bookieRegistrationPath)) {
- handleUpdatedBookieNode(n);
+ sequencer.sequential(() -> {
+ switch (n.getType()) {
+ case Created:
+ if (path.equals(bookieReadonlyRegistrationPath)) {
+ return getReadOnlyBookies().thenAccept(bookies ->
+ readOnlyBookiesWatchers.forEach(w ->
+ executor.execute(() -> w.onBookiesChanged(bookies))));
+ }
+ return getWritableBookies().thenAccept(bookies ->
+ writableBookiesWatchers.forEach(w ->
+ executor.execute(() -> w.onBookiesChanged(bookies))));
+ case Modified:
+ if (bookieId == null) {
+ return completedFuture(null);
+ }
+ if (path.equals(bookieReadonlyRegistrationPath)) {
+ return readBookieInfoAsReadonlyBookie(bookieId).thenApply(__ -> null);
+ }
+ return readBookieServiceInfoAsync(bookieId).thenApply(__ -> null);
Review Comment:
Is it also better to remove the cache first? since it has been updated.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] mattisonchao commented on a diff in pull request #20642: [fix][meta] Bookie Info lost by notification race condition.
Posted by "mattisonchao (via GitHub)" <gi...@apache.org>.
mattisonchao commented on code in PR #20642:
URL: https://github.com/apache/pulsar/pull/20642#discussion_r1243210715
##########
pulsar-metadata/src/main/java/org/apache/pulsar/metadata/bookkeeper/PulsarRegistrationClient.java:
##########
@@ -91,38 +96,58 @@ public void close() {
@Override
public CompletableFuture<Versioned<Set<BookieId>>> getWritableBookies() {
- return getChildren(bookieRegistrationPath);
+ return getBookiesThenFreshCache(bookieRegistrationPath);
}
@Override
public CompletableFuture<Versioned<Set<BookieId>>> getAllBookies() {
// this method is meant to return all the known bookies, even the bookies
// that are not in a running state
- return getChildren(bookieAllRegistrationPath);
+ return getBookiesThenFreshCache(bookieAllRegistrationPath);
}
@Override
public CompletableFuture<Versioned<Set<BookieId>>> getReadOnlyBookies() {
- return getChildren(bookieReadonlyRegistrationPath);
+ return getBookiesThenFreshCache(bookieReadonlyRegistrationPath);
}
- private CompletableFuture<Versioned<Set<BookieId>>> getChildren(String path) {
+ /**
+ * @throws IllegalArgumentException if parameter path is null or empty.
+ */
+ private CompletableFuture<Versioned<Set<BookieId>>> getBookiesThenFreshCache(String path) {
+ if (path == null || path.isEmpty()) {
+ return failedFuture(
+ new IllegalArgumentException(String.format("parameter [path] can not be null or empty.")));
+ }
return store.getChildren(path)
.thenComposeAsync(children -> {
- Set<BookieId> bookieIds = PulsarRegistrationClient.convertToBookieAddresses(children);
- List<CompletableFuture<?>> bookieInfoUpdated =
- new ArrayList<>(bookieIds.size());
+ final Set<BookieId> bookieIds = PulsarRegistrationClient.convertToBookieAddresses(children);
+ final List<CompletableFuture<?>> bookieInfoUpdated = new ArrayList<>(bookieIds.size());
for (BookieId id : bookieIds) {
// update the cache for new bookies
- if (!bookieServiceInfoCache.containsKey(id)) {
+ if (path.equals(bookieRegistrationPath) && writableBookieInfo.get(id) == null) {
bookieInfoUpdated.add(readBookieServiceInfoAsync(id));
+ continue;
+ }
+ if (path.equals(bookieReadonlyRegistrationPath) && readOnlyBookieInfo.get(id) == null) {
+ bookieInfoUpdated.add(readBookieInfoAsReadonlyBookie(id));
+ continue;
+ }
+ if (path.equals(bookieAllRegistrationPath)) {
+ if (writableBookieInfo.get(id) != null || readOnlyBookieInfo != null) {
Review Comment:
fixed.
##########
pulsar-metadata/src/main/java/org/apache/pulsar/metadata/bookkeeper/PulsarRegistrationClient.java:
##########
@@ -91,38 +96,58 @@ public void close() {
@Override
public CompletableFuture<Versioned<Set<BookieId>>> getWritableBookies() {
- return getChildren(bookieRegistrationPath);
+ return getBookiesThenFreshCache(bookieRegistrationPath);
}
@Override
public CompletableFuture<Versioned<Set<BookieId>>> getAllBookies() {
// this method is meant to return all the known bookies, even the bookies
// that are not in a running state
- return getChildren(bookieAllRegistrationPath);
+ return getBookiesThenFreshCache(bookieAllRegistrationPath);
}
@Override
public CompletableFuture<Versioned<Set<BookieId>>> getReadOnlyBookies() {
- return getChildren(bookieReadonlyRegistrationPath);
+ return getBookiesThenFreshCache(bookieReadonlyRegistrationPath);
}
- private CompletableFuture<Versioned<Set<BookieId>>> getChildren(String path) {
+ /**
+ * @throws IllegalArgumentException if parameter path is null or empty.
+ */
+ private CompletableFuture<Versioned<Set<BookieId>>> getBookiesThenFreshCache(String path) {
+ if (path == null || path.isEmpty()) {
+ return failedFuture(
+ new IllegalArgumentException(String.format("parameter [path] can not be null or empty.")));
+ }
return store.getChildren(path)
.thenComposeAsync(children -> {
- Set<BookieId> bookieIds = PulsarRegistrationClient.convertToBookieAddresses(children);
- List<CompletableFuture<?>> bookieInfoUpdated =
- new ArrayList<>(bookieIds.size());
+ final Set<BookieId> bookieIds = PulsarRegistrationClient.convertToBookieAddresses(children);
+ final List<CompletableFuture<?>> bookieInfoUpdated = new ArrayList<>(bookieIds.size());
for (BookieId id : bookieIds) {
// update the cache for new bookies
- if (!bookieServiceInfoCache.containsKey(id)) {
+ if (path.equals(bookieRegistrationPath) && writableBookieInfo.get(id) == null) {
bookieInfoUpdated.add(readBookieServiceInfoAsync(id));
+ continue;
+ }
+ if (path.equals(bookieReadonlyRegistrationPath) && readOnlyBookieInfo.get(id) == null) {
+ bookieInfoUpdated.add(readBookieInfoAsReadonlyBookie(id));
+ continue;
+ }
+ if (path.equals(bookieAllRegistrationPath)) {
+ if (writableBookieInfo.get(id) != null || readOnlyBookieInfo != null) {
+ // jump to next bookie id
+ continue;
+ }
+ bookieInfoUpdated.add(readBookieServiceInfoAsync(id) // check writable first
+ .thenCompose(updated ->
+ updated ? completedFuture(null)
+ : readBookieInfoAsReadonlyBookie(id))); // check read-only then
Review Comment:
fixed
##########
pulsar-metadata/src/main/java/org/apache/pulsar/metadata/bookkeeper/PulsarRegistrationClient.java:
##########
@@ -153,42 +178,57 @@ public void unwatchReadOnlyBookies(RegistrationListener registrationListener) {
readOnlyBookiesWatchers.remove(registrationListener);
}
- private void handleDeletedBookieNode(Notification n) {
- if (n.getType() == NotificationType.Deleted) {
- BookieId bookieId = stripBookieIdFromPath(n.getPath());
- if (bookieId != null) {
- log.info("Bookie {} disappeared", bookieId);
- bookieServiceInfoCache.remove(bookieId);
- }
+ /**
+ * This method will receive metadata store notifications and then update the
+ * local cache in background sequentially.
+ */
+ private void updatedBookies(Notification n) {
+ // make the notification callback run sequential in background.
+ final String path = n.getPath();
+ if (!path.startsWith(bookieReadonlyRegistrationPath) || !path.startsWith(bookieRegistrationPath)) {
Review Comment:
fixed
##########
pulsar-metadata/src/main/java/org/apache/pulsar/metadata/bookkeeper/PulsarRegistrationClient.java:
##########
@@ -153,42 +178,57 @@ public void unwatchReadOnlyBookies(RegistrationListener registrationListener) {
readOnlyBookiesWatchers.remove(registrationListener);
}
- private void handleDeletedBookieNode(Notification n) {
- if (n.getType() == NotificationType.Deleted) {
- BookieId bookieId = stripBookieIdFromPath(n.getPath());
- if (bookieId != null) {
- log.info("Bookie {} disappeared", bookieId);
- bookieServiceInfoCache.remove(bookieId);
- }
+ /**
+ * This method will receive metadata store notifications and then update the
+ * local cache in background sequentially.
+ */
+ private void updatedBookies(Notification n) {
+ // make the notification callback run sequential in background.
+ final String path = n.getPath();
+ if (!path.startsWith(bookieReadonlyRegistrationPath) || !path.startsWith(bookieRegistrationPath)) {
+ // ignore unknown path
+ return;
}
- }
-
- private void handleUpdatedBookieNode(Notification n) {
- BookieId bookieId = stripBookieIdFromPath(n.getPath());
+ final BookieId bookieId = stripBookieIdFromPath(n.getPath());
if (bookieId != null) {
- log.info("Bookie {} info updated", bookieId);
- readBookieServiceInfoAsync(bookieId);
+ log.info("Bookie {} do {}. path: {}", bookieId, n.getType(), n.getPath());
}
- }
-
- private void updatedBookies(Notification n) {
- if (n.getType() == NotificationType.Created || n.getType() == NotificationType.Deleted) {
- if (n.getPath().startsWith(bookieReadonlyRegistrationPath)) {
- getReadOnlyBookies().thenAccept(bookies -> {
- readOnlyBookiesWatchers.forEach(w -> executor.execute(() -> w.onBookiesChanged(bookies)));
- });
- handleDeletedBookieNode(n);
- } else if (n.getPath().startsWith(bookieRegistrationPath)) {
- getWritableBookies().thenAccept(bookies ->
- writableBookiesWatchers.forEach(w -> executor.execute(() -> w.onBookiesChanged(bookies))));
- handleDeletedBookieNode(n);
- }
- } else if (n.getType() == NotificationType.Modified) {
- if (n.getPath().startsWith(bookieReadonlyRegistrationPath)
- || n.getPath().startsWith(bookieRegistrationPath)) {
- handleUpdatedBookieNode(n);
+ sequencer.sequential(() -> {
+ switch (n.getType()) {
+ case Created:
+ if (path.equals(bookieReadonlyRegistrationPath)) {
Review Comment:
fixed
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] codecov-commenter commented on pull request #20642: [fix][meta] Bookie Info lost by notification race condition.
Posted by "codecov-commenter (via GitHub)" <gi...@apache.org>.
codecov-commenter commented on PR #20642:
URL: https://github.com/apache/pulsar/pull/20642#issuecomment-1613541142
## [Codecov](https://app.codecov.io/gh/apache/pulsar/pull/20642?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) Report
> Merging [#20642](https://app.codecov.io/gh/apache/pulsar/pull/20642?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) (40d326a) into [master](https://app.codecov.io/gh/apache/pulsar/commit/43b3622cc7e7f746ca9920fdc704dc7448767ac7?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) (43b3622) will **increase** coverage by `1.06%`.
> The diff coverage is `81.11%`.
[](https://app.codecov.io/gh/apache/pulsar/pull/20642?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)
```diff
@@ Coverage Diff @@
## master #20642 +/- ##
============================================
+ Coverage 72.05% 73.12% +1.06%
- Complexity 31718 32061 +343
============================================
Files 1855 1867 +12
Lines 138376 138782 +406
Branches 15198 15266 +68
============================================
+ Hits 99703 101479 +1776
+ Misses 30656 29262 -1394
- Partials 8017 8041 +24
```
| Flag | Coverage Δ | |
|---|---|---|
| inttests | `24.11% <37.77%> (+0.01%)` | :arrow_up: |
| systests | `24.97% <25.55%> (?)` | |
| unittests | `72.41% <81.11%> (+0.60%)` | :arrow_up: |
Flags with carried forward coverage won't be shown. [Click here](https://docs.codecov.io/docs/carryforward-flags?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#carryforward-flags-in-the-pull-request-comment) to find out more.
| [Impacted Files](https://app.codecov.io/gh/apache/pulsar/pull/20642?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Coverage Δ | |
|---|---|---|
| [.../metadata/bookkeeper/PulsarRegistrationClient.java](https://app.codecov.io/gh/apache/pulsar/pull/20642?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cHVsc2FyLW1ldGFkYXRhL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9wdWxzYXIvbWV0YWRhdGEvYm9va2tlZXBlci9QdWxzYXJSZWdpc3RyYXRpb25DbGllbnQuamF2YQ==) | `80.85% <81.11%> (-4.11%)` | :arrow_down: |
... and [243 files with indirect coverage changes](https://app.codecov.io/gh/apache/pulsar/pull/20642/indirect-changes?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] mattisonchao commented on a diff in pull request #20642: [fix][meta] Bookie Info lost by notification race condition.
Posted by "mattisonchao (via GitHub)" <gi...@apache.org>.
mattisonchao commented on code in PR #20642:
URL: https://github.com/apache/pulsar/pull/20642#discussion_r1243349627
##########
pulsar-metadata/src/main/java/org/apache/pulsar/metadata/bookkeeper/PulsarRegistrationClient.java:
##########
@@ -153,42 +178,57 @@ public void unwatchReadOnlyBookies(RegistrationListener registrationListener) {
readOnlyBookiesWatchers.remove(registrationListener);
}
- private void handleDeletedBookieNode(Notification n) {
- if (n.getType() == NotificationType.Deleted) {
- BookieId bookieId = stripBookieIdFromPath(n.getPath());
- if (bookieId != null) {
- log.info("Bookie {} disappeared", bookieId);
- bookieServiceInfoCache.remove(bookieId);
- }
+ /**
+ * This method will receive metadata store notifications and then update the
+ * local cache in background sequentially.
+ */
+ private void updatedBookies(Notification n) {
+ // make the notification callback run sequential in background.
+ final String path = n.getPath();
+ if (!path.startsWith(bookieReadonlyRegistrationPath) || !path.startsWith(bookieRegistrationPath)) {
+ // ignore unknown path
+ return;
}
- }
-
- private void handleUpdatedBookieNode(Notification n) {
- BookieId bookieId = stripBookieIdFromPath(n.getPath());
+ final BookieId bookieId = stripBookieIdFromPath(n.getPath());
if (bookieId != null) {
- log.info("Bookie {} info updated", bookieId);
- readBookieServiceInfoAsync(bookieId);
+ log.info("Bookie {} do {}. path: {}", bookieId, n.getType(), n.getPath());
}
- }
-
- private void updatedBookies(Notification n) {
- if (n.getType() == NotificationType.Created || n.getType() == NotificationType.Deleted) {
- if (n.getPath().startsWith(bookieReadonlyRegistrationPath)) {
- getReadOnlyBookies().thenAccept(bookies -> {
- readOnlyBookiesWatchers.forEach(w -> executor.execute(() -> w.onBookiesChanged(bookies)));
- });
- handleDeletedBookieNode(n);
- } else if (n.getPath().startsWith(bookieRegistrationPath)) {
- getWritableBookies().thenAccept(bookies ->
- writableBookiesWatchers.forEach(w -> executor.execute(() -> w.onBookiesChanged(bookies))));
- handleDeletedBookieNode(n);
- }
- } else if (n.getType() == NotificationType.Modified) {
- if (n.getPath().startsWith(bookieReadonlyRegistrationPath)
- || n.getPath().startsWith(bookieRegistrationPath)) {
- handleUpdatedBookieNode(n);
+ sequencer.sequential(() -> {
+ switch (n.getType()) {
+ case Created:
+ if (path.equals(bookieReadonlyRegistrationPath)) {
+ return getReadOnlyBookies().thenAccept(bookies ->
+ readOnlyBookiesWatchers.forEach(w ->
+ executor.execute(() -> w.onBookiesChanged(bookies))));
+ }
+ return getWritableBookies().thenAccept(bookies ->
+ writableBookiesWatchers.forEach(w ->
+ executor.execute(() -> w.onBookiesChanged(bookies))));
+ case Modified:
+ if (bookieId == null) {
+ return completedFuture(null);
+ }
+ if (path.equals(bookieReadonlyRegistrationPath)) {
+ return readBookieInfoAsReadonlyBookie(bookieId).thenApply(__ -> null);
+ }
+ return readBookieServiceInfoAsync(bookieId).thenApply(__ -> null);
Review Comment:
It is enough to ensure ultimate consistency. If it is deleted first, there is a gap in the cache between the deletion and the new one. :)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] codelipenghui commented on a diff in pull request #20642: [fix][meta] Bookie Info lost by notification race condition.
Posted by "codelipenghui (via GitHub)" <gi...@apache.org>.
codelipenghui commented on code in PR #20642:
URL: https://github.com/apache/pulsar/pull/20642#discussion_r1243459294
##########
pulsar-metadata/src/main/java/org/apache/pulsar/metadata/bookkeeper/PulsarRegistrationClient.java:
##########
@@ -227,46 +276,48 @@ public CompletableFuture<Versioned<BookieServiceInfo>> getBookieServiceInfo(Book
// this is because there are a few cases in which some operations on the main thread
// wait for the result. This is due to the fact that resolving the address of a bookie
// is needed in many code paths.
- Versioned<BookieServiceInfo> resultFromCache = bookieServiceInfoCache.get(bookieId);
+ Versioned<BookieServiceInfo> info;
+ if ((info = writableBookieInfo.get(bookieId)) == null) {
+ info = readOnlyBookieInfo.get(bookieId);
+ }
if (log.isDebugEnabled()) {
- log.debug("getBookieServiceInfo {} -> {}", bookieId, resultFromCache);
+ log.debug("getBookieServiceInfo {} -> {}", bookieId, info);
}
- if (resultFromCache != null) {
- return CompletableFuture.completedFuture(resultFromCache);
+ if (info != null) {
+ return completedFuture(info);
} else {
return FutureUtils.exception(new BKException.BKBookieHandleNotAvailableException());
}
}
- public CompletableFuture<Void> readBookieServiceInfoAsync(BookieId bookieId) {
- String asWritable = bookieRegistrationPath + "/" + bookieId;
- return bookieServiceInfoMetadataCache.get(asWritable)
- .thenCompose((Optional<BookieServiceInfo> getResult) -> {
- if (getResult.isPresent()) {
- Versioned<BookieServiceInfo> res =
- new Versioned<>(getResult.get(), new LongVersion(-1));
- log.info("Update BookieInfoCache (writable bookie) {} -> {}", bookieId, getResult.get());
- bookieServiceInfoCache.put(bookieId, res);
- return CompletableFuture.completedFuture(null);
- } else {
- return readBookieInfoAsReadonlyBookie(bookieId);
- }
- }
- );
+ public CompletableFuture<Optional<CacheGetResult<BookieServiceInfo>>> readBookieServiceInfoAsync(
Review Comment:
+1
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] mattisonchao closed pull request #20642: [fix][metadata] Bookie Info lost by notifier race condition.
Posted by "mattisonchao (via GitHub)" <gi...@apache.org>.
mattisonchao closed pull request #20642: [fix][metadata] Bookie Info lost by notifier race condition.
URL: https://github.com/apache/pulsar/pull/20642
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org