@VisibleForTesting
        List<Integer> checkAndGetEqualityFieldIds() {
            List<Integer> equalityFieldIds = Lists.newArrayList(table.schema().identifierFieldIds());
            if (equalityFieldColumns != null && equalityFieldColumns.size() > 0) {
                Set<Integer> equalityFieldSet =
                        Sets.newHashSetWithExpectedSize(equalityFieldColumns.size());
                for (String column : equalityFieldColumns) {
                    org.apache.iceberg.types.Types.NestedField field = table.schema().findField(column);
                    Preconditions.checkNotNull(
                            field,
                            "Missing required equality field column '%s' in table schema %s",
                            column,
                            table.schema());
                    equalityFieldSet.add(field.fieldId());
                }

                if (!equalityFieldSet.equals(table.schema().identifierFieldIds())) {
                    LOG.warn(
                            "The configured equality field column IDs {} are not matched with the schema identifier "
                                    + "field IDs {}, use job specified equality field columns as the equality fields "
                                    + "by default.",
                            equalityFieldSet,
                            table.schema().identifierFieldIds());
                }
                equalityFieldIds = Lists.newArrayList(equalityFieldSet);
            }
            LOG.info("equalityFieldIds in iceberg: {}", equalityFieldIds);
            return equalityFieldIds;
        }
Append the iceberg sink operators to write records to iceberg table.

@return {@link DataStreamSink} for sink.
inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/iceberg/src/main/java/org/apache/inlong/sort/iceberg/sink/FlinkSink.java