feat: Add externalized commit retries and commit lock using etcd distributed locks (#162)

Co-authored-by: Andrew Xie <dev@xie.is>

Author: subkanthi

examples/grafana/METRICS.md

@@ -58,6 +58,10 @@ These metrics are reported by Iceberg clients when they perform operations on ta
 | `iceberg_commit_added_equality_deletes_total` | Counter | catalog, namespace, table, operation | Total number of equality deletes added in commits |
 | `iceberg_commit_total_files_size_bytes` | Counter | catalog, namespace, table, operation | Total size in bytes of files involved in commits |
 | `iceberg_commit_duration_seconds` | Histogram | catalog, namespace, table, operation | Duration of commit operations in seconds |
+| `iceberg_commit_retries_total` | Counter | catalog, namespace, table | Server-side retries after a commit CAS conflict (`CommitFailedException`) in the REST catalog commit loop; tune `commitRetry` in `.ice-rest-catalog.yaml` if this grows under parallel writers |
+| `iceberg_commit_lock_acquire_seconds` | Histogram | catalog | Time to acquire the etcd per-table commit lock (`commitLock` in `.ice-rest-catalog.yaml`; etcd backend only) |
+| `iceberg_commit_lock_held_seconds` | Histogram | catalog | Time the etcd commit lock was held during a table commit |
+| `iceberg_commit_lock_acquire_timeouts_total` | Counter | catalog | Acquire attempts that exceeded `commitLock.acquireTimeoutMs` (HTTP 503 to clients) |
 
 #### Reporter Metrics
 

examples/scratch/.ice-rest-catalog.yaml

@@ -17,4 +17,4 @@ bearerTokens:
 
 anonymousAccess:
   enabled: true
-  accessConfig: {}
+  accessConfig: {}

ice-rest-catalog/README.md

@@ -11,6 +11,19 @@ That's it.
 
 Examples of `.ice-rest-catalog.yaml` (as well as Kubernetes deployment manifests) can be found [here](../examples/).
 
+## Parallel writers (`commitLock`)
+
+Many concurrent commits to the **same table** can cause repeated `CommitFailedException` (optimistic concurrency). For the **etcd** metastore you can serialize commits per table using etcd’s lock API:
+
+```yaml
+commitLock:
+  enabled: true
+  leaseTtlSeconds: 30
+  acquireTimeoutMs: 30000
+```
+
+If `enabled` is true but the catalog backend is not etcd, the lock is ignored (warning in logs). When lock acquisition exceeds `acquireTimeoutMs`, the server responds with HTTP **503** so clients can retry.
+
 ## Documentation
 
 - [Architecture](../docs/architecture.md) -- components, design principles, HA, backup/recovery

ice-rest-catalog/src/main/java/com/altinity/ice/rest/catalog/Main.java

@@ -19,6 +19,7 @@
 import com.altinity.ice.rest.catalog.internal.aws.CredentialsProvider;
 import com.altinity.ice.rest.catalog.internal.config.Config;
 import com.altinity.ice.rest.catalog.internal.config.MaintenanceConfig;
+import com.altinity.ice.rest.catalog.internal.etcd.CommitLock;
 import com.altinity.ice.rest.catalog.internal.etcd.EtcdCatalog;
 import com.altinity.ice.rest.catalog.internal.maintenance.DataCompaction;
 import com.altinity.ice.rest.catalog.internal.maintenance.MaintenanceJob;
@@ -279,7 +280,8 @@ private static Server createBaseServer(
     if (requireAuth) {
       mux.insertHandler(createAuthorizationHandler(config.bearerTokens(), config));
 
-      restCatalogAdapter = new RESTCatalogAdapter(catalog);
+      restCatalogAdapter =
+          new RESTCatalogAdapter(catalog, config.commitRetry(), maybeCommitLock(catalog, config));
       var globalConfig = config.toIcebergConfigDefaults();
       if (!globalConfig.isEmpty()) {
         restCatalogAdapter = new RESTCatalogMiddlewareConfig(restCatalogAdapter, globalConfig);
@@ -299,7 +301,8 @@ private static Server createBaseServer(
                 new RESTCatalogMiddlewareCredentials(restCatalogAdapter, auth), auth);
       }
     } else {
-      restCatalogAdapter = new RESTCatalogAdapter(catalog);
+      restCatalogAdapter =
+          new RESTCatalogAdapter(catalog, config.commitRetry(), maybeCommitLock(catalog, config));
       var globalConfig = config.toIcebergConfigDefaults();
       if (!globalConfig.isEmpty()) {
         restCatalogAdapter = new RESTCatalogMiddlewareConfig(restCatalogAdapter, globalConfig);
@@ -319,6 +322,18 @@ private static Server createBaseServer(
       }
     }
 
+    logger.info(
+        "Commit retry config: numRetries={} minWaitMs={} maxWaitMs={} totalTimeoutMs={}",
+        config.commitRetry().numRetries(),
+        config.commitRetry().minWaitMs(),
+        config.commitRetry().maxWaitMs(),
+        config.commitRetry().totalTimeoutMs());
+    logger.info(
+        "Commit lock config: enabled={} leaseTtlSeconds={} acquireTimeoutMs={}",
+        config.commitLock().enabled(),
+        config.commitLock().leaseTtlSeconds(),
+        config.commitLock().acquireTimeoutMs());
+
     var h = new ServletHolder(new RESTCatalogServlet(restCatalogAdapter));
     mux.addServlet(h, "/*");
 
@@ -395,6 +410,22 @@ private static RESTCatalogAuthorizationHandler createAuthorizationHandler(
     return new RESTCatalogAuthorizationHandler(tokens, anonymousSession);
   }
 
+  /**
+   * Per-table etcd commit lock for {@link EtcdCatalog}; ignored when disabled or when not using
+   * etcd.
+   */
+  static CommitLock maybeCommitLock(Catalog catalog, Config config) {
+    if (!config.commitLock().enabled()) {
+      return null;
+    }
+    if (!(catalog instanceof EtcdCatalog etcd)) {
+      logger.warn(
+          "commitLock.enabled is true but catalog is not EtcdCatalog; commit lock disabled");
+      return null;
+    }
+    return new CommitLock(etcd.etcdClient(), catalog.name(), config.commitLock());
+  }
+
   private static void overrideJettyDefaults(Server s) {
     ServerConfig.setQuiet(s);
     s.setErrorHandler(new PlainErrorHandler());

ice-rest-catalog/src/main/java/com/altinity/ice/rest/catalog/internal/config/CommitLockConfig.java

@@ -0,0 +1,42 @@
+/*
+ * Copyright (c) 2025 Altinity Inc and/or its affiliates. All rights reserved.
+ *
+ * Licensed 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
+ */
+package com.altinity.ice.rest.catalog.internal.config;
+
+import com.fasterxml.jackson.annotation.JsonPropertyDescription;
+
+/**
+ * Optional per-table etcd commit lock for {@code ice-rest-catalog} when using the etcd metastore.
+ *
+ * <p>Serializes commits to the same table so concurrent writers do not lose optimistic concurrency
+ * races indefinitely.
+ */
+public record CommitLockConfig(
+    @JsonPropertyDescription(
+            "Enable etcd mutual-exclusion lock around table commits (etcd backend only; default false)")
+        boolean enabled,
+    @JsonPropertyDescription(
+            "Lease TTL for the lock in seconds (must exceed slow commits; default 30)")
+        long leaseTtlSeconds,
+    @JsonPropertyDescription("Max time to wait to acquire the lock in milliseconds (default 30000)")
+        long acquireTimeoutMs) {
+
+  public CommitLockConfig {
+    if (leaseTtlSeconds <= 0) {
+      leaseTtlSeconds = 30;
+    }
+    if (acquireTimeoutMs <= 0) {
+      acquireTimeoutMs = 30_000L;
+    }
+  }
+
+  public static CommitLockConfig defaults() {
+    return new CommitLockConfig(false, 30, 30_000L);
+  }
+}

ice-rest-catalog/src/main/java/com/altinity/ice/rest/catalog/internal/config/CommitRetryConfig.java

@@ -0,0 +1,56 @@
+/*
+ * Copyright (c) 2025 Altinity Inc and/or its affiliates. All rights reserved.
+ *
+ * Licensed 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
+ */
+package com.altinity.ice.rest.catalog.internal.config;
+
+import com.fasterxml.jackson.annotation.JsonPropertyDescription;
+import org.apache.iceberg.TableProperties;
+
+/**
+ * Server-side tuning for the REST catalog commit retry loop (OCC compare-and-swap failures).
+ *
+ * <p>Defaults match Iceberg's {@link TableProperties} commit retry defaults.
+ */
+public record CommitRetryConfig(
+    @JsonPropertyDescription(
+            "Number of retries on CommitFailedException (default: Iceberg commit.retry.num-retries = 4)")
+        int numRetries,
+    @JsonPropertyDescription(
+            "Minimum backoff between retries in ms (default: Iceberg commit.retry.min-wait-ms)")
+        long minWaitMs,
+    @JsonPropertyDescription(
+            "Maximum backoff between retries in ms (default: Iceberg commit.retry.max-wait-ms)")
+        long maxWaitMs,
+    @JsonPropertyDescription(
+            "Total time budget for the retry loop in ms (default: Iceberg commit.retry.total-timeout-ms)")
+        long totalTimeoutMs) {
+
+  public CommitRetryConfig {
+    if (numRetries <= 0) {
+      numRetries = TableProperties.COMMIT_NUM_RETRIES_DEFAULT;
+    }
+    if (minWaitMs <= 0) {
+      minWaitMs = TableProperties.COMMIT_MIN_RETRY_WAIT_MS_DEFAULT;
+    }
+    if (maxWaitMs <= 0) {
+      maxWaitMs = TableProperties.COMMIT_MAX_RETRY_WAIT_MS_DEFAULT;
+    }
+    if (totalTimeoutMs <= 0) {
+      totalTimeoutMs = TableProperties.COMMIT_TOTAL_RETRY_TIME_MS_DEFAULT;
+    }
+  }
+
+  public static CommitRetryConfig defaults() {
+    return new CommitRetryConfig(
+        TableProperties.COMMIT_NUM_RETRIES_DEFAULT,
+        TableProperties.COMMIT_MIN_RETRY_WAIT_MS_DEFAULT,
+        TableProperties.COMMIT_MAX_RETRY_WAIT_MS_DEFAULT,
+        TableProperties.COMMIT_TOTAL_RETRY_TIME_MS_DEFAULT);
+  }
+}

ice-rest-catalog/src/main/java/com/altinity/ice/rest/catalog/internal/config/Config.java

@@ -56,6 +56,12 @@ public record Config(
             "Maintenance schedule in https://github.com/shyiko/skedule?tab=readme-ov-file#format format, e.g. \"every day 00:00\". Empty schedule disables automatic maintenance (default)")
         String maintenanceSchedule,
     @JsonPropertyDescription("Maintenance config") MaintenanceConfig maintenance,
+    @JsonPropertyDescription(
+            "Server-side commit retry config; tune up for high-contention workloads (e.g., parallel `ice insert` to one table)")
+        CommitRetryConfig commitRetry,
+    @JsonPropertyDescription(
+            "Optional etcd per-table commit lock (etcd metastore only). Reduces CommitFailedException under concurrent writers.")
+        CommitLockConfig commitLock,
     @JsonPropertyDescription(
             "(experimental) Extra properties to include in loadTable REST response.")
         Map<String, String> loadTableProperties,
@@ -81,6 +87,8 @@ public Config(
       AnonymousAccess anonymousAccess,
       String maintenanceSchedule,
       MaintenanceConfig maintenance,
+      CommitRetryConfig commitRetry,
+      CommitLockConfig commitLock,
       Map<String, String> loadTableProperties,
       @JsonProperty("iceberg") Map<String, String> icebergProperties) {
     this.addr = Strings.orDefault(addr, DEFAULT_ADDR);
@@ -98,6 +106,8 @@ public Config(
     this.maintenance =
         Objects.requireNonNullElseGet(
             maintenance, () -> new MaintenanceConfig(null, 0, 0, 0, 0, 0, 0, null, false));
+    this.commitRetry = Objects.requireNonNullElse(commitRetry, CommitRetryConfig.defaults());
+    this.commitLock = Objects.requireNonNullElse(commitLock, CommitLockConfig.defaults());
     this.loadTableProperties = Objects.requireNonNullElse(loadTableProperties, Map.of());
     this.icebergProperties = Objects.requireNonNullElse(icebergProperties, Map.of());
   }