Add support for token exchange in Iceberg REST catalog

docs/src/main/sphinx/object-storage/metastores.md

@@ -509,6 +509,9 @@ following properties:
 * - `iceberg.rest-catalog.oauth2.token-refresh-enabled`
   - Controls whether a token should be refreshed if information about its expiration time is available.
     Defaults to `true`
+* - `iceberg.rest-catalog.oauth2.token-exchange-enabled`
+  - Controls whether to use the token exchange flow to acquire new tokens.
+    Defaults to `true` 
 * - `iceberg.rest-catalog.vended-credentials-enabled`
   - Use credentials provided by the REST backend for file system access.
     Defaults to `false`.

plugin/trino-iceberg/src/main/java/io/trino/plugin/iceberg/catalog/rest/OAuth2SecurityConfig.java

@@ -29,6 +29,7 @@ public class OAuth2SecurityConfig
     private String token;
     private URI serverUri;
     private boolean tokenRefreshEnabled = OAuth2Properties.TOKEN_REFRESH_ENABLED_DEFAULT;
+    private boolean tokenExchangeEnabled = OAuth2Properties.TOKEN_EXCHANGE_ENABLED_DEFAULT;
 
     public Optional<String> getCredential()
     {
@@ -97,6 +98,19 @@ public OAuth2SecurityConfig setTokenRefreshEnabled(boolean tokenRefreshEnabled)
         return this;
     }
 
+    public boolean isTokenExchangeEnabled()
+    {
+        return tokenExchangeEnabled;
+    }
+
+    @Config("iceberg.rest-catalog.oauth2.token-exchange-enabled")
+    @ConfigDescription("Controls whether to use the token exchange flow to acquire new tokens")
+    public OAuth2SecurityConfig setTokenExchangeEnabled(boolean tokenExchangeEnabled)
+    {
+        this.tokenExchangeEnabled = tokenExchangeEnabled;
+        return this;
+    }
+
     @AssertTrue(message = "OAuth2 requires a credential or token")
     public boolean credentialOrTokenPresent()
     {

plugin/trino-iceberg/src/main/java/io/trino/plugin/iceberg/catalog/rest/OAuth2SecurityProperties.java

@@ -45,6 +45,7 @@ public OAuth2SecurityProperties(OAuth2SecurityConfig securityConfig)
         securityConfig.getServerUri().ifPresent(
                 value -> propertiesBuilder.put(OAuth2Properties.OAUTH2_SERVER_URI, value.toString()));
         propertiesBuilder.put(OAuth2Properties.TOKEN_REFRESH_ENABLED, String.valueOf(securityConfig.isTokenRefreshEnabled()));
+        propertiesBuilder.put(OAuth2Properties.TOKEN_EXCHANGE_ENABLED, String.valueOf(securityConfig.isTokenExchangeEnabled()));
 
         this.securityProperties = propertiesBuilder.buildOrThrow();
     }

plugin/trino-iceberg/src/test/java/io/trino/plugin/iceberg/catalog/rest/TestOAuth2SecurityConfig.java

@@ -35,7 +35,8 @@ public void testDefaults()
                 .setToken(null)
                 .setScope(null)
                 .setServerUri(null)
-                .setTokenRefreshEnabled(OAuth2Properties.TOKEN_REFRESH_ENABLED_DEFAULT));
+                .setTokenRefreshEnabled(OAuth2Properties.TOKEN_REFRESH_ENABLED_DEFAULT)
+                .setTokenExchangeEnabled(OAuth2Properties.TOKEN_EXCHANGE_ENABLED_DEFAULT));
     }
 
     @Test
@@ -47,14 +48,16 @@ public void testExplicitPropertyMappings()
                 .put("iceberg.rest-catalog.oauth2.scope", "scope")
                 .put("iceberg.rest-catalog.oauth2.server-uri", "http://localhost:8080/realms/iceberg/protocol/openid-connect/token")
                 .put("iceberg.rest-catalog.oauth2.token-refresh-enabled", "false")
+                .put("iceberg.rest-catalog.oauth2.token-exchange-enabled", "false")
                 .buildOrThrow();
 
         OAuth2SecurityConfig expected = new OAuth2SecurityConfig()
                 .setCredential("credential")
                 .setToken("token")
                 .setScope("scope")
                 .setServerUri(URI.create("http://localhost:8080/realms/iceberg/protocol/openid-connect/token"))
-                .setTokenRefreshEnabled(false);
+                .setTokenRefreshEnabled(false)
+                .setTokenExchangeEnabled(false);
         assertThat(expected.credentialOrTokenPresent()).isTrue();
         assertThat(expected.scopePresentOnlyWithCredential()).isFalse();
         assertFullMapping(properties, expected);