inject PolarisMetaStoreManager into DefaultOAuth2ApiService

this avoids `TokenBrokerFactory` impls having to call
`MetaStoreManagerFactory.createMetaStoreManager`

Author: XN137

runtime/service/src/main/java/org/apache/polaris/service/auth/internal/broker/JWTBroker.java

@@ -27,7 +27,6 @@
 import java.util.Optional;
 import java.util.UUID;
 import org.apache.iceberg.exceptions.NotAuthorizedException;
-import org.apache.polaris.core.PolarisCallContext;
 import org.apache.polaris.core.entity.PolarisEntityType;
 import org.apache.polaris.core.entity.PrincipalEntity;
 import org.apache.polaris.core.persistence.PolarisMetaStoreManager;
@@ -50,11 +49,9 @@ public abstract class JWTBroker implements TokenBroker {
   private static final String CLAIM_KEY_PRINCIPAL_ID = "principalId";
   private static final String CLAIM_KEY_SCOPE = "scope";
 
-  private final PolarisMetaStoreManager metaStoreManager;
   private final int maxTokenGenerationInSeconds;
 
-  JWTBroker(PolarisMetaStoreManager metaStoreManager, int maxTokenGenerationInSeconds) {
-    this.metaStoreManager = metaStoreManager;
+  JWTBroker(int maxTokenGenerationInSeconds) {
     this.maxTokenGenerationInSeconds = maxTokenGenerationInSeconds;
   }
 
@@ -88,7 +85,7 @@ public TokenResponse generateFromToken(
       String subjectToken,
       String grantType,
       String scope,
-      PolarisCallContext polarisCallContext,
+      PolarisMetaStoreManager metaStoreManager,
       TokenType requestedTokenType) {
     if (requestedTokenType != null && !TokenType.ACCESS_TOKEN.equals(requestedTokenType)) {
       return TokenResponse.of(OAuthError.invalid_request);
@@ -128,7 +125,7 @@ public TokenResponse generateFromClientSecrets(
       String clientSecret,
       String grantType,
       String scope,
-      PolarisCallContext polarisCallContext,
+      PolarisMetaStoreManager metaStoreManager,
       TokenType requestedTokenType) {
     // Initial sanity checks
     TokenRequestValidator validator = new TokenRequestValidator();
@@ -138,7 +135,8 @@ public TokenResponse generateFromClientSecrets(
       return TokenResponse.of(initialValidationResponse.get());
     }
 
-    Optional<PrincipalEntity> principal = findPrincipalEntity(clientId, clientSecret);
+    Optional<PrincipalEntity> principal =
+        findPrincipalEntity(metaStoreManager, clientId, clientSecret);
     if (principal.isEmpty()) {
       return TokenResponse.of(OAuthError.unauthorized_client);
     }
@@ -178,7 +176,8 @@ private String scopes(String scope) {
     return scope == null || scope.isBlank() ? DefaultAuthenticator.PRINCIPAL_ROLE_ALL : scope;
   }
 
-  private Optional<PrincipalEntity> findPrincipalEntity(String clientId, String clientSecret) {
+  private Optional<PrincipalEntity> findPrincipalEntity(
+      PolarisMetaStoreManager metaStoreManager, String clientId, String clientSecret) {
     // Validate the principal is present and secrets match
     PrincipalSecretsResult principalSecrets = metaStoreManager.loadPrincipalSecrets(clientId);
     if (!principalSecrets.isSuccess()) {

runtime/service/src/main/java/org/apache/polaris/service/auth/internal/broker/RSAKeyPairJWTBroker.java

@@ -21,18 +21,14 @@
 import com.auth0.jwt.algorithms.Algorithm;
 import java.security.interfaces.RSAPrivateKey;
 import java.security.interfaces.RSAPublicKey;
-import org.apache.polaris.core.persistence.PolarisMetaStoreManager;
 
 /** Generates a JWT using a Public/Private RSA Key */
 public class RSAKeyPairJWTBroker extends JWTBroker {
 
   private final KeyProvider keyProvider;
 
-  RSAKeyPairJWTBroker(
-      PolarisMetaStoreManager metaStoreManager,
-      int maxTokenGenerationInSeconds,
-      KeyProvider keyProvider) {
-    super(metaStoreManager, maxTokenGenerationInSeconds);
+  RSAKeyPairJWTBroker(int maxTokenGenerationInSeconds, KeyProvider keyProvider) {
+    super(maxTokenGenerationInSeconds);
     this.keyProvider = keyProvider;
   }
 

runtime/service/src/main/java/org/apache/polaris/service/auth/internal/broker/RSAKeyPairJWTBrokerFactory.java

@@ -26,8 +26,6 @@
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ConcurrentMap;
 import org.apache.polaris.core.context.RealmContext;
-import org.apache.polaris.core.persistence.MetaStoreManagerFactory;
-import org.apache.polaris.core.persistence.PolarisMetaStoreManager;
 import org.apache.polaris.service.auth.AuthenticationConfiguration;
 import org.apache.polaris.service.auth.AuthenticationRealmConfiguration;
 import org.apache.polaris.service.auth.AuthenticationRealmConfiguration.TokenBrokerConfiguration.RSAKeyPairConfiguration;
@@ -36,21 +34,17 @@
 @Identifier("rsa-key-pair")
 public class RSAKeyPairJWTBrokerFactory implements TokenBrokerFactory {
 
-  private final MetaStoreManagerFactory metaStoreManagerFactory;
   private final AuthenticationConfiguration authenticationConfiguration;
 
   private final ConcurrentMap<String, RSAKeyPairJWTBroker> tokenBrokers = new ConcurrentHashMap<>();
 
   @Inject
-  public RSAKeyPairJWTBrokerFactory(
-      MetaStoreManagerFactory metaStoreManagerFactory,
-      AuthenticationConfiguration authenticationConfiguration) {
-    this.metaStoreManagerFactory = metaStoreManagerFactory;
+  public RSAKeyPairJWTBrokerFactory(AuthenticationConfiguration authenticationConfiguration) {
     this.authenticationConfiguration = authenticationConfiguration;
   }
 
   @Override
-  public TokenBroker apply(RealmContext realmContext) {
+  public TokenBroker newTokenBroker(RealmContext realmContext) {
     return tokenBrokers.computeIfAbsent(
         realmContext.getRealmIdentifier(), k -> createTokenBroker(realmContext));
   }
@@ -64,10 +58,7 @@ private RSAKeyPairJWTBroker createTokenBroker(RealmContext realmContext) {
             .rsaKeyPair()
             .map(this::fileSystemKeyPair)
             .orElseGet(this::generateEphemeralKeyPair);
-    PolarisMetaStoreManager metaStoreManager =
-        metaStoreManagerFactory.createMetaStoreManager(realmContext, null);
-    return new RSAKeyPairJWTBroker(
-        metaStoreManager, (int) maxTokenGeneration.toSeconds(), keyProvider);
+    return new RSAKeyPairJWTBroker((int) maxTokenGeneration.toSeconds(), keyProvider);
   }
 
   private KeyProvider fileSystemKeyPair(RSAKeyPairConfiguration config) {

runtime/service/src/main/java/org/apache/polaris/service/auth/internal/broker/SymmetricKeyJWTBroker.java

@@ -20,17 +20,13 @@
 
 import com.auth0.jwt.algorithms.Algorithm;
 import java.util.function.Supplier;
-import org.apache.polaris.core.persistence.PolarisMetaStoreManager;
 
 /** Generates a JWT using a Symmetric Key. */
 public class SymmetricKeyJWTBroker extends JWTBroker {
   private final Supplier<String> secretSupplier;
 
-  public SymmetricKeyJWTBroker(
-      PolarisMetaStoreManager metaStoreManager,
-      int maxTokenGenerationInSeconds,
-      Supplier<String> secretSupplier) {
-    super(metaStoreManager, maxTokenGenerationInSeconds);
+  public SymmetricKeyJWTBroker(int maxTokenGenerationInSeconds, Supplier<String> secretSupplier) {
+    super(maxTokenGenerationInSeconds);
     this.secretSupplier = secretSupplier;
   }
 

runtime/service/src/main/java/org/apache/polaris/service/auth/internal/broker/SymmetricKeyJWTBrokerFactory.java

@@ -31,8 +31,6 @@
 import java.util.concurrent.ConcurrentMap;
 import java.util.function.Supplier;
 import org.apache.polaris.core.context.RealmContext;
-import org.apache.polaris.core.persistence.MetaStoreManagerFactory;
-import org.apache.polaris.core.persistence.PolarisMetaStoreManager;
 import org.apache.polaris.service.auth.AuthenticationConfiguration;
 import org.apache.polaris.service.auth.AuthenticationRealmConfiguration;
 import org.apache.polaris.service.auth.AuthenticationRealmConfiguration.TokenBrokerConfiguration.SymmetricKeyConfiguration;
@@ -41,22 +39,18 @@
 @Identifier("symmetric-key")
 public class SymmetricKeyJWTBrokerFactory implements TokenBrokerFactory {
 
-  private final MetaStoreManagerFactory metaStoreManagerFactory;
   private final AuthenticationConfiguration authenticationConfiguration;
 
   private final ConcurrentMap<String, SymmetricKeyJWTBroker> tokenBrokers =
       new ConcurrentHashMap<>();
 
   @Inject
-  public SymmetricKeyJWTBrokerFactory(
-      MetaStoreManagerFactory metaStoreManagerFactory,
-      AuthenticationConfiguration authenticationConfiguration) {
-    this.metaStoreManagerFactory = metaStoreManagerFactory;
+  public SymmetricKeyJWTBrokerFactory(AuthenticationConfiguration authenticationConfiguration) {
     this.authenticationConfiguration = authenticationConfiguration;
   }
 
   @Override
-  public TokenBroker apply(RealmContext realmContext) {
+  public TokenBroker newTokenBroker(RealmContext realmContext) {
     return tokenBrokers.computeIfAbsent(
         realmContext.getRealmIdentifier(), k -> createTokenBroker(realmContext));
   }
@@ -73,10 +67,7 @@ private SymmetricKeyJWTBroker createTokenBroker(RealmContext realmContext) {
     Path file = symmetricKeyConfiguration.file().orElse(null);
     checkState(secret != null || file != null, "Either file or secret must be set");
     Supplier<String> secretSupplier = secret != null ? () -> secret : readSecretFromDisk(file);
-    PolarisMetaStoreManager metaStoreManager =
-        metaStoreManagerFactory.createMetaStoreManager(realmContext, null);
-    return new SymmetricKeyJWTBroker(
-        metaStoreManager, (int) maxTokenGeneration.toSeconds(), secretSupplier);
+    return new SymmetricKeyJWTBroker((int) maxTokenGeneration.toSeconds(), secretSupplier);
   }
 
   private static Supplier<String> readSecretFromDisk(Path file) {

runtime/service/src/main/java/org/apache/polaris/service/auth/internal/broker/TokenBroker.java

@@ -18,7 +18,7 @@
  */
 package org.apache.polaris.service.auth.internal.broker;
 
-import org.apache.polaris.core.PolarisCallContext;
+import org.apache.polaris.core.persistence.PolarisMetaStoreManager;
 import org.apache.polaris.service.auth.PolarisCredential;
 import org.apache.polaris.service.types.TokenType;
 
@@ -39,7 +39,7 @@ TokenResponse generateFromClientSecrets(
       final String clientSecret,
       final String grantType,
       final String scope,
-      PolarisCallContext polarisCallContext,
+      PolarisMetaStoreManager metaStoreManager,
       TokenType requestedTokenType);
 
   /**
@@ -52,7 +52,7 @@ TokenResponse generateFromToken(
       String subjectToken,
       final String grantType,
       final String scope,
-      PolarisCallContext polarisCallContext,
+      PolarisMetaStoreManager metaStoreManager,
       TokenType requestedTokenType);
 
   /** Decodes and verifies the token, then returns the associated {@link PolarisCredential}. */

runtime/service/src/main/java/org/apache/polaris/service/auth/internal/broker/TokenBrokerFactory.java

@@ -18,11 +18,13 @@
  */
 package org.apache.polaris.service.auth.internal.broker;
 
-import java.util.function.Function;
 import org.apache.polaris.core.context.RealmContext;
 
 /**
  * Factory that creates a {@link TokenBroker} for generating and parsing. The {@link TokenBroker} is
  * created based on the realm context.
  */
-public interface TokenBrokerFactory extends Function<RealmContext, TokenBroker> {}
+public interface TokenBrokerFactory {
+
+  TokenBroker newTokenBroker(RealmContext realmContext);
+}

runtime/service/src/main/java/org/apache/polaris/service/auth/internal/service/DefaultOAuth2ApiService.java

@@ -27,8 +27,8 @@
 import jakarta.ws.rs.core.SecurityContext;
 import java.util.Base64;
 import org.apache.iceberg.rest.responses.OAuthTokenResponse;
-import org.apache.polaris.core.context.CallContext;
 import org.apache.polaris.core.context.RealmContext;
+import org.apache.polaris.core.persistence.PolarisMetaStoreManager;
 import org.apache.polaris.service.auth.internal.broker.TokenBroker;
 import org.apache.polaris.service.auth.internal.broker.TokenResponse;
 import org.apache.polaris.service.catalog.api.IcebergRestOAuth2ApiService;
@@ -49,12 +49,13 @@ public class DefaultOAuth2ApiService implements IcebergRestOAuth2ApiService {
   private static final String BEARER = "bearer";
 
   private final TokenBroker tokenBroker;
-  private final CallContext callContext;
+  private final PolarisMetaStoreManager metaStoreManager;
 
   @Inject
-  public DefaultOAuth2ApiService(TokenBroker tokenBroker, CallContext callContext) {
+  public DefaultOAuth2ApiService(
+      TokenBroker tokenBroker, PolarisMetaStoreManager metaStoreManager) {
     this.tokenBroker = tokenBroker;
-    this.callContext = callContext;
+    this.metaStoreManager = metaStoreManager;
   }
 
   @Override
@@ -104,20 +105,15 @@ public Response getToken(
     if (clientSecret != null) {
       tokenResponse =
           tokenBroker.generateFromClientSecrets(
-              clientId,
-              clientSecret,
-              grantType,
-              scope,
-              callContext.getPolarisCallContext(),
-              requestedTokenType);
+              clientId, clientSecret, grantType, scope, metaStoreManager, requestedTokenType);
     } else if (subjectToken != null) {
       tokenResponse =
           tokenBroker.generateFromToken(
               subjectTokenType,
               subjectToken,
               grantType,
               scope,
-              callContext.getPolarisCallContext(),
+              metaStoreManager,
               requestedTokenType);
     } else {
       return OAuthUtils.getResponseFromError(OAuthError.invalid_request);

runtime/service/src/main/java/org/apache/polaris/service/config/ServiceProducers.java

@@ -353,7 +353,7 @@ public TokenBroker tokenBroker(
         config.type() == AuthenticationType.EXTERNAL ? "none" : config.tokenBroker().type();
     TokenBrokerFactory tokenBrokerFactory =
         tokenBrokerFactories.select(Identifier.Literal.of(type)).get();
-    return tokenBrokerFactory.apply(realmContext);
+    return tokenBrokerFactory.newTokenBroker(realmContext);
   }
 
   // other beans

runtime/service/src/test/java/org/apache/polaris/service/auth/internal/broker/JWTSymmetricKeyGeneratorTest.java

@@ -24,7 +24,6 @@
 import com.auth0.jwt.JWTVerifier;
 import com.auth0.jwt.algorithms.Algorithm;
 import com.auth0.jwt.interfaces.DecodedJWT;
-import org.apache.polaris.core.PolarisCallContext;
 import org.apache.polaris.core.entity.PolarisBaseEntity;
 import org.apache.polaris.core.entity.PolarisEntitySubType;
 import org.apache.polaris.core.entity.PolarisEntityType;
@@ -41,7 +40,6 @@ public class JWTSymmetricKeyGeneratorTest {
   /** Sanity test to verify that we can generate a token */
   @Test
   public void testJWTSymmetricKeyGenerator() {
-    PolarisCallContext polarisCallContext = new PolarisCallContext(null, null);
     PolarisMetaStoreManager metastoreManager = Mockito.mock(PolarisMetaStoreManager.class);
     String mainSecret = "test_secret";
     String clientId = "test_client_id";
@@ -59,14 +57,14 @@ public void testJWTSymmetricKeyGenerator() {
             "principal");
     Mockito.when(metastoreManager.loadEntity(0L, 1L, PolarisEntityType.PRINCIPAL))
         .thenReturn(new EntityResult(principal));
-    TokenBroker generator = new SymmetricKeyJWTBroker(metastoreManager, 666, () -> "polaris");
+    TokenBroker generator = new SymmetricKeyJWTBroker(666, () -> "polaris");
     TokenResponse token =
         generator.generateFromClientSecrets(
             clientId,
             mainSecret,
             TokenRequestValidator.CLIENT_CREDENTIALS,
             "PRINCIPAL_ROLE:TEST",
-            polarisCallContext,
+            metastoreManager,
             TokenType.ACCESS_TOKEN);
     assertThat(token).isNotNull();