From 883a3bbba5c3878377ec50fe462cf5feb7fc38ad Mon Sep 17 00:00:00 2001
From: litiliu <38579068+litiliu@users.noreply.github.com>
Date: Thu, 3 Sep 2026 15:55:59 +0800
Subject: [PATCH] [lake/iceberg] Ignore Fluss properties in compatibility
 checks

---
 .../org/apache/fluss/lake/iceberg/IcebergLakeCatalog.java    | 3 +++
 .../apache/fluss/lake/iceberg/IcebergLakeCatalogTest.java    | 5 +++++
 2 files changed, 8 insertions(+)

diff --git a/fluss-lake/fluss-lake-iceberg/src/main/java/org/apache/fluss/lake/iceberg/IcebergLakeCatalog.java b/fluss-lake/fluss-lake-iceberg/src/main/java/org/apache/fluss/lake/iceberg/IcebergLakeCatalog.java
index 70886c9c839..2170696721f 100644
--- a/fluss-lake/fluss-lake-iceberg/src/main/java/org/apache/fluss/lake/iceberg/IcebergLakeCatalog.java
+++ b/fluss-lake/fluss-lake-iceberg/src/main/java/org/apache/fluss/lake/iceberg/IcebergLakeCatalog.java
@@ -519,6 +519,9 @@ boolean isIcebergSortOrderCompatible(
     boolean isIcebergPropertiesCompatible(
             Map<String, String> existingProperties, Map<String, String> expectedProperties) {
         for (Map.Entry<String, String> entry : expectedProperties.entrySet()) {
+            if (entry.getKey().startsWith(FLUSS_CONF_PREFIX)) {
+                continue;
+            }
             String actual = existingProperties.get(entry.getKey());
             if (actual == null || !actual.equals(entry.getValue())) {
                 return false;
diff --git a/fluss-lake/fluss-lake-iceberg/src/test/java/org/apache/fluss/lake/iceberg/IcebergLakeCatalogTest.java b/fluss-lake/fluss-lake-iceberg/src/test/java/org/apache/fluss/lake/iceberg/IcebergLakeCatalogTest.java
index 868a6bfd47b..daf82218f02 100644
--- a/fluss-lake/fluss-lake-iceberg/src/test/java/org/apache/fluss/lake/iceberg/IcebergLakeCatalogTest.java
+++ b/fluss-lake/fluss-lake-iceberg/src/test/java/org/apache/fluss/lake/iceberg/IcebergLakeCatalogTest.java
@@ -1775,6 +1775,11 @@ void testIsIcebergPropertiesCompatible() {
         existing.put("engine.internal", "whatever");
         assertThat(flussIcebergCatalog.isIcebergPropertiesCompatible(existing, expected)).isTrue();
 
+        // Fluss properties are metadata owned by Fluss and may differ between descriptors.
+        expected.put("fluss.table.replication.factor", "1");
+        existing.put("fluss.table.replication.factor", "2");
+        assertThat(flussIcebergCatalog.isIcebergPropertiesCompatible(existing, expected)).isTrue();
+
         // Existing missing an expected key -> incompatible.
         existing = new HashMap<>();
         existing.put("k1", "v1");