[fix][meta] Complete handleMetadataEvent future exceptionally when the initial get fails

pulsar-metadata/src/main/java/org/apache/pulsar/metadata/impl/AbstractMetadataStore.java

@@ -235,14 +235,14 @@ protected long getChildrenCacheMaxSizeBytes() {
     @Override
     public CompletableFuture<Void> handleMetadataEvent(MetadataEvent event) {
         CompletableFuture<Void> result = new CompletableFuture<>();
-        get(event.getPath()).thenApply(res -> {
+        get(event.getPath()).thenAccept(res -> {
             Set<CreateOption> options = event.getOptions() != null ? event.getOptions()
                     : Collections.emptySet();
             if (res.isPresent()) {
                 GetResult existingValue = res.get();
                 if (shouldIgnoreEvent(event, existingValue)) {
                     result.complete(null);
-                    return result;
+                    return;
                 }
             }
             // else update the event
@@ -262,7 +262,11 @@ public CompletableFuture<Void> handleMetadataEvent(MetadataEvent event) {
                 }
                 return false;
             });
-            return result;
+        }).exceptionally(ex -> {
+            Throwable cause = FutureUtil.unwrapCompletionException(ex);
+            log.warn().attr("path", event.getPath()).exception(cause).log("Failed to handle metadata event");
+            result.completeExceptionally(cause);
+            return null;
         });
         return result;
     }

pulsar-metadata/src/test/java/org/apache/pulsar/metadata/impl/MetadataEventSynchronizerTest.java

@@ -20,13 +20,24 @@
 
 import static org.testng.Assert.assertFalse;
 import static org.testng.Assert.assertTrue;
+import static org.testng.Assert.expectThrows;
 import java.nio.charset.StandardCharsets;
+import java.util.HashSet;
 import java.util.Optional;
+import java.util.Set;
 import java.util.UUID;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.TimeUnit;
 import lombok.Cleanup;
+import org.apache.pulsar.metadata.api.GetResult;
+import org.apache.pulsar.metadata.api.MetadataEvent;
 import org.apache.pulsar.metadata.api.MetadataStore;
 import org.apache.pulsar.metadata.api.MetadataStoreConfig;
+import org.apache.pulsar.metadata.api.MetadataStoreException;
 import org.apache.pulsar.metadata.api.MetadataStoreFactory;
+import org.apache.pulsar.metadata.api.NotificationType;
+import org.apache.pulsar.metadata.api.Option;
 import org.awaitility.Awaitility;
 import org.testng.annotations.Test;
 
@@ -75,6 +86,28 @@ public void testSharedInstance() throws Exception {
         });
     }
 
+    @Test
+    public void testHandleMetadataEventCompletesWhenGetFails() throws Exception {
+        @Cleanup
+        LocalMemoryMetadataStore store = new LocalMemoryMetadataStore("memory:local",
+                MetadataStoreConfig.builder().build()) {
+            @Override
+            public CompletableFuture<Optional<GetResult>> storeGet(String path, Set<Option> opts) {
+                return CompletableFuture.failedFuture(
+                        new MetadataStoreException("injected storeGet failure"));
+            }
+        };
+
+        MetadataEvent event = new MetadataEvent("/test", "value".getBytes(StandardCharsets.UTF_8),
+                new HashSet<>(), null, System.currentTimeMillis(), "test-cluster", NotificationType.Modified);
+
+        CompletableFuture<Void> result = store.handleMetadataEvent(event);
+        // The future must not hang when the initial get() fails: it should complete exceptionally
+        ExecutionException ex = expectThrows(ExecutionException.class, () -> result.get(5, TimeUnit.SECONDS));
+        assertTrue(ex.getCause() instanceof MetadataStoreException,
+                "expected MetadataStoreException cause but got: " + ex.getCause());
+    }
+
     @Test
     public void testPathValid() {
         assertFalse(AbstractMetadataStore.isValidPath(null));