private void serializeWithDirtyHandle(Map<String, Object> baseMap, JsonNode rootNode,
            JsonNode dataNode, List<ProducerRecord<byte[], byte[]>> values) {
        String topic = null;
        try {
            byte[] data = JsonDynamicSchemaFormat.OBJECT_MAPPER.writeValueAsBytes(baseMap);
            topic = jsonDynamicSchemaFormat.parse(rootNode, topicPattern);
            values.add(new ProducerRecord<>(topic,
                    extractPartition(null, null, data), null, data));
        } catch (Exception e) {
            LOG.error(String.format("serialize error, raw data: %s", baseMap), e);
            if (!dirtyOptions.ignoreDirty()) {
                throw new RuntimeException(e);
            }
            if (dirtySink != null) {
                DirtyData.Builder<Object> builder = DirtyData.builder();
                try {
                    builder.setData(dataNode)
                            .setDirtyType(DirtyType.VALUE_DESERIALIZE_ERROR)
                            .setLabels(jsonDynamicSchemaFormat.parse(rootNode, dirtyOptions.getLabels()))
                            .setLogTag(jsonDynamicSchemaFormat.parse(rootNode, dirtyOptions.getLogTag()))
                            .setDirtyMessage(e.getMessage())
                            .setIdentifier(jsonDynamicSchemaFormat.parse(rootNode, dirtyOptions.getIdentifier()));
                    dirtySink.invoke(builder.build());
                } catch (Exception ex) {
                    if (!dirtyOptions.ignoreSideOutputErrors()) {
                        throw new RuntimeException(ex);
                    }
                    LOG.warn("Dirty sink failed", ex);
                }
            }
            metricData.sendOutMetrics(topic, 1, dataNode.toString().getBytes(StandardCharsets.UTF_8).length);
        }
    }
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