[pulsar] branch master updated: [fix][schema] Only handle exception when there has (#20730)

[technoboy]

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 <zhangyong1025...@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);
+                }
             }
         });
     }