Handle RequestScoped instance injection gracefully for DefaultConfigurationStore (#1758)

Although we do a check of !realmContextInstance.isUnsatisfied() it only checks if there is matching bean, however, it doesn't check if the context is active or not. Similarly isResolvable also don't check if the scope is active. Therefore if getConfiguration is called when there is no active scope, it will get Exception like ContextNotActiveException.
This actually fails the TaskExecutor because it runs in a separate thread.

In order to fix the problem, we introduces a new getConfiguration function to handle the background tasks, and also use isResolvable instead of isUnsatisfied to handle ambiguities.

Author: gh-yzou

polaris-core/src/main/java/org/apache/polaris/core/config/PolarisConfigurationStore.java

@@ -25,6 +25,7 @@
 import java.util.List;
 import java.util.Map;
 import org.apache.polaris.core.PolarisCallContext;
+import org.apache.polaris.core.context.RealmContext;
 import org.apache.polaris.core.entity.CatalogEntity;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -39,11 +40,18 @@ public interface PolarisConfigurationStore {
   /**
    * Retrieve the current value for a configuration key. May be null if not set.
    *
+   * <p>This function will be deprecated, it can not be called outside of active request scope, such
+   * as background tasks (TaskExecutor). Please use the function getConfiguration(RealmContext
+   * realmContext, String configName) to get the configuration value in a more robust way. TODO:
+   * Remove this function and replace the usage with the function takes realm. Github issue
+   * https://github.com/apache/polaris/issues/1775
+   *
    * @param ctx the current call context
    * @param configName the name of the configuration key to check
    * @return the current value set for the configuration key or null if not set
    * @param <T> the type of the configuration value
    */
+  @Deprecated
   default <T> @Nullable T getConfiguration(PolarisCallContext ctx, String configName) {
     return null;
   }
@@ -52,19 +60,54 @@ public interface PolarisConfigurationStore {
    * Retrieve the current value for a configuration key. If not set, return the non-null default
    * value.
    *
+   * <p>This function will be deprecated, it can not be called outside of active request scope, such
+   * as background tasks (TaskExecutor). Please use the function getConfiguration(RealmContext
+   * realmContext, String configName, T defaultValue) to get the configuration value in a more
+   * robust way. TODO: Remove this function and replace the usage with the function takes realm.
+   * Github issue https://github.com/apache/polaris/issues/1775
+   *
    * @param ctx the current call context
    * @param configName the name of the configuration key to check
    * @param defaultValue the default value if the configuration key has no value
    * @return the current value or the supplied default value
-   * @param <T> the type of the configuration value
    */
+  @Deprecated
   default <T> @Nonnull T getConfiguration(
       PolarisCallContext ctx, String configName, @Nonnull T defaultValue) {
     Preconditions.checkNotNull(defaultValue, "Cannot pass null as a default value");
     T configValue = getConfiguration(ctx, configName);
     return configValue != null ? configValue : defaultValue;
   }
 
+  /**
+   * Retrieve the current value for a configuration key for a given realm. May be null if not set.
+   *
+   * @param realmContext the realm context
+   * @param configName the name of the configuration key to check
+   * @return the current value set for the configuration key for the given realm, or null if not set
+   * @param <T> the type of the configuration value
+   */
+  default <T> @Nullable T getConfiguration(@Nonnull RealmContext realmContext, String configName) {
+    return null;
+  }
+
+  /**
+   * Retrieve the current value for a configuration key for the given realm. If not set, return the
+   * non-null default value.
+   *
+   * @param realmContext the realm context
+   * @param configName the name of the configuration key to check
+   * @param defaultValue the default value if the configuration key has no value
+   * @return the current value or the supplied default value
+   * @param <T> the type of the configuration value
+   */
+  default <T> @Nonnull T getConfiguration(
+      RealmContext realmContext, String configName, @Nonnull T defaultValue) {
+    Preconditions.checkNotNull(defaultValue, "Cannot pass null as a default value");
+    T configValue = getConfiguration(realmContext, configName);
+    return configValue != null ? configValue : defaultValue;
+  }
+
   /**
    * In some cases, we may extract a value that doesn't match the expected type for a config. This
    * method can be used to attempt to force-cast it using `String.valueOf`
@@ -90,7 +133,8 @@ public interface PolarisConfigurationStore {
   }
 
   /**
-   * Retrieve the current value for a configuration.
+   * Retrieve the current value for a configuration. TODO: update the function to take RealmContext
+   * instead of PolarisCallContext. Github issue https://github.com/apache/polaris/issues/1775
    *
    * @param ctx the current call context
    * @param config the configuration to load
@@ -104,7 +148,8 @@ public interface PolarisConfigurationStore {
 
   /**
    * Retrieve the current value for a configuration, overriding with a catalog config if it is
-   * present.
+   * present. TODO: update the function to take RealmContext instead of PolarisCallContext Github
+   * issue https://github.com/apache/polaris/issues/1775
    *
    * @param ctx the current call context
    * @param catalogEntity the catalog to check for an override

quarkus/service/src/test/java/org/apache/polaris/service/quarkus/config/DefaultConfigurationStoreTest.java

@@ -74,6 +74,7 @@ public Map<String, String> getConfigOverrides() {
   }
 
   private PolarisCallContext polarisContext;
+  private RealmContext realmContext;
 
   @Inject MetaStoreManagerFactory managerFactory;
   @Inject PolarisConfigurationStore configurationStore;
@@ -87,8 +88,7 @@ public void before(TestInfo testInfo) {
             .formatted(
                 testInfo.getTestMethod().map(java.lang.reflect.Method::getName).orElse("test"),
                 System.nanoTime());
-    RealmContext realmContext = () -> realmName;
-    QuarkusMock.installMockForType(realmContext, RealmContext.class);
+    realmContext = () -> realmName;
     polarisContext =
         new PolarisCallContext(
             managerFactory.getOrCreateSessionSupplier(realmContext).get(),
@@ -97,8 +97,16 @@ public void before(TestInfo testInfo) {
             Clock.systemDefaultZone());
   }
 
+  @Test
+  public void testGetConfigurationWithNoRealmContext() {
+    Assertions.assertThatThrownBy(
+            () -> configurationStore.getConfiguration(polarisContext, "missingKeyWithoutDefault"))
+        .isInstanceOf(IllegalStateException.class);
+  }
+
   @Test
   public void testGetConfiguration() {
+    QuarkusMock.installMockForType(realmContext, RealmContext.class);
     Object value = configurationStore.getConfiguration(polarisContext, "missingKeyWithoutDefault");
     assertThat(value).isNull();
     Object defaultValue =
@@ -137,8 +145,28 @@ public void testGetRealmConfiguration() {
         .isFalse();
   }
 
+  @Test
+  void testGetConfigurationWithRealm() {
+    // the falseByDefaultKey is set to `false` for all realms, but overwrite with value `true` for
+    // realmOne.
+    assertThat((Boolean) configurationStore.getConfiguration(realmOneContext, falseByDefaultKey))
+        .isTrue();
+    // the trueByDefaultKey is set to `false` for all realms, no overwrite for realmOne
+    assertThat((Boolean) configurationStore.getConfiguration(realmOneContext, trueByDefaultKey))
+        .isTrue();
+
+    // the falseByDefaultKey is set to `false` for all realms, no overwrite for realmTwo
+    assertThat((Boolean) configurationStore.getConfiguration(realmTwoContext, falseByDefaultKey))
+        .isFalse();
+    // the trueByDefaultKey is set to `false` for all realms, and overwrite with value `false` for
+    // realmTwo
+    assertThat((Boolean) configurationStore.getConfiguration(realmTwoContext, trueByDefaultKey))
+        .isFalse();
+  }
+
   @Test
   public void testInjectedConfigurationStore() {
+    QuarkusMock.installMockForType(realmContext, RealmContext.class);
     // the default value for trueByDefaultKey is `true`
     boolean featureDefaultValue =
         configurationStore.getConfiguration(polarisContext, trueByDefaultKey);
@@ -159,6 +187,7 @@ public void testInjectedConfigurationStore() {
 
   @Test
   public void testInjectedFeaturesConfiguration() {
+    QuarkusMock.installMockForType(realmContext, RealmContext.class);
     assertThat(featuresConfiguration).isInstanceOf(QuarkusResolvedFeaturesConfiguration.class);
 
     assertThat(featuresConfiguration.defaults())
@@ -179,6 +208,7 @@ public void testInjectedFeaturesConfiguration() {
 
   @Test
   public void testRegisterAndUseFeatureConfigurations() {
+    QuarkusMock.installMockForType(realmContext, RealmContext.class);
     String prefix = "testRegisterAndUseFeatureConfigurations";
 
     FeatureConfiguration<Boolean> safeConfig =

service/common/src/main/java/org/apache/polaris/service/catalog/io/FileIOUtil.java

@@ -87,7 +87,7 @@ public static AccessConfig refreshAccessConfig(
 
     boolean skipCredentialSubscopingIndirection =
         configurationStore.getConfiguration(
-            callContext.getPolarisCallContext(),
+            callContext.getRealmContext(),
             FeatureConfiguration.SKIP_CREDENTIAL_SUBSCOPING_INDIRECTION.key,
             FeatureConfiguration.SKIP_CREDENTIAL_SUBSCOPING_INDIRECTION.defaultValue);
     if (skipCredentialSubscopingIndirection) {

service/common/src/main/java/org/apache/polaris/service/config/DefaultConfigurationStore.java

@@ -49,18 +49,23 @@ public DefaultConfigurationStore(
     this.realmOverrides = Map.copyOf(configurations.parseRealmOverrides(objectMapper));
   }
 
+  @Override
+  public <T> @Nullable T getConfiguration(@Nonnull RealmContext realmContext, String configName) {
+    String realm = realmContext.getRealmIdentifier();
+    LOGGER.debug("Get configuration value for {} with realm {}", configName, realm);
+    @SuppressWarnings("unchecked")
+    T confgValue =
+        (T)
+            Optional.ofNullable(realmOverrides.getOrDefault(realm, Map.of()).get(configName))
+                .orElseGet(() -> getDefaultConfiguration(configName));
+    return confgValue;
+  }
+
   @Override
   public <T> @Nullable T getConfiguration(@Nonnull PolarisCallContext ctx, String configName) {
-    if (!realmContextInstance.isUnsatisfied()) {
+    if (realmContextInstance.isResolvable()) {
       RealmContext realmContext = realmContextInstance.get();
-      String realm = realmContext.getRealmIdentifier();
-      LOGGER.debug("Get configuration value for {} with realm {}", configName, realm);
-      @SuppressWarnings("unchecked")
-      T confgValue =
-          (T)
-              Optional.ofNullable(realmOverrides.getOrDefault(realm, Map.of()).get(configName))
-                  .orElseGet(() -> getDefaultConfiguration(configName));
-      return confgValue;
+      return getConfiguration(realmContext, configName);
     } else {
       LOGGER.debug(
           "No RealmContext is injected when lookup value for configuration {} ", configName);

service/common/src/main/java/org/apache/polaris/service/task/TableCleanupTaskHandler.java

@@ -115,12 +115,7 @@ public boolean handleTask(TaskEntity cleanupTask, CallContext callContext) {
 
       Stream<TaskEntity> metadataFileCleanupTasks =
           getMetadataTaskStream(
-              cleanupTask,
-              tableMetadata,
-              fileIO,
-              tableEntity,
-              metaStoreManager,
-              polarisCallContext);
+              cleanupTask, tableMetadata, fileIO, tableEntity, metaStoreManager, callContext);
 
       List<TaskEntity> taskEntities =
           Stream.concat(manifestCleanupTasks, metadataFileCleanupTasks).toList();
@@ -206,11 +201,12 @@ private Stream<TaskEntity> getMetadataTaskStream(
       FileIO fileIO,
       IcebergTableLikeEntity tableEntity,
       PolarisMetaStoreManager metaStoreManager,
-      PolarisCallContext polarisCallContext) {
+      CallContext callContext) {
+    PolarisCallContext polarisCallContext = callContext.getPolarisCallContext();
     int batchSize =
         polarisCallContext
             .getConfigurationStore()
-            .getConfiguration(polarisCallContext, BATCH_SIZE_CONFIG_KEY, 10);
+            .getConfiguration(callContext.getRealmContext(), BATCH_SIZE_CONFIG_KEY, 10);
     return getMetadataFileBatches(tableMetadata, batchSize).stream()
         .map(
             metadataBatch -> {