This is an automated email from the ASF dual-hosted git repository.
bharos pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/gravitino.git
The following commit(s) were added to refs/heads/main by this push:
new 5af40fd4d9 [#13573] fix: slow metadata reads caused by NoOpsCache lock
contention (#13574)
5af40fd4d9 is described below
commit 5af40fd4d9f0a7672cdc4443a4786544b6741052
Author: Abhijeet Singh <[email protected]>
AuthorDate: Tue Oct 6 09:38:17 2026 +0530
[#13573] fix: slow metadata reads caused by NoOpsCache lock contention
(#13574)
<!--
1. Title: [#<issue>] <type>(<scope>): <subject>
Examples:
- "[#123] feat(operator): Support xxx"
- "[#233] fix: Check null before access result in xxx"
- "[MINOR] refactor: Fix typo in variable name"
- "[MINOR] docs: Fix typo in README"
- "[#255] test: Fix flaky test NameOfTheTest"
Reference: https://www.conventionalcommits.org/en/v1.0.0/
2. If the PR is unfinished, please mark this PR as draft.
-->
### What changes were proposed in this pull request?
Remove the single shared lock from `NoOpsCache.withCacheLock()`. When
entity caching is disabled, both overloads now validate their arguments
and run the supplied action directly. Add tests for concurrent reads,
result and exception propagation, and null arguments.
### Why are the changes needed?
When entity caching is disabled (gravitino.cache.enabled=false),
Gravitino uses NoOpsCache, which acquires a single shared lock for all
entity reads. The lock remains held during database access, forcing
unrelated reads to execute sequentially. Under concurrent load, this
slows down metadata reads and increases request latency.
Since NoOpsCache stores no cached state, this locking is unnecessary.
Reads should execute
concurrently when caching is disabled.
Fix: #13573
### Does this PR introduce any user-facing change?
With entity caching disabled, unrelated metadata reads can run
concurrently. The default cache-enabled path is unchanged. There are no
API or configuration changes.
### How was this patch tested?
unit tests passed, covering concurrent reads, result and exception
propagation, and null arguments.
```
./gradlew :core:test --tests org.apache.gravitino.cache.TestNoOpsCache
```
---------
Co-authored-by: Abhijeet Singh <[email protected]>
---
.../org/apache/gravitino/cache/NoOpsCache.java | 14 ++-
.../org/apache/gravitino/cache/TestNoOpsCache.java | 132 +++++++++++++++++++++
2 files changed, 141 insertions(+), 5 deletions(-)
diff --git a/core/src/main/java/org/apache/gravitino/cache/NoOpsCache.java
b/core/src/main/java/org/apache/gravitino/cache/NoOpsCache.java
index c581f6f51f..0f839737be 100644
--- a/core/src/main/java/org/apache/gravitino/cache/NoOpsCache.java
+++ b/core/src/main/java/org/apache/gravitino/cache/NoOpsCache.java
@@ -19,6 +19,7 @@
package org.apache.gravitino.cache;
+import com.google.common.base.Preconditions;
import java.util.Optional;
import org.apache.gravitino.Config;
import org.apache.gravitino.Entity;
@@ -27,7 +28,6 @@ import org.apache.gravitino.NameIdentifier;
/** A cache implementation that does not cache anything. */
public class NoOpsCache extends BaseEntityCache {
- private final SegmentedLock opLock = new SegmentedLock(1);
/**
* Constructs a new {@link BaseEntityCache} instance.
*
@@ -61,18 +61,22 @@ public class NoOpsCache extends BaseEntityCache {
return 0;
}
- /** {@inheritDoc} */
+ /** {@inheritDoc} Runs the action directly because this cache holds no state
to lock. */
@Override
public <E extends Exception> void withCacheLock(EntityCacheKey key,
ThrowingRunnable<E> action)
throws E {
- opLock.withLockAndThrow(key, action);
+ Preconditions.checkArgument(key != null, "Key cannot be null");
+ Preconditions.checkArgument(action != null, "Action cannot be null");
+ action.run();
}
- /** {@inheritDoc} */
+ /** {@inheritDoc} Runs the action directly because this cache holds no state
to lock. */
@Override
public <T, E extends Exception> T withCacheLock(EntityCacheKey key,
ThrowingSupplier<T, E> action)
throws E {
- return opLock.withLockAndThrow(key, action);
+ Preconditions.checkArgument(key != null, "Key cannot be null");
+ Preconditions.checkArgument(action != null, "Action cannot be null");
+ return action.get();
}
/** {@inheritDoc} */
diff --git a/core/src/test/java/org/apache/gravitino/cache/TestNoOpsCache.java
b/core/src/test/java/org/apache/gravitino/cache/TestNoOpsCache.java
new file mode 100644
index 0000000000..21fc3c26a7
--- /dev/null
+++ b/core/src/test/java/org/apache/gravitino/cache/TestNoOpsCache.java
@@ -0,0 +1,132 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.gravitino.cache;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+import org.apache.gravitino.Config;
+import org.apache.gravitino.Entity;
+import org.apache.gravitino.NameIdentifier;
+import org.junit.jupiter.api.Test;
+
+/** Tests for when entity caching is disabled. */
+public class TestNoOpsCache {
+
+ @Test
+ void testWithCacheLockDoesNotSerializeUnrelatedReads() throws Exception {
+ NoOpsCache cache = new NoOpsCache(new Config(false) {});
+ EntityCacheKey userKey =
+ EntityCacheKey.of(NameIdentifier.of("metalake", "user-a"),
Entity.EntityType.USER);
+ EntityCacheKey tableKey =
+ EntityCacheKey.of(
+ NameIdentifier.of("metalake", "catalog", "schema", "table-b"),
Entity.EntityType.TABLE);
+
+ CountDownLatch entered = new CountDownLatch(2);
+ CountDownLatch release = new CountDownLatch(1);
+ AtomicInteger inFlight = new AtomicInteger();
+ AtomicInteger maxInFlight = new AtomicInteger();
+
+ ExecutorService executor = Executors.newFixedThreadPool(2);
+ try {
+ Future<String> userRead =
+ executor.submit(
+ () ->
+ cache.withCacheLock(
+ userKey,
+ () -> holdAndReturn(entered, release, inFlight,
maxInFlight, "user")));
+ Future<String> tableRead =
+ executor.submit(
+ () ->
+ cache.withCacheLock(
+ tableKey,
+ () -> holdAndReturn(entered, release, inFlight,
maxInFlight, "table")));
+
+ assertTrue(
+ entered.await(2, TimeUnit.SECONDS),
+ "Cache-disabled reads of unrelated entities must overlap. A single
lock serialized them.");
+ release.countDown();
+ assertEquals("user", userRead.get(2, TimeUnit.SECONDS));
+ assertEquals("table", tableRead.get(2, TimeUnit.SECONDS));
+ assertEquals(2, maxInFlight.get());
+ } finally {
+ executor.shutdownNow();
+ }
+ }
+
+ @Test
+ void testWithCacheLockPropagatesActionResultAndException() throws Exception {
+ NoOpsCache cache = new NoOpsCache(new Config(false) {});
+ EntityCacheKey key =
+ EntityCacheKey.of(NameIdentifier.of("metalake", "user"),
Entity.EntityType.USER);
+
+ assertEquals("loaded", cache.withCacheLock(key, () -> "loaded"));
+ IllegalStateException failure =
+ assertThrows(IllegalStateException.class, () ->
cache.withCacheLock(key, this::fail));
+ assertEquals("backend failed", failure.getMessage());
+ }
+
+ @Test
+ void testWithCacheLockRejectsNullArguments() throws Exception {
+ NoOpsCache cache = new NoOpsCache(new Config(false) {});
+ EntityCacheKey key =
+ EntityCacheKey.of(NameIdentifier.of("metalake", "user"),
Entity.EntityType.USER);
+
+ assertThrows(IllegalArgumentException.class, () ->
cache.withCacheLock(null, () -> {}));
+ assertThrows(
+ IllegalArgumentException.class,
+ () -> cache.withCacheLock(key,
(EntityCache.ThrowingRunnable<Exception>) null));
+ assertThrows(IllegalArgumentException.class, () ->
cache.withCacheLock(null, () -> "loaded"));
+ assertThrows(
+ IllegalArgumentException.class,
+ () -> cache.withCacheLock(key, (EntityCache.ThrowingSupplier<String,
Exception>) null));
+ }
+
+ private void fail() {
+ throw new IllegalStateException("backend failed");
+ }
+
+ private static String holdAndReturn(
+ CountDownLatch entered,
+ CountDownLatch release,
+ AtomicInteger inFlight,
+ AtomicInteger maxInFlight,
+ String value)
+ throws InterruptedException {
+ int current = inFlight.incrementAndGet();
+ maxInFlight.accumulateAndGet(current, Math::max);
+ entered.countDown();
+ try {
+ if (!release.await(2, TimeUnit.SECONDS)) {
+ throw new IllegalStateException("Timed out waiting for the other read
to enter");
+ }
+ return value;
+ } finally {
+ inFlight.decrementAndGet();
+ }
+ }
+}
