@Override
    public ProducerRecord<byte[], byte[]> serialize(RowData consumedRow, @Nullable Long timestamp) {
        // shortcut in case no input projection is required
        if (keySerialization == null && !hasMetadata) {
            final byte[] valueSerialized = serializeWithDirtyHandle(consumedRow,
                    DirtyType.VALUE_SERIALIZE_ERROR, valueSerialization);
            if (valueSerialized != null) {
                return new ProducerRecord<>(
                        getTargetTopic(consumedRow),
                        extractPartition(consumedRow, null, valueSerialized),
                        null,
                        valueSerialized);
            }
            return null;
        }
        final byte[] keySerialized;
        boolean mayDirtyData = false;
        if (keySerialization == null) {
            keySerialized = null;
        } else {
            final RowData keyRow = createProjectedRow(consumedRow, RowKind.INSERT, keyFieldGetters);
            keySerialized = serializeWithDirtyHandle(keyRow, DirtyType.KEY_SERIALIZE_ERROR, keySerialization);
            mayDirtyData = keySerialized == null;
        }

        final byte[] valueSerialized;
        final RowKind kind = consumedRow.getRowKind();
        final RowData valueRow = createProjectedRow(consumedRow, kind, valueFieldGetters);
        if (upsertMode) {
            if (kind == RowKind.DELETE || kind == RowKind.UPDATE_BEFORE) {
                // transform the message as the tombstone message
                valueSerialized = null;
            } else {
                // make the message to be INSERT to be compliant with the INSERT-ONLY format
                valueRow.setRowKind(RowKind.INSERT);
                valueSerialized = serializeWithDirtyHandle(valueRow,
                        DirtyType.VALUE_SERIALIZE_ERROR, valueSerialization);
                mayDirtyData = mayDirtyData || valueSerialized == null;
            }
        } else {
            valueSerialized = serializeWithDirtyHandle(valueRow, DirtyType.VALUE_SERIALIZE_ERROR, valueSerialization);
            mayDirtyData = mayDirtyData || valueSerialized == null;
        }
        if (mayDirtyData) {
            return null;
        }
        return new ProducerRecord<>(
                getTargetTopic(consumedRow),
                extractPartition(consumedRow, keySerialized, valueSerialized),
                readMetadata(consumedRow, KafkaDynamicSink.WritableMetadata.TIMESTAMP),
                keySerialized,
                valueSerialized,
                readMetadata(consumedRow, KafkaDynamicSink.WritableMetadata.HEADERS));
    }
Contains the position for each value of {@link WritableMetadata} in the
consumed row or -1 if this metadata key is not used.
inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/kafka/src/main/java/org/apache/inlong/sort/kafka/DynamicKafkaSerializationSchema.java