Repository: apache/iceberg (language: python)

Issue:
Bad Table Properties cause commit failure
### Apache Iceberg version

1.6.1 (latest release)

### Query engine

Spark

### Please describe the bug 🐞

Today iceberg does not validate the type of table properties values on and this can be problematic if those table properties value are used for commit. 

Example setup
```
                sql(
                    "CREATE TABLE foo.bar "
                        + "(id BIGINT NOT NULL, data STRING) "
                        + "USING iceberg "
                        + "TBLPROPERTIES ('commit.retry.num-retries'='x', p2='x')",
```

See the value for `commit.retry.num-retries` is accidentally set to some non-integer value, we will unable to rectify this through either spark SQL or iceberg API 

```
table.updateProperties.remove("commit.retry.max-wait-ms").apply.commit

table.updateProperties.remove("commit.retry.max-wait-ms").commit
java.lang.NumberFormatException: For input string: "commit.retry.max-wait-ms"
  at java.base/java.lang.NumberFormatException.forInputString(NumberFormatException.java:67)
  at java.base/java.lang.Integer.parseInt(Integer.java:668)
  at java.base/java.lang.Integer.parseInt(Integer.java:786)
  at org.apache.iceberg.util.PropertyUtil.propertyAsInt(PropertyUtil.java:64)
  at org.apache.iceberg.TableMetadata.propertyAsInt(TableMetadata.java:472)
  at org.apache.iceberg.PropertiesUpdate.commit(PropertiesUpdate.java:105)
  ... 48 elided
```

After some look, I believe this block of code might be the problem: https://github.com/apache/iceberg/blob/main/core/src/main/java/org/apache/iceberg/PropertiesUpdate.java#L100-L114, where  propertyAsInt will throw NumberFormatException before commit can proceed to rectify or remove incorrectly set table properties. 


Current workaround is move back to previous table metadata before the table properties change, but I plan to contribute the patch to fix the problem in 2 ways
1. add validation for new table where commit related table properties need to have value type checked (as integer)
2. relax the condition in `PropertiesUpdate` class to allow update when existing value is corrupted, this help with existing table to move forward. 



### Willingness to contribute

- [X] I can contribute a fix for this bug independently
- [ ] I would be willing to contribute a fix for this bug with guidance from the Iceberg community
- [ ] I cannot contribute a fix for this bug at this time


Discussion / hints:

Fix patch:
diff --git a/core/src/main/java/org/apache/iceberg/PropertiesUpdate.java b/core/src/main/java/org/apache/iceberg/PropertiesUpdate.java
index 35338a689205..9389aec50c0a 100644
--- a/core/src/main/java/org/apache/iceberg/PropertiesUpdate.java
+++ b/core/src/main/java/org/apache/iceberg/PropertiesUpdate.java
@@ -98,12 +98,13 @@ public Map<String, String> apply() {
 
   @Override
   public void commit() {
+    // If existing table commit properties in base are corrupted, allow rectification
     Tasks.foreach(ops)
-        .retry(base.propertyAsInt(COMMIT_NUM_RETRIES, COMMIT_NUM_RETRIES_DEFAULT))
+        .retry(base.propertyTryAsInt(COMMIT_NUM_RETRIES, COMMIT_NUM_RETRIES_DEFAULT))
         .exponentialBackoff(
-            base.propertyAsInt(COMMIT_MIN_RETRY_WAIT_MS, COMMIT_MIN_RETRY_WAIT_MS_DEFAULT),
-            base.propertyAsInt(COMMIT_MAX_RETRY_WAIT_MS, COMMIT_MAX_RETRY_WAIT_MS_DEFAULT),
-            base.propertyAsInt(COMMIT_TOTAL_RETRY_TIME_MS, COMMIT_TOTAL_RETRY_TIME_MS_DEFAULT),
+            base.propertyTryAsInt(COMMIT_MIN_RETRY_WAIT_MS, COMMIT_MIN_RETRY_WAIT_MS_DEFAULT),
+            base.propertyTryAsInt(COMMIT_MAX_RETRY_WAIT_MS, COMMIT_MAX_RETRY_WAIT_MS_DEFAULT),
+            base.propertyTryAsInt(COMMIT_TOTAL_RETRY_TIME_MS, COMMIT_TOTAL_RETRY_TIME_MS_DEFAULT),
             2.0 /* exponential */)
         .onlyRetryOn(CommitFailedException.class)
         .run(
diff --git a/core/src/main/java/org/apache/iceberg/TableMetadata.java b/core/src/main/java/org/apache/iceberg/TableMetadata.java
index d20dd59d2b97..3cdc53995dce 100644
--- a/core/src/main/java/org/apache/iceberg/TableMetadata.java
+++ b/core/src/main/java/org/apache/iceberg/TableMetadata.java
@@ -134,6 +134,8 @@ static TableMetadata newTableMetadata(
     // break existing tables.
     MetricsConfig.fromProperties(properties).validateReferencedColumns(schema);
 
+    PropertyUtil.validateCommitProperties(properties);
+
     return new Builder()
         .setInitialFormatVersion(formatVersion)
         .setCurrentSchema(freshSchema, lastColumnId.get())
@@ -486,6 +488,10 @@ public int propertyAsInt(String property, int defaultValue) {
     return PropertyUtil.propertyAsInt(properties, property, defaultValue);
   }
 
+  public int propertyTryAsInt(String property, int defaultValue) {
+    return PropertyUtil.propertyTryAsInt(properties, property, defaultValue);
+  }
+
   public long propertyAsLong(String property, long defaultValue) {
     return PropertyUtil.propertyAsLong(properties, property, defaultValue);
   }
diff --git a/core/src/main/java/org/apache/iceberg/util/PropertyUtil.java b/core/src/main/java/org/apache/iceberg/util/PropertyUtil.java
index 68c8f3e9efda..633b0a6ae739 100644
--- a/core/src/main/java/org/apache/iceberg/util/PropertyUtil.java
+++ b/core/src/main/java/org/apache/iceberg/util/PropertyUtil.java
@@ -24,10 +24,23 @@
 import java.util.Set;
 import java.util.function.Predicate;
 import java.util.stream.Collectors;
+import org.apache.iceberg.TableProperties;
+import org.apache.iceberg.exceptions.ValidationException;
 import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableSet;
 import org.apache.iceberg.relocated.com.google.common.collect.Maps;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 public class PropertyUtil {
+  private static final Logger LOG = LoggerFactory.getLogger(PropertyUtil.class);
+
+  private static final Set<String> COMMIT_PROPERTIES =
+      ImmutableSet.of(
+          TableProperties.COMMIT_NUM_RETRIES,
+          TableProperties.COMMIT_MIN_RETRY_WAIT_MS,
+          TableProperties.COMMIT_MAX_RETRY_WAIT_MS,
+          TableProperties.COMMIT_TOTAL_RETRY_TIME_MS);
 
   private PropertyUtil() {}
 
@@ -57,6 +70,20 @@ public static double propertyAsDouble(
     return defaultValue;
   }
 
+  public static int propertyTryAsInt(
+      Map<String, String> properties, String property, int defaultValue) {
+    String value = properties.get(property);
+    if (value == null) {
+      return defaultValue;
+    }
+    try {
+      return Integer.parseInt(value);
+    } catch (NumberFormatException e) {
+      LOG.warn("Failed to parse value of {} as integer, default to {}", property, defaultValue, e);
+      return defaultValue;
+    }
+  }
+
   public static int propertyAsInt(
       Map<String, String> properties, String property, int defaultValue) {
     String value = properties.get(property);
@@ -100,6 +127,29 @@ public static String propertyAsString(
     return defaultValue;
   }
 
+  /**
+   * Validate the table commit related properties to have non-negative integer on table creation to
+   * prevent commit failure
+   */
+  public static void validateCommitProperties(Map<String, String> properties) {
+    for (String commitProperty : COMMIT_PROPERTIES) {
+      String value = properties.get(commitProperty);
+      if (value != null) {
+        int parsedValue;
+        try {
+          parsedValue = Integer.parseInt(value);
+        } catch (NumberFormatException e) {
+          throw new ValidationException(
+              "Table property %s must have integer value", commitProperty);
+        }
+        ValidationException.check(
+            parsedValue >= 0,
+            "Table property %s must have non negative integer value",
+            commitProperty);
+      }
+    }
+  }
+
   /**
    * Returns subset of provided map with keys matching the provided prefix. Matching is
    * case-sensitive and the matching prefix is removed from the keys in returned map.


Test patch:
diff --git a/core/src/test/java/org/apache/iceberg/TestTransaction.java b/core/src/test/java/org/apache/iceberg/TestTransaction.java
index 8fed7134fae1..8770e24f8e40 100644
--- a/core/src/test/java/org/apache/iceberg/TestTransaction.java
+++ b/core/src/test/java/org/apache/iceberg/TestTransaction.java
@@ -714,4 +714,22 @@ public void testTransactionRecommit() {
     assertThat(paths).isEqualTo(expectedPaths);
     assertThat(table.currentSnapshot().allManifests(table.io())).hasSize(2);
   }
+
+  @TestTemplate
+  public void testCommitProperties() {
+    table
+        .updateProperties()
+        .set(TableProperties.COMMIT_MAX_RETRY_WAIT_MS, "foo")
+        .set(TableProperties.COMMIT_NUM_RETRIES, "bar")
+        .set(TableProperties.COMMIT_TOTAL_RETRY_TIME_MS, Integer.toString(60 * 60 * 1000))
+        .commit();
+    table.updateProperties().remove(TableProperties.COMMIT_MAX_RETRY_WAIT_MS).commit();
+    table.updateProperties().remove(TableProperties.COMMIT_NUM_RETRIES).commit();
+
+    assertThat(table.properties())
+        .doesNotContainKey(TableProperties.COMMIT_NUM_RETRIES)
+        .doesNotContainKey(TableProperties.COMMIT_MAX_RETRY_WAIT_MS)
+        .containsEntry(
+            TableProperties.COMMIT_TOTAL_RETRY_TIME_MS, Integer.toString(60 * 60 * 1000));
+  }
 }
diff --git a/spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/sql/TestCreateTable.java b/spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/sql/TestCreateTable.java
index ae0aa2cda49b..11d4cfebfea6 100644
--- a/spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/sql/TestCreateTable.java
+++ b/spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/sql/TestCreateTable.java
@@ -31,6 +31,7 @@
 import org.apache.iceberg.Table;
 import org.apache.iceberg.TableOperations;
 import org.apache.iceberg.TableProperties;
+import org.apache.iceberg.exceptions.ValidationException;
 import org.apache.iceberg.hadoop.HadoopCatalog;
 import org.apache.iceberg.spark.CatalogTestBase;
 import org.apache.iceberg.types.Types;
@@ -348,6 +349,47 @@ public void testCreateTableProperties() {
     assertThat(table.properties()).containsEntry("p1", "2").containsEntry("p2", "x");
   }
 
+  @TestTemplate
+  public void testCreateTableCommitProperties() {
+    assertThat(validationCatalog.tableExists(tableIdent))
+        .as("Table should not already exist")
+        .isFalse();
+
+    assertThatThrownBy(
+            () ->
+                sql(
+                    "CREATE TABLE %s "
+                        + "(id BIGINT NOT NULL, data STRING) "
+                        + "USING iceberg "
+                        + "TBLPROPERTIES ('commit.retry.num-retries'='x', p2='x')",
+                    tableName))
+        .isInstanceOf(ValidationException.class)
+        .hasMessage("Table property commit.retry.num-retries must have integer value");
+
+    assertThatThrownBy(
+            () ->
+                sql(
+                    "CREATE TABLE %s "
+                        + "(id BIGINT NOT NULL, data STRING) "
+                        + "USING iceberg "
+                        + "TBLPROPERTIES ('commit.retry.max-wait-ms'='-1')",
+                    tableName))
+        .isInstanceOf(ValidationException.class)
+        .hasMessage("Table property commit.retry.max-wait-ms must have non negative integer value");
+
+    sql(
+        "CREATE TABLE %s "
+            + "(id BIGINT NOT NULL, data STRING) "
+            + "USING iceberg "
+            + "TBLPROPERTIES ('commit.retry.num-retries'='1', 'commit.retry.max-wait-ms'='3000')",
+        tableName);
+
+    Table table = validationCatalog.loadTable(tableIdent);
+    assertThat(table.properties())
+        .containsEntry(TableProperties.COMMIT_NUM_RETRIES, "1")
+        .containsEntry(TableProperties.COMMIT_MAX_RETRY_WAIT_MS, "3000");
+  }
+
   @TestTemplate
   public void testCreateTableWithFormatV2ThroughTableProperty() {
     assertThat(validationCatalog.tableExists(tableIdent))