You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by te...@apache.org on 2023/07/06 08:08:53 UTC
[pulsar] branch master updated: [fix][schema] Only handle exception when there has (#20730)
This is an automated email from the ASF dual-hosted git repository.
technoboy pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/pulsar.git
The following commit(s) were added to refs/heads/master by this push:
new 881a1f41202 [fix][schema] Only handle exception when there has (#20730)
881a1f41202 is described below
commit 881a1f41202cffaf3b4fd6ab1235fac844d2dddb
Author: Yong Zhang <zh...@gmail.com>
AuthorDate: Thu Jul 6 16:08:45 2023 +0800
[fix][schema] Only handle exception when there has (#20730)
---
.../service/schema/BookkeeperSchemaStorage.java | 22 ++++++++++++----------
1 file changed, 12 insertions(+), 10 deletions(-)
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/schema/BookkeeperSchemaStorage.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/schema/BookkeeperSchemaStorage.java
index a8fc15f2965..78e30f6fff8 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/schema/BookkeeperSchemaStorage.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/schema/BookkeeperSchemaStorage.java
@@ -472,18 +472,20 @@ public class BookkeeperSchemaStorage implements SchemaStorage {
.build()
, locatorEntry.version
).thenApply(ignore -> nextVersion).whenComplete((__, ex) -> {
- Throwable cause = FutureUtil.unwrapCompletionException(ex);
- log.warn("[{}] Failed to update schema locator with position {}", schemaId, position, cause);
- if (cause instanceof AlreadyExistsException || cause instanceof BadVersionException) {
- bookKeeper.asyncDeleteLedger(position.getLedgerId(), new AsyncCallback.DeleteCallback() {
- @Override
- public void deleteComplete(int rc, Object ctx) {
- if (rc != BKException.Code.OK) {
- log.warn("[{}] Failed to delete ledger {} after updating schema locator failed, rc: {}",
+ if (ex != null) {
+ Throwable cause = FutureUtil.unwrapCompletionException(ex);
+ log.warn("[{}] Failed to update schema locator with position {}", schemaId, position, cause);
+ if (cause instanceof AlreadyExistsException || cause instanceof BadVersionException) {
+ bookKeeper.asyncDeleteLedger(position.getLedgerId(), new AsyncCallback.DeleteCallback() {
+ @Override
+ public void deleteComplete(int rc, Object ctx) {
+ if (rc != BKException.Code.OK) {
+ log.warn("[{}] Failed to delete ledger {} after updating schema locator failed, rc: {}",
schemaId, position.getLedgerId(), rc);
+ }
}
- }
- }, null);
+ }, null);
+ }
}
});
}