Copilot commented on code in PR #13166:
URL: https://github.com/apache/gravitino/pull/13166#discussion_r4012280228
##########
spark-connector/spark-common/src/main/java/org/apache/gravitino/spark/connector/catalog/GravitinoCatalogManager.java:
##########
@@ -259,4 +264,27 @@ private static void closeClient(GravitinoIdentity
identity, GravitinoClient clie
LOG.warn("Failed to close the Gravitino client of {}.", identity, e);
}
}
+
+ /**
+ * Holds a cached client so that the shutdown drain and the removal listener
can both try to close
+ * it while the underlying client is closed at most once.
+ */
+ private static class CachedClient {
+ private final GravitinoClient client;
+ private final AtomicBoolean closed = new AtomicBoolean(false);
+
+ private CachedClient(GravitinoClient client) {
+ this.client = client;
+ }
+
+ private GravitinoClient client() {
+ return client;
+ }
+
+ private void close() {
+ if (closed.compareAndSet(false, true)) {
+ client.close();
+ }
+ }
+ }
}
Review Comment:
The CAS is set before invoking `client.close()`. If a removal listener wins
the CAS while `close()`'s `asMap().forEach` is holding the same entry, the
drain returns immediately on `closed == true` without waiting for the
listener's close to finish, so `GravitinoCatalogManager.close()` can return
while the underlying client is still closing. Serialize this method (or
otherwise await completion) so the synchronous shutdown path cannot lose the
close and return early.
##########
docs/review-6c2878fb1.md:
##########
@@ -0,0 +1,51 @@
+# Review：6c2878fb1 — 跳过 EXPLICIT 修掉了双关，但换来一个漏关窗口
+
+## 一、改动概述
+
+`GravitinoCatalogManager` 的 client cache 原来在 removal listener 里无条件
`closeClient`，而 `close()` 已经先同步遍历关过一遍，于是每个 client 被关两次，第二次由 Caffeine 派发到
`ForkJoinPool.commonPool()`、在 `close()` 返回之后才跑。
+
+改动分两部分：生产代码让 removal listener 跳过 `RemovalCause.EXPLICIT`（只有 `close()` 会产生这个
cause）；测试把 `ClientFactory` 的 per-client `AtomicBoolean` 换成 `AtomicInteger`
计数，`testCloseClosesEveryCachedClient` 在 `close()` 之后 drain common pool，再断言每个
client 只关一次。
Review Comment:
This added document describes the rejected `RemovalCause.EXPLICIT`-skipping
implementation, but the production diff now closes on every removal and uses
`CachedClient`'s CAS for idempotence. Keeping this text under `docs/` leaves an
inaccurate description of the current behavior; update the whole report to the
final implementation or remove this review artifact.
##########
spark-connector/spark-common/src/test/java/org/apache/gravitino/spark/connector/catalog/TestGravitinoCatalogManager.java:
##########
@@ -145,7 +147,18 @@ void testCloseClosesEveryCachedClient() {
manager.close();
- assertEquals(3, clientFactory.closedCount());
+ // Exact already here: no client the shutdown drain saw may outlive
close().
+ assertEquals(
+ List.of(1, 1, 1),
+ clientFactory.closeCounts(),
+ "Shutdown must close each cached client on the calling thread");
+ // Caffeine dispatches removal listeners on the common pool, so a second
close would land after
+ // close() returned. Draining the pool makes that visible instead of
leaving it to timing.
+ ForkJoinPool.commonPool().awaitQuiescence(30, TimeUnit.SECONDS);
Review Comment:
The return value of `awaitQuiescence` is ignored. If the shared common pool
remains busy for 30 seconds, this returns `false` while a removal-listener task
can still be pending, and the following assertion may pass before a second
close is delivered. Use a per-test completion signal or otherwise verify that
the removal task has completed before asserting the final counts.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
