private byte[] serializeWithDirtyHandle(RowData consumedRow, DirtyType dirtyType,
            SerializationSchema<RowData> serialization) {
        if (!dirtyOptions.ignoreDirty()) {
            return serialization.serialize(consumedRow);
        }
        byte[] value = null;
        try {
            value = serialization.serialize(consumedRow);
        } catch (Exception e) {
            LOG.error(String.format("serialize error, raw data: %s", consumedRow.toString()), e);
            if (dirtySink != null) {
                DirtyData.Builder<Object> builder = DirtyData.builder();
                try {
                    builder.setData(consumedRow)
                            .setDirtyType(dirtyType)
                            .setLabels(dirtyOptions.getLabels())
                            .setLogTag(dirtyOptions.getLogTag())
                            .setDirtyMessage(e.getMessage())
                            .setIdentifier(dirtyOptions.getIdentifier());
                    dirtySink.invoke(builder.build());
                } catch (Exception ex) {
                    if (!dirtyOptions.ignoreSideOutputErrors()) {
                        throw new RuntimeException(ex);
                    }
                    LOG.warn("Dirty sink failed", ex);
                }
            }
            metricData.invokeDirtyWithEstimate(consumedRow);
        }
        return value;
    }
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