Core: Extended header support for RESTClient implementations

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

@@ -89,6 +89,18 @@ public <T extends RESTResponse> T get(
     return execute(request, responseType, errorHandler, h -> {}, parserContext);
   }
 
+  @Override
+  public <T extends RESTResponse> T get(
+      String path,
+      Map<String, String> queryParams,
+      Class<T> responseType,
+      Map<String, String> headers,
+      Consumer<ErrorResponse> errorHandler,
+      Consumer<Map<String, String>> responseHeaders) {
+    HTTPRequest request = buildRequest(HTTPMethod.GET, path, queryParams, headers, null);
+    return execute(request, responseType, errorHandler, responseHeaders);
+  }
+
   @Override
   public <T extends RESTResponse> T post(
       String path,

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

@@ -0,0 +1,37 @@
+/*
+ * 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.nio.charset.StandardCharsets;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.relocated.com.google.common.hash.HashFunction;
+import org.apache.iceberg.relocated.com.google.common.hash.Hashing;
+
+class ETagProvider {
+  private static final HashFunction MURMUR3 = Hashing.murmur3_32_fixed();
+
+  private ETagProvider() {}
+
+  public static String of(String metadataLocation) {
+    Preconditions.checkArgument(null != metadataLocation, "Invalid metadata location: null");
+    Preconditions.checkArgument(!metadataLocation.isEmpty(), "Invalid metadata location: empty");
+
+    return MURMUR3.hashString(metadataLocation, StandardCharsets.UTF_8).toString();
+  }
+}

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

@@ -110,6 +110,30 @@ default <T extends RESTResponse> T get(
     return get(path, queryParams, responseType, headers, errorHandler);
   }
 
+  default <T extends RESTResponse> T get(
+      String path,
+      Map<String, String> queryParams,
+      Class<T> responseType,
+      Supplier<Map<String, String>> headers,
+      Consumer<ErrorResponse> errorHandler,
+      Consumer<Map<String, String>> responseHeaders) {
+    return get(path, queryParams, responseType, headers.get(), errorHandler, responseHeaders);
+  }
+
+  default <T extends RESTResponse> T get(
+      String path,
+      Map<String, String> queryParams,
+      Class<T> responseType,
+      Map<String, String> headers,
+      Consumer<ErrorResponse> errorHandler,
+      Consumer<Map<String, String>> responseHeaders) {
+    if (null != responseHeaders) {
+      throw new UnsupportedOperationException("Returning response headers is not supported");
+    }
+
+    return get(path, queryParams, responseType, headers, errorHandler);
+  }
+
   <T extends RESTResponse> T get(
       String path,
       Map<String, String> queryParams,

core/src/test/java/org/apache/iceberg/rest/RESTCatalogAdapter.java

@@ -26,6 +26,7 @@
 import java.util.Map;
 import java.util.function.Consumer;
 import java.util.stream.Collectors;
+import org.apache.http.HttpHeaders;
 import org.apache.iceberg.BaseTable;
 import org.apache.iceberg.BaseTransaction;
 import org.apache.iceberg.Table;
@@ -287,7 +288,12 @@ public RESTClient withAuthSession(AuthSession session) {
 
   @SuppressWarnings({"MethodLength", "checkstyle:CyclomaticComplexity"})
   public <T extends RESTResponse> T handleRequest(
-      Route route, Map<String, String> vars, Object body, Class<T> responseType) {
+      Route route,
+      Map<String, String> vars,
+      HTTPRequest httpRequest,
+      Class<T> responseType,
+      Consumer<Map<String, String>> responseHeaders) {
+    Object body = httpRequest.body();
     switch (route) {
       case TOKENS:
         return castResponse(responseType, handleOAuthRequest(body));
@@ -388,12 +394,15 @@ public <T extends RESTResponse> T handleRequest(
           Namespace namespace = namespaceFromPathVars(vars);
           CreateTableRequest request = castRequest(CreateTableRequest.class, body);
           request.validate();
+
           if (request.stageCreate()) {
             return castResponse(
                 responseType, CatalogHandlers.stageTableCreate(catalog, namespace, request));
           } else {
-            return castResponse(
-                responseType, CatalogHandlers.createTable(catalog, namespace, request));
+            LoadTableResponse resp = CatalogHandlers.createTable(catalog, namespace, request);
+            responseHeaders.accept(
+                ImmutableMap.of(HttpHeaders.ETAG, ETagProvider.of(resp.metadataLocation())));
+            return castResponse(responseType, resp);
           }
         }
 
@@ -416,23 +425,40 @@ public <T extends RESTResponse> T handleRequest(
 
       case LOAD_TABLE:
         {
-          TableIdentifier ident = tableIdentFromPathVars(vars);
-          return castResponse(responseType, CatalogHandlers.loadTable(catalog, ident));
+          LoadTableResponse resp = CatalogHandlers.loadTable(catalog, tableIdentFromPathVars(vars));
+
+          responseHeaders.accept(
+              ImmutableMap.of(HttpHeaders.ETAG, ETagProvider.of(resp.metadataLocation())));
+
+          return castResponse(responseType, resp);
         }
 
       case REGISTER_TABLE:
         {
-          Namespace namespace = namespaceFromPathVars(vars);
-          RegisterTableRequest request = castRequest(RegisterTableRequest.class, body);
-          return castResponse(
-              responseType, CatalogHandlers.registerTable(catalog, namespace, request));
+          LoadTableResponse resp =
+              CatalogHandlers.registerTable(
+                  catalog,
+                  namespaceFromPathVars(vars),
+                  castRequest(RegisterTableRequest.class, body));
+
+          responseHeaders.accept(
+              ImmutableMap.of(HttpHeaders.ETAG, ETagProvider.of(resp.metadataLocation())));
+
+          return castResponse(responseType, resp);
         }
 
       case UPDATE_TABLE:
         {
-          TableIdentifier ident = tableIdentFromPathVars(vars);
-          UpdateTableRequest request = castRequest(UpdateTableRequest.class, body);
-          return castResponse(responseType, CatalogHandlers.updateTable(catalog, ident, request));
+          LoadTableResponse resp =
+              CatalogHandlers.updateTable(
+                  catalog,
+                  tableIdentFromPathVars(vars),
+                  castRequest(UpdateTableRequest.class, body));
+
+          responseHeaders.accept(
+              ImmutableMap.of(HttpHeaders.ETAG, ETagProvider.of(resp.metadataLocation())));
+
+          return castResponse(responseType, resp);
         }
 
       case RENAME_TABLE:
@@ -625,8 +651,8 @@ protected <T extends RESTResponse> T execute(
         vars.putAll(request.queryParameters());
         vars.putAll(routeAndVars.second());
 
-        return handleRequest(routeAndVars.first(), vars.build(), request.body(), responseType);
-
+        return handleRequest(
+            routeAndVars.first(), vars.build(), request, responseType, responseHeaders);
       } catch (RuntimeException e) {
         configureResponseFromException(e, errorBuilder);
       }

core/src/test/java/org/apache/iceberg/rest/RESTCatalogServlet.java

@@ -37,6 +37,7 @@
 import org.apache.hc.core5.http.HttpHeaders;
 import org.apache.iceberg.exceptions.RESTException;
 import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
+import org.apache.iceberg.relocated.com.google.common.collect.Maps;
 import org.apache.iceberg.relocated.com.google.common.io.CharStreams;
 import org.apache.iceberg.rest.HTTPRequest.HTTPMethod;
 import org.apache.iceberg.rest.RESTCatalogAdapter.Route;
@@ -52,10 +53,11 @@
 public class RESTCatalogServlet extends HttpServlet {
   private static final Logger LOG = LoggerFactory.getLogger(RESTCatalogServlet.class);
 
-  private final RESTCatalogAdapter restCatalogAdapter;
-  private final Map<String, String> responseHeaders =
+  private static final Map<String, String> DEFAULT_RESPONSE_HEADERS =
       ImmutableMap.of(HttpHeaders.CONTENT_TYPE, ContentType.APPLICATION_JSON.getMimeType());
 
+  private final RESTCatalogAdapter restCatalogAdapter;
+
   public RESTCatalogServlet(RESTCatalogAdapter restCatalogAdapter) {
     this.restCatalogAdapter = restCatalogAdapter;
   }
@@ -87,7 +89,7 @@ protected void doDelete(HttpServletRequest request, HttpServletResponse response
   protected void execute(ServletRequestContext context, HttpServletResponse response)
       throws IOException {
     response.setStatus(HttpServletResponse.SC_OK);
-    responseHeaders.forEach(response::setHeader);
+    DEFAULT_RESPONSE_HEADERS.forEach(response::setHeader);
 
     if (context.error().isPresent()) {
       response.setStatus(HttpServletResponse.SC_BAD_REQUEST);
@@ -96,17 +98,20 @@ protected void execute(ServletRequestContext context, HttpServletResponse respon
     }
 
     try {
-
       HTTPRequest request =
           restCatalogAdapter.buildRequest(
               context.method(),
               context.path(),
               context.queryParams(),
               context.headers(),
               context.body());
+
+      Map<String, String> responseHeaders = Maps.newHashMap();
       Object responseBody =
           restCatalogAdapter.execute(
-              request, context.route().responseClass(), handle(response), h -> {});
+              request, context.route().responseClass(), handle(response), responseHeaders::putAll);
+
+      responseHeaders.forEach(response::setHeader);
 
       if (responseBody != null) {
         RESTObjectMapper.mapper().writeValue(response.getWriter(), responseBody);

core/src/test/java/org/apache/iceberg/rest/TestETagProvider.java

@@ -0,0 +1,50 @@
+/*
+ * 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 static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+import org.junit.jupiter.api.Test;
+
+public class TestETagProvider {
+  @Test
+  public void testNullInput() {
+    assertThatThrownBy(() -> ETagProvider.of(null))
+        .isInstanceOf(IllegalArgumentException.class)
+        .hasMessageContaining("Invalid metadata location: null");
+  }
+
+  @Test
+  public void testEmptyInput() {
+    assertThatThrownBy(() -> ETagProvider.of(""))
+        .isInstanceOf(IllegalArgumentException.class)
+        .hasMessageContaining("Invalid metadata location: empty");
+  }
+
+  @Test
+  public void testETagContent() {
+    assertThat("1f865717")
+        .isEqualTo(
+            ETagProvider.of(
+                "/var/folders/20/290st0_52y5fyjcj2mlg49500000gn/T/junit-3064022805908958416/db_name/tbl_name/metadata/00000-f7a7956e-61d0-499b-be60-b141283f8229.metadata.json"));
+
+    assertThat("55faa5d9").isEqualTo(ETagProvider.of("/short/path"));
+  }
+}

core/src/test/java/org/apache/iceberg/rest/TestHTTPClient.java

@@ -70,7 +70,7 @@
 
 /**
  * * Exercises the RESTClient interface, specifically over a mocked-server using the actual
- * HttpRESTClient code.
+ * HTTPClient code.
  */
 public class TestHTTPClient {
 
@@ -602,7 +602,8 @@ private static Item doExecuteRequest(
       case POST:
         return restClient.post(path, body, Item.class, headers, onError, responseHeaders);
       case GET:
-        return restClient.get(path, Item.class, headers, onError);
+        return restClient.get(
+            path, ImmutableMap.of(), Item.class, headers, onError, responseHeaders);
       case HEAD:
         restClient.head(path, headers, onError);
         return null;