@Override
    RowDataDeltaWriter route(RowData row) {
        partitionKey.partition(wrapper().wrap(row));

        RowDataDeltaWriter writer = writers.get(partitionKey);
        if (writer == null) {
            // NOTICE: we need to copy a new partition key here, in case of messing up the keys in writers.
            PartitionKey copiedKey = partitionKey.copy();
            writer = new RowDataDeltaWriter(copiedKey);
            writers.put(copiedKey, writer);
        }

        return writer;
    }
Copy from iceberg-flink:iceberg-flink-1.13:0.13.2
inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/iceberg/src/main/java/org/apache/inlong/sort/iceberg/sink/PartitionedDeltaWriter.java