REST: implement handling of OAuth error responses followup (apache#5820)

* WIP error handling for OAuth

* cleanup

* tests

* handle non-oauth errors in oauth

* add comment

* allow null fields

* more tests

* more cleanup

* remove unneeded precondition checks

* Fix test

* use assert4j

* Keep client API the same

* Keep ErrorResponse the same

* PR feedback

* handle invalid error response body format

core/src/main/java/org/apache/iceberg/rest/ErrorHandler.java

@@ -0,0 +1,27 @@
+/*
+ * 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.iceberg.rest;
+
+import java.util.function.Consumer;
+import org.apache.iceberg.rest.responses.ErrorResponse;
+
+public abstract class ErrorHandler implements Consumer<ErrorResponse> {
+
+  public abstract ErrorResponse parseResponse(int code, String json);
+}

core/src/main/java/org/apache/iceberg/rest/ErrorHandlers.java

@@ -30,30 +30,49 @@
 import org.apache.iceberg.exceptions.RESTException;
 import org.apache.iceberg.exceptions.ServiceFailureException;
 import org.apache.iceberg.exceptions.ServiceUnavailableException;
+import org.apache.iceberg.rest.auth.OAuth2Properties;
 import org.apache.iceberg.rest.responses.ErrorResponse;
+import org.apache.iceberg.rest.responses.ErrorResponseParser;
+import org.apache.iceberg.rest.responses.OAuthErrorResponseParser;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 /**
  * A set of consumers to handle errors for requests for table entities or for namespace entities, to
  * throw the correct exception.
  */
 public class ErrorHandlers {
 
+  private static final Logger LOG = LoggerFactory.getLogger(ErrorHandlers.class);
+
   private ErrorHandlers() {}
 
   public static Consumer<ErrorResponse> namespaceErrorHandler() {
-    return baseNamespaceErrorHandler().andThen(defaultErrorHandler());
+    return NamespaceErrorHandler.INSTANCE;
   }
 
   public static Consumer<ErrorResponse> tableErrorHandler() {
-    return baseTableErrorHandler().andThen(defaultErrorHandler());
+    return TableErrorHandler.INSTANCE;
   }
 
   public static Consumer<ErrorResponse> tableCommitHandler() {
-    return baseCommitErrorHandler().andThen(defaultErrorHandler());
+    return CommitErrorHandler.INSTANCE;
+  }
+
+  public static Consumer<ErrorResponse> defaultErrorHandler() {
+    return DefaultErrorHandler.INSTANCE;
   }
 
-  private static Consumer<ErrorResponse> baseCommitErrorHandler() {
-    return error -> {
+  public static Consumer<ErrorResponse> oauthErrorHandler() {
+    return OAuthErrorHandler.INSTANCE;
+  }
+
+  /** Table commit error handler. */
+  private static class CommitErrorHandler extends DefaultErrorHandler {
+    private static final ErrorHandler INSTANCE = new CommitErrorHandler();
+
+    @Override
+    public void accept(ErrorResponse error) {
       switch (error.code()) {
         case 404:
           throw new NoSuchTableException("%s", error.message());
@@ -64,15 +83,17 @@ private static Consumer<ErrorResponse> baseCommitErrorHandler() {
           throw new CommitStateUnknownException(
               new ServiceFailureException("Service failed: %s: %s", error.code(), error.message()));
       }
-    };
+
+      super.accept(error);
+    }
   }
 
-  /**
-   * Table level error handlers. Should be chained wih the {@link #defaultErrorHandler}, which takes
-   * care of common cases.
-   */
-  private static Consumer<ErrorResponse> baseTableErrorHandler() {
-    return error -> {
+  /** Table level error handler. */
+  private static class TableErrorHandler extends DefaultErrorHandler {
+    private static final ErrorHandler INSTANCE = new TableErrorHandler();
+
+    @Override
+    public void accept(ErrorResponse error) {
       switch (error.code()) {
         case 404:
           if (NoSuchNamespaceException.class.getSimpleName().equals(error.type())) {
@@ -83,15 +104,17 @@ private static Consumer<ErrorResponse> baseTableErrorHandler() {
         case 409:
           throw new AlreadyExistsException("%s", error.message());
       }
-    };
+
+      super.accept(error);
+    }
   }
 
-  /**
-   * Request error handlers specifically for CRUD ops on namespaces. Should be chained wih the
-   * {@link #defaultErrorHandler}, which takes care of common cases.
-   */
-  private static Consumer<ErrorResponse> baseNamespaceErrorHandler() {
-    return error -> {
+  /** Request error handler specifically for CRUD ops on namespaces. */
+  private static class NamespaceErrorHandler extends DefaultErrorHandler {
+    private static final ErrorHandler INSTANCE = new NamespaceErrorHandler();
+
+    @Override
+    public void accept(ErrorResponse error) {
       switch (error.code()) {
         case 404:
           throw new NoSuchNamespaceException("%s", error.message());
@@ -100,15 +123,30 @@ private static Consumer<ErrorResponse> baseNamespaceErrorHandler() {
         case 422:
           throw new RESTException("Unable to process: %s", error.message());
       }
-    };
+
+      super.accept(error);
+    }
   }
 
   /**
    * Request error handler that handles the common cases that are included with all responses, such
    * as 400, 500, etc.
    */
-  public static Consumer<ErrorResponse> defaultErrorHandler() {
-    return error -> {
+  private static class DefaultErrorHandler extends ErrorHandler {
+    private static final ErrorHandler INSTANCE = new DefaultErrorHandler();
+
+    @Override
+    public ErrorResponse parseResponse(int code, String json) {
+      try {
+        return ErrorResponseParser.fromJson(json);
+      } catch (Exception x) {
+        LOG.warn("Unable to parse error response", x);
+      }
+      return ErrorResponse.builder().responseCode(code).withMessage(json).build();
+    }
+
+    @Override
+    public void accept(ErrorResponse error) {
       switch (error.code()) {
         case 400:
           throw new BadRequestException("Malformed request: %s", error.message());
@@ -128,6 +166,39 @@ public static Consumer<ErrorResponse> defaultErrorHandler() {
       }
 
       throw new RESTException("Unable to process: %s", error.message());
-    };
+    }
+  }
+
+  private static class OAuthErrorHandler extends ErrorHandler {
+    private static final ErrorHandler INSTANCE = new OAuthErrorHandler();
+
+    @Override
+    public ErrorResponse parseResponse(int code, String json) {
+      try {
+        return OAuthErrorResponseParser.fromJson(code, json);
+      } catch (Exception x) {
+        LOG.warn("Unable to parse error response", x);
+      }
+      return ErrorResponse.builder().responseCode(code).withMessage(json).build();
+    }
+
+    @Override
+    public void accept(ErrorResponse error) {
+      if (error.type() != null) {
+        switch (error.type()) {
+          case OAuth2Properties.INVALID_CLIENT_ERROR:
+            throw new NotAuthorizedException(
+                "Not authorized: %s: %s", error.type(), error.message());
+          case OAuth2Properties.INVALID_REQUEST_ERROR:
+          case OAuth2Properties.INVALID_GRANT_ERROR:
+          case OAuth2Properties.UNAUTHORIZED_CLIENT_ERROR:
+          case OAuth2Properties.UNSUPPORTED_GRANT_TYPE_ERROR:
+          case OAuth2Properties.INVALID_SCOPE_ERROR:
+            throw new BadRequestException(
+                "Malformed request: %s: %s", error.type(), error.message());
+        }
+      }
+      throw new RESTException("Unable to process: %s", error.message());
+    }
   }
 }

core/src/main/java/org/apache/iceberg/rest/HTTPClient.java

@@ -46,7 +46,6 @@
 import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
 import org.apache.iceberg.relocated.com.google.common.collect.Maps;
 import org.apache.iceberg.rest.responses.ErrorResponse;
-import org.apache.iceberg.rest.responses.ErrorResponseParser;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -110,7 +109,20 @@ private static void throwFailure(
 
     if (responseBody != null) {
       try {
-        errorResponse = ErrorResponseParser.fromJson(responseBody);
+        if (errorHandler instanceof ErrorHandler) {
+          errorResponse =
+              ((ErrorHandler) errorHandler).parseResponse(response.getCode(), responseBody);
+        } else {
+          LOG.warn(
+              "Unknown error handler {}, response body won't be parsed",
+              errorHandler.getClass().getName());
+          errorResponse =
+              ErrorResponse.builder()
+                  .responseCode(response.getCode())
+                  .withMessage(responseBody)
+                  .build();
+        }
+
       } catch (UncheckedIOException | IllegalArgumentException e) {
         // It's possible to receive a non-successful response that isn't a properly defined
         // ErrorResponse

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

@@ -48,4 +48,12 @@ private OAuth2Properties() {}
   public static final String SAML1_TOKEN_TYPE = "urn:ietf:params:oauth:token-type:saml1";
   public static final String SAML2_TOKEN_TYPE = "urn:ietf:params:oauth:token-type:saml2";
   public static final String JWT_TOKEN_TYPE = "urn:ietf:params:oauth:token-type:jwt";
+
+  // error type constants
+  public static final String INVALID_REQUEST_ERROR = "invalid_request";
+  public static final String INVALID_CLIENT_ERROR = "invalid_client";
+  public static final String INVALID_GRANT_ERROR = "invalid_grant";
+  public static final String UNAUTHORIZED_CLIENT_ERROR = "unauthorized_client";
+  public static final String UNSUPPORTED_GRANT_TYPE_ERROR = "unsupported_grant_type";
+  public static final String INVALID_SCOPE_ERROR = "invalid_scope";
 }

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

@@ -132,7 +132,7 @@ private static OAuthTokenResponse refreshToken(
             request,
             OAuthTokenResponse.class,
             headers,
-            ErrorHandlers.defaultErrorHandler());
+            ErrorHandlers.oauthErrorHandler());
     response.validate();
 
     return response;
@@ -160,7 +160,7 @@ public static OAuthTokenResponse exchangeToken(
             request,
             OAuthTokenResponse.class,
             headers,
-            ErrorHandlers.defaultErrorHandler());
+            ErrorHandlers.oauthErrorHandler());
     response.validate();
 
     return response;
@@ -178,7 +178,7 @@ public static OAuthTokenResponse fetchToken(
             request,
             OAuthTokenResponse.class,
             headers,
-            ErrorHandlers.defaultErrorHandler());
+            ErrorHandlers.oauthErrorHandler());
     response.validate();
 
     return response;

core/src/main/java/org/apache/iceberg/rest/responses/OAuthErrorResponseParser.java

@@ -0,0 +1,61 @@
+/*
+ * 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.iceberg.rest.responses;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import java.io.IOException;
+import java.io.UncheckedIOException;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.util.JsonUtil;
+
+public class OAuthErrorResponseParser {
+
+  private OAuthErrorResponseParser() {}
+
+  private static final String ERROR = "error";
+  private static final String ERROR_DESCRIPTION = "error_description";
+
+  /**
+   * Read OAuthErrorResponse from a JSON string.
+   *
+   * @param json a JSON string of an OAuthErrorResponse
+   * @return an OAuthErrorResponse object
+   */
+  public static ErrorResponse fromJson(int code, String json) {
+    try {
+      return fromJson(code, JsonUtil.mapper().readValue(json, JsonNode.class));
+    } catch (IOException e) {
+      throw new UncheckedIOException("Failed to read JSON string: " + json, e);
+    }
+  }
+
+  public static ErrorResponse fromJson(int code, JsonNode jsonNode) {
+    Preconditions.checkArgument(
+        jsonNode != null && jsonNode.isObject(),
+        "Cannot parse error response from non-object value: %s",
+        jsonNode);
+    String error = JsonUtil.getString(ERROR, jsonNode);
+    String errorDescription = JsonUtil.getStringOrNull(ERROR_DESCRIPTION, jsonNode);
+    return ErrorResponse.builder()
+        .responseCode(code)
+        .withType(error)
+        .withMessage(errorDescription)
+        .build();
+  }
+}