Core: Stop fetchToken from sending the Authorization header

Author: okumin

core/src/main/java/org/apache/iceberg/rest/auth/OAuth2Util.java

@@ -34,6 +34,7 @@
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicReference;
 import java.util.regex.Pattern;
+import java.util.stream.Collectors;
 import org.apache.iceberg.relocated.com.google.common.annotations.VisibleForTesting;
 import org.apache.iceberg.relocated.com.google.common.base.Joiner;
 import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
@@ -243,7 +244,14 @@ public static OAuthTokenResponse fetchToken(
             oauth2ServerUri,
             request,
             OAuthTokenResponse.class,
-            headers,
+            // RFC 6749 proposes two ways to send a credential: HTTP Basic authentication or
+            // request-body. Historically, the Iceberg library prefers the latter one for
+            // compatibility while the RFC recommends the former one. As sending both the
+            // Authorization header and the request-body parameters might confuse some authorization
+            // servers, the following line is excluding the Authorization header.
+            headers.entrySet().stream()
+                .filter(entry -> !AUTHORIZATION_HEADER.equals(entry.getKey()))
+                .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue)),
             ErrorHandlers.oauthErrorHandler());
     response.validate();
 

core/src/test/java/org/apache/iceberg/rest/auth/TestOAuth2Manager.java

@@ -230,7 +230,7 @@ void contextualSessionCredentialsProvided() {
     SessionCatalog.SessionContext context =
         new SessionCatalog.SessionContext(
             "test", "test", Map.of(OAuth2Properties.CREDENTIAL, "client:secret"), Map.of());
-    Map<String, String> properties = Map.of();
+    Map<String, String> properties = Map.of(OAuth2Properties.CREDENTIAL, "client:secret");
     try (OAuth2Manager manager = new OAuth2Manager("test");
         OAuth2Util.AuthSession catalogSession = manager.catalogSession(client, properties);
         OAuth2Util.AuthSession contextualSession =
@@ -243,7 +243,7 @@ void contextualSessionCredentialsProvided() {
           .as("should create session cache for context with credentials")
           .satisfies(cache -> assertThat(cache.sessionCache().asMap()).hasSize(1));
     }
-    Mockito.verify(client)
+    Mockito.verify(client, times(2))
         .postForm(
             any(),
             eq(
@@ -255,7 +255,7 @@ void contextualSessionCredentialsProvided() {
             eq(OAuthTokenResponse.class),
             eq(Map.of()),
             any());
-    Mockito.verify(client).withAuthSession(any());
+    Mockito.verify(client, times(2)).withAuthSession(any());
     Mockito.verifyNoMoreInteractions(client);
   }