[fix][broker] Handle empty LocalPolicies content gracefully (#25158)

Author: zooo-code

pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/AdminResource.java

@@ -333,7 +333,12 @@ protected CompletableFuture<Policies> getNamespacePoliciesAsync(NamespaceName na
                             .getBundlesAsync(namespaceName)
                             .thenApply(bundles -> bundles.getBundlesData());
                     CompletableFuture<Optional<LocalPolicies>> localPoliciesFuture =
-                            getLocalPolicies().getLocalPoliciesAsync(namespaceName);
+                            getLocalPolicies().getLocalPoliciesAsync(namespaceName)
+                                    .exceptionally(ex -> {
+                                        log.warn("Failed to load local policies for {}, using empty: {}",
+                                                namespaceName, ex.getMessage());
+                                        return Optional.empty();
+                                    });
 
                     return bundlesFuture.thenCombine(localPoliciesFuture, (bundleData, localPolicies) -> {
                         policies.bundles = bundleData != null ? bundleData : policies.bundles;
@@ -839,6 +844,11 @@ protected CompletableFuture<SchemaCompatibilityStrategy> getSchemaCompatibilityS
                     }
                 }
                 return schemaCompatibilityStrategy;
+            }).exceptionally(ex -> {
+                // If namespace policies cannot be loaded, fall back to broker config
+                log.warn("Failed to load namespace policies for {}, using broker config default: {}",
+                        namespaceName, ex.getMessage());
+                return pulsar().getConfig().getSchemaCompatibilityStrategy();
             });
         });
     }

pulsar-broker/src/main/java/org/apache/pulsar/common/naming/NamespaceBundleFactory.java

@@ -177,7 +177,12 @@ private CompletableFuture<NamespaceBundles> copyToLocalPolicies(NamespaceName na
                     return pulsar.getPulsarResources().getLocalPolicies()
                             .createLocalPoliciesAsync(namespace, localPolicies)
                             .thenApply(stat -> getBundles(namespace,
-                                    Optional.of(Pair.of(localPolicies, 0L))));
+                                    Optional.of(Pair.of(localPolicies, 0L))))
+                            .exceptionally(ex -> {
+                                // If LocalPolicies already exists (possibly with empty content),
+                                // just use the bundles from global policies
+                                return getBundles(namespace, Optional.empty());
+                            });
                 });
     }
 

pulsar-metadata/src/main/java/org/apache/pulsar/metadata/cache/impl/MetadataCacheImpl.java

@@ -136,7 +136,15 @@ private CompletableFuture<Optional<CacheGetResult<T>>> readValueFromStore(String
 
                     try {
                         GetResult res = optRes.get();
-                        T obj = serde.deserialize(path, res.getValue(), res.getStat());
+                        byte[] value = res.getValue();
+
+                        // Handle empty content before attempting deserialization
+                        if (value == null || value.length == 0) {
+                            log.warn("Empty content found for key '{}', treating as non-existent", path);
+                            return FutureUtils.value(Optional.empty());
+                        }
+
+                        T obj = serde.deserialize(path, value, res.getStat());
                         return FutureUtils
                                 .value(Optional.of(new CacheGetResult<>(obj, res.getStat())));
                     } catch (Throwable t) {