From b19f719c6a9f7f397c78714cda6acaf0cf766b94 Mon Sep 17 00:00:00 2001
From: Jiang Haiting <jianghaiting@didichuxing.com>
Date: Sat, 6 Nov 2021 17:20:42 +0800
Subject: [PATCH] revert the wrong modification in
 org.apache.pulsar.broker.namespace.OwnershipCache#checkOwnership

---
 .../apache/pulsar/broker/namespace/OwnershipCache.java    | 8 +++++++-
 1 file changed, 7 insertions(+), 1 deletion(-)

diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/namespace/OwnershipCache.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/namespace/OwnershipCache.java
index c27575c81a842..6cf73593137e1 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/namespace/OwnershipCache.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/namespace/OwnershipCache.java
@@ -210,7 +210,13 @@ public CompletableFuture<Boolean> checkOwnership(NamespaceBundle bundle) {
             return CompletableFuture.completedFuture(true);
         }
         String bundlePath = ServiceUnitUtils.path(bundle);
-        return resolveOwnership(bundlePath).thenApply(Optional::isPresent);
+        return resolveOwnership(bundlePath).thenApply(optionalOwnedDataWithStat -> {
+            if (!optionalOwnedDataWithStat.isPresent()) {
+                return false;
+            }
+            Stat stat = optionalOwnedDataWithStat.get().getValue();
+            return stat.getEphemeralOwner() == localZkCache.getZooKeeper().getSessionId();
+        });
     }
 
     /**