updating my fork

.github/workflows/docs-check-links.yml

@@ -0,0 +1,43 @@
+#
+# 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.
+#
+
+name: Check Markdown docs links
+
+on:
+  push:
+    paths:
+      - docs/**
+      - site/**
+    branches:
+      - 'main'
+  pull_request:
+    paths:
+      - docs/**
+      - site/**
+  workflow_dispatch:
+
+jobs:
+  markdown-link-check:
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v4
+      - uses: gaurav-nelson/github-action-markdown-link-check@v1
+        with:
+          config-file: 'site/link-checker-config.json'
+          use-verbose-mode: yes

README.md

@@ -17,7 +17,7 @@
   - under the License.
   -->
 
-![Iceberg](https://iceberg.apache.org/docs/latest/img/Iceberg-logo.png)
+![Iceberg](https://iceberg.apache.org/assets/images/Iceberg-logo.svg)
 
 [![](https://github.com/apache/iceberg/actions/workflows/java-ci.yml/badge.svg)](https://github.com/apache/iceberg/actions/workflows/java-ci.yml)
 [![Slack](https://img.shields.io/badge/chat-on%20Slack-brightgreen.svg)](https://apache-iceberg.slack.com/)
@@ -37,11 +37,8 @@ The core Java library is located in this repository and is the reference impleme
 
 [Documentation][iceberg-docs] is available for all libraries and integrations.
 
-Current work is tracked in the [roadmap][roadmap].
-
 [iceberg-docs]: https://iceberg.apache.org/docs/latest/
-[iceberg-spec]: https://iceberg.apache.org/spec
-[roadmap]: https://iceberg.apache.org/roadmap/
+[iceberg-spec]: https://iceberg.apache.org/spec/
 
 ## Collaboration
 

aws/src/main/java/org/apache/iceberg/aws/AwsClientFactories.java

@@ -113,6 +113,7 @@ public S3Client s3() {
               b -> s3FileIOProperties.applyCredentialConfigurations(awsClientProperties, b))
           .applyMutation(s3FileIOProperties::applySignerConfiguration)
           .applyMutation(s3FileIOProperties::applyS3AccessGrantsConfigurations)
+          .applyMutation(s3FileIOProperties::applyUserAgentConfigurations)
           .build();
     }
 

aws/src/main/java/org/apache/iceberg/aws/s3/DefaultS3FileIOAwsClientFactory.java

@@ -54,6 +54,7 @@ public S3Client s3() {
                     awsClientProperties, s3ClientBuilder))
         .applyMutation(s3FileIOProperties::applySignerConfiguration)
         .applyMutation(s3FileIOProperties::applyS3AccessGrantsConfigurations)
+        .applyMutation(s3FileIOProperties::applyUserAgentConfigurations)
         .build();
   }
 }

aws/src/main/java/org/apache/iceberg/aws/s3/S3FileIOProperties.java

@@ -24,6 +24,7 @@
 import java.util.Map;
 import java.util.Set;
 import java.util.stream.Collectors;
+import org.apache.iceberg.EnvironmentContext;
 import org.apache.iceberg.aws.AwsClientProperties;
 import org.apache.iceberg.aws.glue.GlueCatalog;
 import org.apache.iceberg.aws.s3.signer.S3V4RestSignerClient;
@@ -375,6 +376,14 @@ public class S3FileIOProperties implements Serializable {
 
   public static final boolean PRELOAD_CLIENT_ENABLED_DEFAULT = false;
 
+  /**
+   * User Agent Prefix set by the S3 client.
+   *
+   * <p>This allows developers to monitor which version of Iceberg they have deployed to a cluster
+   * (for example, through the S3 Access Logs, which contain the user agent field).
+   */
+  private static final String S3_FILE_IO_USER_AGENT = "s3fileio/" + EnvironmentContext.get();
+
   private String sseType;
   private String sseKey;
   private String sseMd5;
@@ -819,6 +828,11 @@ public <T extends S3ClientBuilder> void applyS3AccessGrantsConfigurations(T buil
     }
   }
 
+  public <T extends S3ClientBuilder> void applyUserAgentConfigurations(T builder) {
+    builder.overrideConfiguration(
+        c -> c.putAdvancedOption(SdkAdvancedClientOption.USER_AGENT_PREFIX, S3_FILE_IO_USER_AGENT));
+  }
+
   /**
    * Dynamically load the http client builder to avoid runtime deps requirements of any optional SDK
    * Plugins

aws/src/test/java/org/apache/iceberg/aws/s3/TestS3FileIOProperties.java

@@ -478,4 +478,14 @@ public void testApplyEndpointConfiguration() {
     s3FileIOProperties.applyEndpointConfigurations(mockS3ClientBuilder);
     Mockito.verify(mockS3ClientBuilder).endpointOverride(Mockito.any(URI.class));
   }
+
+  @Test
+  public void testApplyUserAgentConfigurations() {
+    Map<String, String> properties = Maps.newHashMap();
+    S3FileIOProperties s3FileIOProperties = new S3FileIOProperties(properties);
+    S3ClientBuilder mockS3ClientBuilder = Mockito.mock(S3ClientBuilder.class);
+    s3FileIOProperties.applyUserAgentConfigurations(mockS3ClientBuilder);
+
+    Mockito.verify(mockS3ClientBuilder).overrideConfiguration(Mockito.any(Consumer.class));
+  }
 }

core/src/main/java/org/apache/iceberg/BaseMetastoreCatalog.java

@@ -285,26 +285,7 @@ private Map<String, String> tableOverrideProperties() {
   }
 
   protected static String fullTableName(String catalogName, TableIdentifier identifier) {
-    StringBuilder sb = new StringBuilder();
-
-    if (catalogName.contains("/") || catalogName.contains(":")) {
-      // use / for URI-like names: thrift://host:port/db.table
-      sb.append(catalogName);
-      if (!catalogName.endsWith("/")) {
-        sb.append("/");
-      }
-    } else {
-      // use . for non-URI named catalogs: prod.db.table
-      sb.append(catalogName).append(".");
-    }
-
-    for (String level : identifier.namespace().levels()) {
-      sb.append(level).append(".");
-    }
-
-    sb.append(identifier.name());
-
-    return sb.toString();
+    return CatalogUtil.fullTableName(catalogName, identifier);
   }
 
   protected MetricsReporter metricsReporter() {

core/src/main/java/org/apache/iceberg/BaseMetastoreOperations.java

@@ -0,0 +1,118 @@
+/*
+ * 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;
+
+import static org.apache.iceberg.TableProperties.COMMIT_NUM_STATUS_CHECKS;
+import static org.apache.iceberg.TableProperties.COMMIT_NUM_STATUS_CHECKS_DEFAULT;
+import static org.apache.iceberg.TableProperties.COMMIT_STATUS_CHECKS_MAX_WAIT_MS;
+import static org.apache.iceberg.TableProperties.COMMIT_STATUS_CHECKS_MAX_WAIT_MS_DEFAULT;
+import static org.apache.iceberg.TableProperties.COMMIT_STATUS_CHECKS_MIN_WAIT_MS;
+import static org.apache.iceberg.TableProperties.COMMIT_STATUS_CHECKS_MIN_WAIT_MS_DEFAULT;
+import static org.apache.iceberg.TableProperties.COMMIT_STATUS_CHECKS_TOTAL_WAIT_MS;
+import static org.apache.iceberg.TableProperties.COMMIT_STATUS_CHECKS_TOTAL_WAIT_MS_DEFAULT;
+
+import java.util.Map;
+import java.util.concurrent.atomic.AtomicReference;
+import java.util.function.Supplier;
+import org.apache.iceberg.util.PropertyUtil;
+import org.apache.iceberg.util.Tasks;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public abstract class BaseMetastoreOperations {
+  private static final Logger LOG = LoggerFactory.getLogger(BaseMetastoreOperations.class);
+
+  public enum CommitStatus {
+    FAILURE,
+    SUCCESS,
+    UNKNOWN
+  }
+
+  /**
+   * Attempt to load the content and see if any current or past metadata location matches the one we
+   * were attempting to set. This is used as a last resort when we are dealing with exceptions that
+   * may indicate the commit has failed but don't have proof that this is the case. Note that all
+   * the previous locations must also be searched on the chance that a second committer was able to
+   * successfully commit on top of our commit.
+   *
+   * @param tableOrViewName full name of the Table/View
+   * @param newMetadataLocation the path of the new commit file
+   * @param properties properties for retry
+   * @param commitStatusSupplier check if the latest metadata presents or not using metadata
+   *     location for table.
+   * @return Commit Status of Success, Failure or Unknown
+   */
+  protected CommitStatus checkCommitStatus(
+      String tableOrViewName,
+      String newMetadataLocation,
+      Map<String, String> properties,
+      Supplier<Boolean> commitStatusSupplier) {
+    int maxAttempts =
+        PropertyUtil.propertyAsInt(
+            properties, COMMIT_NUM_STATUS_CHECKS, COMMIT_NUM_STATUS_CHECKS_DEFAULT);
+    long minWaitMs =
+        PropertyUtil.propertyAsLong(
+            properties, COMMIT_STATUS_CHECKS_MIN_WAIT_MS, COMMIT_STATUS_CHECKS_MIN_WAIT_MS_DEFAULT);
+    long maxWaitMs =
+        PropertyUtil.propertyAsLong(
+            properties, COMMIT_STATUS_CHECKS_MAX_WAIT_MS, COMMIT_STATUS_CHECKS_MAX_WAIT_MS_DEFAULT);
+    long totalRetryMs =
+        PropertyUtil.propertyAsLong(
+            properties,
+            COMMIT_STATUS_CHECKS_TOTAL_WAIT_MS,
+            COMMIT_STATUS_CHECKS_TOTAL_WAIT_MS_DEFAULT);
+
+    AtomicReference<CommitStatus> status = new AtomicReference<>(CommitStatus.UNKNOWN);
+
+    Tasks.foreach(newMetadataLocation)
+        .retry(maxAttempts)
+        .suppressFailureWhenFinished()
+        .exponentialBackoff(minWaitMs, maxWaitMs, totalRetryMs, 2.0)
+        .onFailure(
+            (location, checkException) ->
+                LOG.error("Cannot check if commit to {} exists.", tableOrViewName, checkException))
+        .run(
+            location -> {
+              boolean commitSuccess = commitStatusSupplier.get();
+
+              if (commitSuccess) {
+                LOG.info(
+                    "Commit status check: Commit to {} of {} succeeded",
+                    tableOrViewName,
+                    newMetadataLocation);
+                status.set(CommitStatus.SUCCESS);
+              } else {
+                LOG.warn(
+                    "Commit status check: Commit to {} of {} unknown, new metadata location is not current "
+                        + "or in history",
+                    tableOrViewName,
+                    newMetadataLocation);
+              }
+            });
+
+    if (status.get() == CommitStatus.UNKNOWN) {
+      LOG.error(
+          "Cannot determine commit state to {}. Failed during checking {} times. "
+              + "Treating commit state as unknown.",
+          tableOrViewName,
+          maxAttempts);
+    }
+    return status.get();
+  }
+}