@Override
    public ChangelogMode getChangelogMode(ChangelogMode requestedMode) {
        if (org.apache.flink.configuration.Configuration.fromMap(catalogTable.getOptions())
                .get(IGNORE_ALL_CHANGELOG)) {
            LOG.warn("Iceberg sink receive all changelog record. "
                    + "Regard any other record as insert-only record.");
            return ChangelogMode.all();
        } else {
            ChangelogMode.Builder builder = ChangelogMode.newBuilder();
            for (RowKind kind : requestedMode.getContainedKinds()) {
                builder.addContainedKind(kind);
            }
            return builder.build();
        }
    }
Copy from iceberg-flink:iceberg-flink-1.13:0.13.2
Add an option `sink.ignore.changelog` to support insert-only mode without primaryKey.
Add a table property `write.compact.enable` to support small file compact.
Add option `inlong.metric` and `metrics.audit.proxy.hosts` to support collect inlong metrics and audit
inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/iceberg/src/main/java/org/apache/inlong/sort/iceberg/IcebergTableSink.java