GH-952: Add OAuth support

docs/source/flight_sql_jdbc_driver.rst

@@ -173,3 +173,126 @@ DriverManager#getConnection()
 <https://docs.oracle.com/javase/8/docs/api/java/sql/DriverManager.html#getConnection-java.lang.String-java.lang.String-java.lang.String->`_,
 the username and password supplied on the URI supercede the username and
 password arguments to the function call.
+
+OAuth 2.0 Authentication
+========================
+
+The driver supports OAuth 2.0 authentication for obtaining access tokens
+from an authorization server. Two OAuth flows are currently supported:
+
+* **Client Credentials** - For service-to-service authentication where no
+  user interaction is required. The application authenticates using its own
+  credentials (client ID and client secret).
+
+* **Token Exchange** (RFC 8693) - For exchanging one token for another,
+  commonly used for federated authentication, delegation, or impersonation
+  scenarios.
+
+OAuth Connection Properties
+---------------------------
+
+The following properties configure OAuth authentication. These properties
+should be provided via the ``Properties`` object when connecting, as they
+may contain special characters that are difficult to encode in a URI.
+
+**Common OAuth Properties**
+
+.. list-table::
+   :header-rows: 1
+
+   * - Parameter
+     - Type
+     - Required
+     - Default
+     - Description
+
+   * - oauth.flow
+     - String
+     - Yes (to enable OAuth)
+     - null
+     - The OAuth grant type. Supported values: ``client_credentials``,
+       ``token_exchange``
+
+   * - oauth.tokenUri
+     - String
+     - Yes
+     - null
+     - The OAuth 2.0 token endpoint URL (e.g.,
+       ``https://auth.example.com/oauth/token``)
+
+   * - oauth.clientId
+     - String
+     - Conditional
+     - null
+     - The OAuth 2.0 client ID. Required for ``client_credentials`` flow,
+       optional for ``token_exchange``
+
+   * - oauth.clientSecret
+     - String
+     - Conditional
+     - null
+     - The OAuth 2.0 client secret. Required for ``client_credentials`` flow,
+       optional for ``token_exchange``
+
+   * - oauth.scope
+     - String
+     - No
+     - null
+     - Space-separated list of OAuth scopes to request
+
+   * - oauth.resource
+     - String
+     - No
+     - null
+     - The resource indicator for the token request (RFC 8707)
+
+**Token Exchange Properties**
+
+These properties are specific to the ``token_exchange`` flow:
+
+.. list-table::
+   :header-rows: 1
+
+   * - Parameter
+     - Type
+     - Required
+     - Default
+     - Description
+
+   * - oauth.exchange.subjectToken
+     - String
+     - Yes
+     - null
+     - The subject token to exchange (e.g., a JWT from an identity provider)
+
+   * - oauth.exchange.subjectTokenType
+     - String
+     - Yes
+     - null
+     - The token type URI of the subject token. Common values:
+       ``urn:ietf:params:oauth:token-type:access_token``,
+       ``urn:ietf:params:oauth:token-type:jwt``
+
+   * - oauth.exchange.actorToken
+     - String
+     - No
+     - null
+     - The actor token for delegation/impersonation scenarios
+
+   * - oauth.exchange.actorTokenType
+     - String
+     - No
+     - null
+     - The token type URI of the actor token
+
+   * - oauth.exchange.aud
+     - String
+     - No
+     - null
+     - The target audience for the exchanged token
+
+   * - oauth.exchange.requestedTokenType
+     - String
+     - No
+     - null
+     - The desired token type for the exchanged token

flight/flight-core/src/main/java/org/apache/arrow/flight/grpc/CallCredentialAdapter.java

@@ -18,6 +18,7 @@
 
 import io.grpc.CallCredentials;
 import io.grpc.Metadata;
+import io.grpc.Status;
 import java.util.concurrent.Executor;
 import java.util.function.Consumer;
 import org.apache.arrow.flight.CallHeaders;
@@ -36,9 +37,14 @@ public void applyRequestMetadata(
       RequestInfo requestInfo, Executor executor, MetadataApplier metadataApplier) {
     executor.execute(
         () -> {
-          final Metadata headers = new Metadata();
-          credentialWriter.accept(new MetadataAdapter(headers));
-          metadataApplier.apply(headers);
+          try {
+            final Metadata headers = new Metadata();
+            credentialWriter.accept(new MetadataAdapter(headers));
+            metadataApplier.apply(headers);
+          } catch (Throwable t) {
+            metadataApplier.fail(
+                Status.UNAUTHENTICATED.withCause(t).withDescription(t.getMessage()));
+          }
         });
   }
 

flight/flight-sql-jdbc-core/pom.xml

@@ -153,6 +153,19 @@ under the License.
       <artifactId>caffeine</artifactId>
       <version>3.2.0</version>
     </dependency>
+
+    <dependency>
+      <groupId>com.nimbusds</groupId>
+      <artifactId>oauth2-oidc-sdk</artifactId>
+      <version>11.20.1</version>
+    </dependency>
+
+    <dependency>
+      <groupId>com.squareup.okhttp3</groupId>
+      <artifactId>mockwebserver</artifactId>
+      <version>4.12.0</version>
+      <scope>test</scope>
+    </dependency>
   </dependencies>
 
   <build>

flight/flight-sql-jdbc-core/src/main/java/org/apache/arrow/driver/jdbc/ArrowFlightConnection.java

@@ -121,6 +121,7 @@ private static ArrowFlightSqlClientHandler createNewClientHandler(
           .withClientCache(config.useClientCache() ? new FlightClientCache() : null)
           .withConnectTimeout(config.getConnectTimeout())
           .withDriverVersion(driverVersion)
+          .withOAuthConfiguration(config.getOauthConfiguration())
           .build();
     } catch (final SQLException e) {
       try {

flight/flight-sql-jdbc-core/src/main/java/org/apache/arrow/driver/jdbc/client/ArrowFlightSqlClientHandler.java

@@ -32,6 +32,9 @@
 import java.util.Map;
 import java.util.Optional;
 import java.util.Set;
+import org.apache.arrow.driver.jdbc.client.oauth.OAuthConfiguration;
+import org.apache.arrow.driver.jdbc.client.oauth.OAuthCredentialWriter;
+import org.apache.arrow.driver.jdbc.client.oauth.OAuthTokenProvider;
 import org.apache.arrow.driver.jdbc.client.utils.ClientAuthenticationUtils;
 import org.apache.arrow.driver.jdbc.client.utils.FlightClientCache;
 import org.apache.arrow.driver.jdbc.client.utils.FlightLocationQueue;
@@ -675,6 +678,8 @@ public static final class Builder {
 
     @VisibleForTesting @Nullable Duration connectTimeout;
 
+    @VisibleForTesting @Nullable OAuthConfiguration oauthConfig;
+
     // These two middleware are for internal use within build() and should not be
     // exposed by builder
     // APIs.
@@ -714,6 +719,7 @@ public Builder() {}
       this.clientKeyPath = original.clientKeyPath;
       this.allocator = original.allocator;
       this.catalog = original.catalog;
+      this.oauthConfig = original.oauthConfig;
 
       if (original.retainCookies) {
         this.cookieFactory = original.cookieFactory;
@@ -983,6 +989,17 @@ public Builder withDriverVersion(DriverVersion driverVersion) {
       return this;
     }
 
+    /**
+     * Sets the OAuth configuration for this handler.
+     *
+     * @param oauthConfig the OAuth configuration
+     * @return this builder instance
+     */
+    public Builder withOAuthConfiguration(final OAuthConfiguration oauthConfig) {
+      this.oauthConfig = oauthConfig;
+      return this;
+    }
+
     public String getCacheKey() {
       return getLocation().toString();
     }
@@ -1070,7 +1087,11 @@ public ArrowFlightSqlClientHandler build() throws SQLException {
             FlightGrpcUtils.createFlightClient(
                 allocator, channelBuilder.build(), clientBuilder.middleware());
         final ArrayList<CallOption> credentialOptions = new ArrayList<>();
-        if (isUsingUserPasswordAuth) {
+        // Authentication priority: OAuth > token > username/password
+        if (oauthConfig != null) {
+          OAuthTokenProvider tokenProvider = oauthConfig.createTokenProvider();
+          credentialOptions.add(new CredentialCallOption(new OAuthCredentialWriter(tokenProvider)));
+        } else if (isUsingUserPasswordAuth) {
           // If the authFactory has already been used for a handshake, use the existing
           // token.
           // This can occur if the authFactory is being re-used for a new connection

flight/flight-sql-jdbc-core/src/main/java/org/apache/arrow/driver/jdbc/client/oauth/AbstractOAuthTokenProvider.java

@@ -0,0 +1,108 @@
+/*
+ * 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.arrow.driver.jdbc.client.oauth;
+
+import com.nimbusds.oauth2.sdk.ParseException;
+import com.nimbusds.oauth2.sdk.Scope;
+import com.nimbusds.oauth2.sdk.TokenErrorResponse;
+import com.nimbusds.oauth2.sdk.TokenRequest;
+import com.nimbusds.oauth2.sdk.TokenResponse;
+import com.nimbusds.oauth2.sdk.auth.ClientAuthentication;
+import com.nimbusds.oauth2.sdk.token.AccessToken;
+import java.io.IOException;
+import java.net.URI;
+import java.sql.SQLException;
+import java.time.Instant;
+import org.apache.arrow.util.VisibleForTesting;
+import org.checkerframework.checker.nullness.qual.Nullable;
+
+/**
+ * Abstract base class for OAuth token providers that handles token caching, refresh logic, and
+ * common request/response handling.
+ */
+public abstract class AbstractOAuthTokenProvider implements OAuthTokenProvider {
+  protected static final int EXPIRATION_BUFFER_SECONDS = 30;
+  protected static final int DEFAULT_EXPIRATION_SECONDS = 3600;
+
+  private final Object tokenLock = new Object();
+  private volatile @Nullable TokenInfo cachedToken;
+
+  @VisibleForTesting URI tokenUri;
+
+  @VisibleForTesting @Nullable ClientAuthentication clientAuth;
+
+  @VisibleForTesting @Nullable Scope scope;
+
+  @Override
+  public String getValidToken() throws SQLException {
+    TokenInfo token = cachedToken;
+    if (token != null && !token.isExpired(EXPIRATION_BUFFER_SECONDS)) {
+      return token.getAccessToken();
+    }
+
+    synchronized (tokenLock) {
+      token = cachedToken;
+      if (token != null && !token.isExpired(EXPIRATION_BUFFER_SECONDS)) {
+        return token.getAccessToken();
+      }
+      cachedToken = fetchNewToken();
+      return cachedToken.getAccessToken();
+    }
+  }
+
+  /**
+   * Fetches a new token from the authorization server. This method handles the common
+   * request/response logic while delegating flow-specific request building to subclasses.
+   *
+   * @return the new token information
+   * @throws SQLException if token cannot be obtained
+   */
+  protected TokenInfo fetchNewToken() throws SQLException {
+    try {
+      TokenRequest request = buildTokenRequest();
+      TokenResponse response = TokenResponse.parse(request.toHTTPRequest().send());
+
+      if (!response.indicatesSuccess()) {
+        TokenErrorResponse errorResponse = response.toErrorResponse();
+        String errorMsg =
+            String.format(
+                "OAuth request failed: %s - %s",
+                errorResponse.getErrorObject().getCode(),
+                errorResponse.getErrorObject().getDescription());
+        throw new SQLException(errorMsg);
+      }
+
+      AccessToken accessToken = response.toSuccessResponse().getTokens().getAccessToken();
+      long expiresIn =
+          accessToken.getLifetime() > 0 ? accessToken.getLifetime() : DEFAULT_EXPIRATION_SECONDS;
+      Instant expiresAt = Instant.now().plusSeconds(expiresIn);
+
+      return new TokenInfo(accessToken.getValue(), expiresAt);
+    } catch (ParseException e) {
+      throw new SQLException("Failed to parse OAuth token response", e);
+    } catch (IOException e) {
+      throw new SQLException("Failed to send OAuth token request", e);
+    }
+  }
+
+  /**
+   * Builds the flow-specific token request.
+   *
+   * @return the token request to send to the authorization server
+   */
+  protected abstract TokenRequest buildTokenRequest();
+}