[DRAFT] [DO NOT REVIEW] Introduce CachedSupplier for BasePersistence objects

extension/persistence/relational-jdbc/src/main/java/org/apache/polaris/extension/persistence/relational/jdbc/JdbcMetaStoreManagerFactory.java

@@ -52,6 +52,7 @@
 import org.apache.polaris.core.persistence.dao.entity.PrincipalSecretsResult;
 import org.apache.polaris.core.storage.PolarisStorageIntegrationProvider;
 import org.apache.polaris.core.storage.cache.StorageCredentialCache;
+import org.apache.polaris.core.utils.CachedSupplier;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -97,12 +98,16 @@ private void initializeForRealm(
     DatasourceOperations databaseOperations = getDatasourceOperations(isBootstrap);
     sessionSupplierMap.put(
         realmContext.getRealmIdentifier(),
-        () ->
-            new JdbcBasePersistenceImpl(
-                databaseOperations,
-                secretsGenerator(realmContext, rootCredentialsSet),
-                storageIntegrationProvider,
-                realmContext.getRealmIdentifier()));
+        new CachedSupplier<>(
+            () ->
+                new JdbcBasePersistenceImpl(
+                    databaseOperations,
+                    secretsGenerator(realmContext, rootCredentialsSet),
+                    storageIntegrationProvider,
+                    realmContext.getRealmIdentifier())));
+
+    // Ensure the supplier caches the first time
+    var unused = sessionSupplierMap.get(realmContext.getRealmIdentifier()).get();
 
     PolarisMetaStoreManager metaStoreManager = createNewMetaStoreManager();
     metaStoreManagerMap.put(realmContext.getRealmIdentifier(), metaStoreManager);

polaris-core/src/main/java/org/apache/polaris/core/persistence/LocalPolarisMetaStoreManagerFactory.java

@@ -41,6 +41,7 @@
 import org.apache.polaris.core.persistence.transactional.TransactionalMetaStoreManagerImpl;
 import org.apache.polaris.core.persistence.transactional.TransactionalPersistence;
 import org.apache.polaris.core.storage.cache.StorageCredentialCache;
+import org.apache.polaris.core.utils.CachedSupplier;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -100,7 +101,12 @@ private void initializeForRealm(
     final StoreType backingStore = createBackingStore(diagnostics);
     sessionSupplierMap.put(
         realmContext.getRealmIdentifier(),
-        () -> createMetaStoreSession(backingStore, realmContext, rootCredentialsSet, diagnostics));
+        new CachedSupplier<>(
+            () ->
+                createMetaStoreSession(
+                    backingStore, realmContext, rootCredentialsSet, diagnostics)));
+    // Ensure the supplier caches the first time
+    var unused = sessionSupplierMap.get(realmContext.getRealmIdentifier()).get();
 
     PolarisMetaStoreManager metaStoreManager = createNewMetaStoreManager();
     metaStoreManagerMap.put(realmContext.getRealmIdentifier(), metaStoreManager);

polaris-core/src/main/java/org/apache/polaris/core/utils/CachedSupplier.java

@@ -0,0 +1,41 @@
+/*
+ * 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.polaris.core.utils;
+
+import java.util.function.Supplier;
+
+public class CachedSupplier<T> implements Supplier<T> {
+  private final Supplier<T> delegate;
+  private T value;
+  private boolean initialized = false;
+
+  public CachedSupplier(Supplier<T> delegate) {
+    this.delegate = delegate;
+  }
+
+  @Override
+  public synchronized T get() {
+    if (!initialized) {
+      value = delegate.get();
+      initialized = true;
+    }
+    return value;
+  }
+}

polaris-core/src/test/java/org/apache/polaris/core/utils/CachedSupplierTest.java

@@ -0,0 +1,70 @@
+/*
+ * 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.polaris.core.utils;
+
+import java.util.function.Supplier;
+import org.apache.polaris.core.context.RealmContext;
+import org.assertj.core.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+public class CachedSupplierTest {
+  private String realmName = "test";
+  private int timesCalled = 0;
+  private final RealmContext realmContext =
+      new RealmContext() {
+        @Override
+        public String getRealmIdentifier() {
+          if (++timesCalled == 1) {
+            return realmName;
+          }
+          throw new IllegalStateException();
+        }
+      };
+
+  private static class ContainerRealmIdentifier {
+    private String realmIdentifier;
+
+    public ContainerRealmIdentifier(RealmContext realmContext) {
+      this.realmIdentifier = realmContext.getRealmIdentifier();
+    }
+
+    public String getRealmIdentifier() {
+      return realmIdentifier;
+    }
+  }
+
+  @Test
+  public void testCachedSupplier() {
+    Supplier<ContainerRealmIdentifier> realmIdentifierSupplier =
+        () -> new ContainerRealmIdentifier(realmContext);
+    Assertions.assertThat(realmName.equals(realmIdentifierSupplier.get().getRealmIdentifier()))
+        .isTrue(); // This will work
+    Assertions.assertThatThrownBy(() -> realmIdentifierSupplier.get().getRealmIdentifier())
+        .isInstanceOf(IllegalStateException.class);
+
+    timesCalled = 0;
+    CachedSupplier<ContainerRealmIdentifier> cachedSupplier =
+        new CachedSupplier<>(() -> new ContainerRealmIdentifier(realmContext));
+    Assertions.assertThat(realmName.equals(cachedSupplier.get().getRealmIdentifier()))
+        .isTrue(); // This will work
+    Assertions.assertThat(realmName.equals(cachedSupplier.get().getRealmIdentifier()))
+        .isTrue(); // This will work
+  }
+}