@Override
        public void commit(TableMetadata base, TableMetadata updatedMetadata) {
            if (base != current) {
                throw new CommitFailedException("Cannot commit changes based on stale metadata");
            }
            synchronized (METADATA) {
                refresh();
                if (base == current) {
                    if (failCommits > 0) {
                        this.failCommits -= 1;
                        throw new CommitFailedException("Injected failure");
                    }
                    Integer version = VERSIONS.get(tableName);
                    // remove changes from the committed metadata
                    this.current = TableMetadata.buildFrom(updatedMetadata).discardChanges().build();
                    VERSIONS.put(tableName, version == null ? 0 : version + 1);
                    METADATA.put(tableName, current);
                } else {
                    throw new CommitFailedException(
                            "Commit failed: table was updated at %d", current.lastUpdatedMillis());
                }
            }
        }
Copy from iceberg-flink:iceberg-flink-1.15:1.1.x
inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/iceberg/src/test/java/org/apache/inlong/sort/iceberg/sink/util/TestTables.java