public List<ProducerRecord<byte[], byte[]>> serializeForList(RowData consumedRow, @Nullable Long timestamp) {
        List<ProducerRecord<byte[], byte[]>> values = new ArrayList<>();
        if (!multipleSink) {
            ProducerRecord<byte[], byte[]> value = serialize(consumedRow, timestamp);
            if (value != null) {
                values.add(value);
            }
            return values;
        }
        String topic = null;
        try {
            JsonNode rootNode = jsonDynamicSchemaFormat.deserialize(consumedRow.getBinary(0));
            boolean isDDL = jsonDynamicSchemaFormat.extractDDLFlag(rootNode);
            if (isDDL) {
                Set<SchemaChangeType> types = extractSchemaChangeType(rootNode);
                if (types.isEmpty()) {
                    return values;
                }

                Set<SchemaChangeType> enableTypes = types.stream()
                        .filter((type) -> SchemaChangePolicy.ENABLE.equals(policyMap.get(type)))
                        .collect(Collectors.toSet());

                if (enableTypes.isEmpty()) {
                    return values;
                }

                values.add(new ProducerRecord<>(
                        jsonDynamicSchemaFormat.parse(rootNode, topicPattern),
                        extractPartition(consumedRow, null, consumedRow.getBinary(0)),
                        null,
                        consumedRow.getBinary(0)));
                return values;
            }
            JsonNode updateBeforeNode = jsonDynamicSchemaFormat.getUpdateBefore(rootNode);
            JsonNode updateAfterNode = jsonDynamicSchemaFormat.getUpdateAfter(rootNode);
            if (!splitRequired(updateBeforeNode, updateAfterNode)) {
                topic = jsonDynamicSchemaFormat.parse(rootNode, topicPattern);
                values.add(new ProducerRecord<>(
                        topic, extractPartition(consumedRow, null, consumedRow.getBinary(0)),
                        null, consumedRow.getBinary(0)));
            } else {
                split2JsonArray(rootNode, updateBeforeNode, updateAfterNode, values);
            }
        } catch (Exception e) {
            LOG.error(String.format("serialize error, raw data: %s", new String(consumedRow.getBinary(0))), e);
            if (!dirtyOptions.ignoreDirty()) {
                throw new RuntimeException(e);
            }
            if (dirtySink != null) {
                DirtyData.Builder<Object> builder = DirtyData.builder();
                try {
                    builder.setData(new String(consumedRow.getBinary(0)))
                            .setDirtyType(DirtyType.VALUE_DESERIALIZE_ERROR)
                            .setLabels(dirtyOptions.getLabels())
                            .setLogTag(dirtyOptions.getLogTag())
                            .setIdentifier(dirtyOptions.getIdentifier());
                    dirtySink.invoke(builder.build());
                } catch (Exception ex) {
                    if (!dirtyOptions.ignoreSideOutputErrors()) {
                        throw new RuntimeException(ex);
                    }
                    LOG.warn("Dirty sink failed", ex);
                }
            }
            metricData.sendDirtyMetrics(topic, 1, consumedRow.getBinary(0).length);
        }
        return values;
    }
Serialize for list it is used for multiple sink scenes when a record contains mulitple real records.

@param consumedRow The consumeRow
@param timestamp The timestamp
@return List of ProducerRecord
inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/kafka/src/main/java/org/apache/inlong/sort/kafka/DynamicKafkaSerializationSchema.java