[fix][meta] Metadata cache refresh might not take effect (#25246)

(cherry picked from commit 24eba10)

Author: BewareMyPower

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

@@ -153,13 +153,6 @@ private CompletableFuture<Optional<CacheGetResult<T>>> readValueFromStore(String
                 return FutureUtils.value(Optional.<CacheGetResult<T>>empty());
             }
             final var res = optRes.get();
-            final var cachedFuture = objCache.getIfPresent(path);
-            if (cachedFuture != null && cachedFuture != future) {
-                if (log.isDebugEnabled()) {
-                    log.debug("A new read on key {} is in progress or completed, ignore this one", path);
-                }
-                return cachedFuture;
-            }
             try {
                 T obj = serde.deserialize(path, res.getValue(), res.getStat());
                 if (log.isDebugEnabled()) {

pulsar-metadata/src/test/java/org/apache/pulsar/metadata/MetadataCacheTest.java

@@ -33,6 +33,7 @@
 import java.io.IOException;
 import java.lang.reflect.Field;
 import java.nio.charset.StandardCharsets;
+import java.time.Duration;
 import java.util.ArrayList;
 import java.util.EnumSet;
 import java.util.List;
@@ -69,6 +70,7 @@
 import org.apache.pulsar.metadata.api.Stat;
 import org.apache.pulsar.metadata.api.extended.CreateOption;
 import org.apache.pulsar.metadata.cache.impl.MetadataCacheImpl;
+import org.apache.pulsar.metadata.impl.LocalMemoryMetadataStore;
 import org.awaitility.Awaitility;
 import org.mockito.stubbing.Answer;
 import org.testng.annotations.Test;
@@ -739,4 +741,18 @@ public void testNoBackoffMetadataCacheConfig() {
         assertTrue(backoff.isMandatoryStopMade());
         assertEquals(backoff.getFirstBackoffTimeInMillis(), 0);
     }
+
+    @Test
+    public void testRefreshRace() throws Exception {
+        @Cleanup final var store = new LocalMemoryMetadataStore("memory:local", MetadataStoreConfig.builder().build());
+        final var cache = store.getMetadataCache(String.class);
+        for (int i = 0; i < 500; i++) {
+            final var key = "/key" + i;
+            assertTrue(cache.get(key).get().isEmpty());
+
+            store.put(key, "\"value\"".getBytes(StandardCharsets.UTF_8), Optional.empty()).get();
+            Awaitility.await().pollInterval(Duration.ofMillis(1)).atMost(Duration.ofSeconds(3)).untilAsserted(() ->
+                assertTrue(cache.get(key).get().isPresent(), "Failed at key " + key));
+        }
+    }
 }