private void split2JsonArray(JsonNode rootNode,
            JsonNode updateBeforeNode, JsonNode updateAfterNode, List<ProducerRecord<byte[], byte[]>> values) {
        Iterator<Entry<String, JsonNode>> iterator = rootNode.fields();
        Map<String, Object> baseMap = new LinkedHashMap<>();
        String updateBeforeKey = null;
        String updateAfterKey = null;
        while (iterator.hasNext()) {
            Entry<String, JsonNode> kv = iterator.next();
            if (kv.getValue() == null || (!kv.getValue().equals(updateBeforeNode) && !kv.getValue()
                    .equals(updateAfterNode))) {
                baseMap.put(kv.getKey(), kv.getValue());
                continue;
            }
            if (kv.getValue().equals(updateAfterNode)) {
                updateAfterKey = kv.getKey();
            } else if (kv.getValue().equals(updateBeforeNode)) {
                updateBeforeKey = kv.getKey();
            }
        }
        if (updateAfterNode != null) {
            for (int i = 0; i < updateAfterNode.size(); i++) {
                baseMap.put(updateAfterKey, Collections.singletonList(updateAfterNode.get(i)));
                if (updateBeforeNode != null && updateBeforeNode.size() > i) {
                    baseMap.put(updateBeforeKey, Collections.singletonList(updateBeforeNode.get(i)));
                } else if (updateBeforeKey != null) {
                    baseMap.remove(updateBeforeKey);
                }
                serializeWithDirtyHandle(baseMap, rootNode, updateAfterNode.get(i), values);
            }
        } else {
            // In general, it will not run to this branch
            for (int i = 0; i < updateBeforeNode.size(); i++) {
                baseMap.put(updateBeforeKey, Collections.singletonList(updateBeforeNode.get(i)));
                serializeWithDirtyHandle(baseMap, rootNode, updateBeforeNode.get(i), 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