Core: Add LoadCredentialsResponse class/parser

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

@@ -59,7 +59,10 @@
 import org.apache.iceberg.rest.responses.ConfigResponseParser;
 import org.apache.iceberg.rest.responses.ErrorResponse;
 import org.apache.iceberg.rest.responses.ErrorResponseParser;
+import org.apache.iceberg.rest.responses.ImmutableLoadCredentialsResponse;
 import org.apache.iceberg.rest.responses.ImmutableLoadViewResponse;
+import org.apache.iceberg.rest.responses.LoadCredentialsResponse;
+import org.apache.iceberg.rest.responses.LoadCredentialsResponseParser;
 import org.apache.iceberg.rest.responses.LoadTableResponse;
 import org.apache.iceberg.rest.responses.LoadTableResponseParser;
 import org.apache.iceberg.rest.responses.LoadViewResponse;
@@ -119,7 +122,13 @@ public static void registerAll(ObjectMapper mapper) {
         .addSerializer(ConfigResponse.class, new ConfigResponseSerializer<>())
         .addDeserializer(ConfigResponse.class, new ConfigResponseDeserializer<>())
         .addSerializer(LoadTableResponse.class, new LoadTableResponseSerializer<>())
-        .addDeserializer(LoadTableResponse.class, new LoadTableResponseDeserializer<>());
+        .addDeserializer(LoadTableResponse.class, new LoadTableResponseDeserializer<>())
+        .addSerializer(LoadCredentialsResponse.class, new LoadCredentialsResponseSerializer<>())
+        .addSerializer(
+            ImmutableLoadCredentialsResponse.class, new LoadCredentialsResponseSerializer<>())
+        .addDeserializer(LoadCredentialsResponse.class, new LoadCredentialsResponseDeserializer<>())
+        .addDeserializer(
+            ImmutableLoadCredentialsResponse.class, new LoadCredentialsResponseDeserializer<>());
 
     mapper.registerModule(module);
   }
@@ -443,4 +452,22 @@ public T deserialize(JsonParser p, DeserializationContext context) throws IOExce
       return (T) LoadTableResponseParser.fromJson(jsonNode);
     }
   }
+
+  static class LoadCredentialsResponseSerializer<T extends LoadCredentialsResponse>
+      extends JsonSerializer<T> {
+    @Override
+    public void serialize(T request, JsonGenerator gen, SerializerProvider serializers)
+        throws IOException {
+      LoadCredentialsResponseParser.toJson(request, gen);
+    }
+  }
+
+  static class LoadCredentialsResponseDeserializer<T extends LoadCredentialsResponse>
+      extends JsonDeserializer<T> {
+    @Override
+    public T deserialize(JsonParser p, DeserializationContext context) throws IOException {
+      JsonNode jsonNode = p.getCodec().readTree(p);
+      return (T) LoadCredentialsResponseParser.fromJson(jsonNode);
+    }
+  }
 }

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

@@ -0,0 +1,34 @@
+/*
+ * 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 java.util.List;
+import org.apache.iceberg.rest.RESTResponse;
+import org.apache.iceberg.rest.credentials.Credential;
+import org.immutables.value.Value;
+
+@Value.Immutable
+public interface LoadCredentialsResponse extends RESTResponse {
+  List<Credential> credentials();
+
+  @Override
+  default void validate() {
+    // nothing to validate
+  }
+}

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

@@ -0,0 +1,77 @@
+/*
+ * 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.core.JsonGenerator;
+import com.fasterxml.jackson.databind.JsonNode;
+import java.io.IOException;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.rest.credentials.Credential;
+import org.apache.iceberg.rest.credentials.CredentialParser;
+import org.apache.iceberg.util.JsonUtil;
+
+public class LoadCredentialsResponseParser {
+  private static final String STORAGE_CREDENTIALS = "storage-credentials";
+
+  private LoadCredentialsResponseParser() {}
+
+  public static String toJson(LoadCredentialsResponse response) {
+    return toJson(response, false);
+  }
+
+  public static String toJson(LoadCredentialsResponse response, boolean pretty) {
+    return JsonUtil.generate(gen -> toJson(response, gen), pretty);
+  }
+
+  public static void toJson(LoadCredentialsResponse response, JsonGenerator gen)
+      throws IOException {
+    Preconditions.checkArgument(null != response, "Invalid load credentials response: null");
+
+    gen.writeStartObject();
+
+    gen.writeArrayFieldStart(STORAGE_CREDENTIALS);
+    for (Credential credential : response.credentials()) {
+      CredentialParser.toJson(credential, gen);
+    }
+
+    gen.writeEndArray();
+
+    gen.writeEndObject();
+  }
+
+  public static LoadCredentialsResponse fromJson(String json) {
+    return JsonUtil.parse(json, LoadCredentialsResponseParser::fromJson);
+  }
+
+  public static LoadCredentialsResponse fromJson(JsonNode json) {
+    Preconditions.checkArgument(
+        null != json, "Cannot parse load credentials response from null object");
+
+    JsonNode credentials = JsonUtil.get(STORAGE_CREDENTIALS, json);
+    Preconditions.checkArgument(
+        credentials.isArray(), "Cannot parse credentials from non-array: %s", credentials);
+
+    ImmutableLoadCredentialsResponse.Builder builder = ImmutableLoadCredentialsResponse.builder();
+    for (JsonNode credential : credentials) {
+      builder.addCredentials(CredentialParser.fromJson(credential));
+    }
+
+    return builder.build();
+  }
+}

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

@@ -98,13 +98,7 @@ public static LoadTableResponse fromJson(JsonNode json) {
     }
 
     if (json.hasNonNull(STORAGE_CREDENTIALS)) {
-      JsonNode credentials = JsonUtil.get(STORAGE_CREDENTIALS, json);
-      Preconditions.checkArgument(
-          credentials.isArray(), "Cannot parse credentials from non-array: %s", credentials);
-
-      for (JsonNode credential : credentials) {
-        builder.addCredential(CredentialParser.fromJson(credential));
-      }
+      builder.addAllCredentials(LoadCredentialsResponseParser.fromJson(json).credentials());
     }
 
     return builder.build();

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

@@ -93,13 +93,7 @@ public static LoadViewResponse fromJson(JsonNode json) {
     }
 
     if (json.hasNonNull(STORAGE_CREDENTIALS)) {
-      JsonNode credentials = JsonUtil.get(STORAGE_CREDENTIALS, json);
-      Preconditions.checkArgument(
-          credentials.isArray(), "Cannot parse credentials from non-array: %s", credentials);
-
-      for (JsonNode credential : credentials) {
-        builder.addCredentials(CredentialParser.fromJson(credential));
-      }
+      builder.addAllCredentials(LoadCredentialsResponseParser.fromJson(json).credentials());
     }
 
     return builder.build();

core/src/test/java/org/apache/iceberg/rest/responses/TestLoadCredentialsResponseParser.java

@@ -0,0 +1,112 @@
+/*
+ * 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 static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
+import org.apache.iceberg.rest.credentials.ImmutableCredential;
+import org.junit.jupiter.api.Test;
+
+public class TestLoadCredentialsResponseParser {
+  @Test
+  public void nullCheck() {
+    assertThatThrownBy(() -> LoadCredentialsResponseParser.toJson(null))
+        .isInstanceOf(IllegalArgumentException.class)
+        .hasMessage("Invalid load credentials response: null");
+
+    assertThatThrownBy(() -> LoadCredentialsResponseParser.fromJson((JsonNode) null))
+        .isInstanceOf(IllegalArgumentException.class)
+        .hasMessage("Cannot parse load credentials response from null object");
+  }
+
+  @Test
+  public void missingFields() {
+    assertThatThrownBy(() -> LoadCredentialsResponseParser.fromJson("{}"))
+        .isInstanceOf(IllegalArgumentException.class)
+        .hasMessage("Cannot parse missing field: storage-credentials");
+
+    assertThatThrownBy(() -> LoadCredentialsResponseParser.fromJson("{\"x\": \"val\"}"))
+        .isInstanceOf(IllegalArgumentException.class)
+        .hasMessage("Cannot parse missing field: storage-credentials");
+  }
+
+  @Test
+  public void roundTripSerde() {
+    LoadCredentialsResponse response =
+        ImmutableLoadCredentialsResponse.builder()
+            .addCredentials(
+                ImmutableCredential.builder()
+                    .prefix("s3://custom-uri")
+                    .config(
+                        ImmutableMap.of(
+                            "s3.access-key-id",
+                            "keyId",
+                            "s3.secret-access-key",
+                            "accessKey",
+                            "s3.session-token",
+                            "sessionToken"))
+                    .build())
+            .addCredentials(
+                ImmutableCredential.builder()
+                    .prefix("gs://custom-uri")
+                    .config(
+                        ImmutableMap.of(
+                            "gcs.oauth2.token", "gcsToken1", "gcs.oauth2.token-expires-at", "1000"))
+                    .build())
+            .addCredentials(
+                ImmutableCredential.builder()
+                    .prefix("gs")
+                    .config(
+                        ImmutableMap.of(
+                            "gcs.oauth2.token", "gcsToken2", "gcs.oauth2.token-expires-at", "2000"))
+                    .build())
+            .build();
+
+    String expectedJson =
+        "{\n"
+            + "  \"storage-credentials\" : [ {\n"
+            + "    \"prefix\" : \"s3://custom-uri\",\n"
+            + "    \"config\" : {\n"
+            + "      \"s3.access-key-id\" : \"keyId\",\n"
+            + "      \"s3.secret-access-key\" : \"accessKey\",\n"
+            + "      \"s3.session-token\" : \"sessionToken\"\n"
+            + "    }\n"
+            + "  }, {\n"
+            + "    \"prefix\" : \"gs://custom-uri\",\n"
+            + "    \"config\" : {\n"
+            + "      \"gcs.oauth2.token\" : \"gcsToken1\",\n"
+            + "      \"gcs.oauth2.token-expires-at\" : \"1000\"\n"
+            + "    }\n"
+            + "  }, {\n"
+            + "    \"prefix\" : \"gs\",\n"
+            + "    \"config\" : {\n"
+            + "      \"gcs.oauth2.token\" : \"gcsToken2\",\n"
+            + "      \"gcs.oauth2.token-expires-at\" : \"2000\"\n"
+            + "    }\n"
+            + "  } ]\n"
+            + "}";
+
+    String json = LoadCredentialsResponseParser.toJson(response, true);
+    assertThat(json).isEqualTo(expectedJson);
+    assertThat(LoadCredentialsResponseParser.fromJson(json)).isEqualTo(response);
+  }
+}