[core] Improve HttpClient error response handling

paimon-api/src/main/java/org/apache/paimon/rest/HttpClient.java

@@ -131,19 +131,14 @@ private <T extends RESTResponse> T exec(HttpUriRequestBase request, Class<T> res
                     response -> {
                         String responseBodyStr = RESTUtil.extractResponseBodyAsString(response);
                         if (!RESTUtil.isSuccessful(response)) {
-                            ErrorResponse error;
+                            ErrorResponse error = null;
                             try {
                                 error = RESTApi.fromJson(responseBodyStr, ErrorResponse.class);
                             } catch (JsonProcessingException e) {
-                                error =
-                                        new ErrorResponse(
-                                                null,
-                                                null,
-                                                responseBodyStr != null
-                                                        ? responseBodyStr
-                                                        : "response body is null",
-                                                response.getCode());
+                                // ignore exception
                             }
+                            error = buildErrorResponse(error, responseBodyStr, response.getCode());
+
                             errorHandler.accept(error, extractRequestId(response));
                         }
                         if (responseType != null && responseBodyStr != null) {
@@ -228,4 +223,22 @@ private static Header[] getHeaders(
                 .map(entry -> new BasicHeader(entry.getKey(), entry.getValue()))
                 .toArray(Header[]::new);
     }
+
+    private static ErrorResponse buildErrorResponse(
+            ErrorResponse error, String responseBodyStr, int errorCode) {
+        if (error == null || error.getMessage() == null || error.getMessage().isEmpty()) {
+            String resourceType =
+                    (error != null && error.getResourceType() != null)
+                            ? error.getResourceType()
+                            : "";
+            String resourceName =
+                    (error != null && error.getResourceName() != null)
+                            ? error.getResourceName()
+                            : "";
+            String message = responseBodyStr != null ? responseBodyStr : "response body is null";
+            int code = (error != null && error.getCode() != null) ? error.getCode() : errorCode;
+            error = new ErrorResponse(resourceType, resourceName, message, code);
+        }
+        return error;
+    }
 }

paimon-core/src/test/java/org/apache/paimon/rest/HttpClientTest.java

@@ -223,4 +223,42 @@ private Map<String, String> getParameters(String path) {
                                         ));
         return parameters;
     }
+
+    @Test
+    public void testGetWithUnparsableJsonErrorResponse() {
+        // Test case for JSON response with mismatched field names that cannot be parsed as
+        // ErrorResponse
+        String jsonWithUppercaseFields =
+                "{\"Message\":\"Your request is denied as lack of ssl protect.\","
+                        + "\"Code\":\"InvalidProtocol.NeedSsl\"}";
+        server.enqueueResponse(jsonWithUppercaseFields, 403);
+
+        try {
+            httpClient.get(MOCK_PATH, MockRESTData.class, restAuthFunction);
+            Assertions.fail("Expected exception to be thrown");
+        } catch (Exception e) {
+            Assertions.assertTrue(
+                    e.getMessage().contains("Your request is denied as lack of ssl protect")
+                            || e.getMessage().contains(jsonWithUppercaseFields),
+                    "Error message should contain the original response body");
+        }
+    }
+
+    @Test
+    public void testPostWithNonJsonErrorResponse() {
+        // Test case for non-JSON response (plain text) that cannot be parsed
+        String plainTextResponse = "Internal Server Error: Database connection failed";
+        server.enqueueResponse(plainTextResponse, 500);
+
+        try {
+            httpClient.post(MOCK_PATH, mockResponseData, MockRESTData.class, restAuthFunction);
+            Assertions.fail("Expected exception to be thrown");
+        } catch (Exception e) {
+            // Verify that the error message contains the original plain text response
+            Assertions.assertTrue(
+                    e.getMessage().contains(plainTextResponse)
+                            || e.getMessage().contains("Database connection failed"),
+                    "Error message should contain the original non-JSON response");
+        }
+    }
 }