@Override
    public final synchronized void writeRecord(In row) throws IOException {
        checkFlushException();
        jsonDynamicSchemaFormat = (JsonDynamicSchemaFormat) DynamicSchemaFormatFactory.getFormat(sinkMultipleFormat);

        if (row instanceof RowData) {
            RowData rowData = (RowData) row;
            JsonNode rootNode = jsonDynamicSchemaFormat.deserialize(rowData.getBinary(0));
            String tableIdentifier;
            try {
                if (StringUtils.isBlank(schemaPattern)) {
                    tableIdentifier = StringUtils.join(
                            jsonDynamicSchemaFormat.parse(rootNode, databasePattern), ".",
                            jsonDynamicSchemaFormat.parse(rootNode, tablePattern));
                } else {
                    tableIdentifier = StringUtils.join(
                            jsonDynamicSchemaFormat.parse(rootNode, databasePattern), ".",
                            jsonDynamicSchemaFormat.parse(rootNode, schemaPattern), ".",
                            jsonDynamicSchemaFormat.parse(rootNode, tablePattern));
                }
            } catch (Exception e) {
                LOG.info("Cal tableIdentifier get Exception:", e);
                return;
            }

            GenericRowData record;
            try {
                RowType rowType = jsonDynamicSchemaFormat.extractSchema(rootNode);
                if (rowType != null) {
                    if (null != rowTypeMap.get(tableIdentifier)) {
                        if (!rowType.equals(rowTypeMap.get(tableIdentifier))) {
                            attemptFlush();
                            rowTypeMap.put(tableIdentifier, rowType);
                            updateOneExecutor(true, tableIdentifier);
                        }
                    } else {
                        rowTypeMap.put(tableIdentifier, rowType);
                    }
                }
                JsonNode physicalData = jsonDynamicSchemaFormat.getPhysicalData(rootNode);
                List<Map<String, String>> physicalDataList = jsonDynamicSchemaFormat.jsonNode2Map(physicalData);
                record = generateRecord(rowType, physicalDataList.get(0));
                List<RowKind> rowKinds = jsonDynamicSchemaFormat
                        .opType2RowKind(jsonDynamicSchemaFormat.getOpType(rootNode));
                record.setRowKind(rowKinds.get(rowKinds.size() - 1));
            } catch (Exception e) {
                LOG.warn("Extract schema failed", e);
                return;
            }
            try {
                recordsMap.computeIfAbsent(tableIdentifier, k -> new ArrayList<>())
                        .add(record);
                batchCount++;
                if (executionOptions.getBatchSize() > 0
                        && batchCount >= executionOptions.getBatchSize()) {
                    flush();
                }
            } catch (Exception e) {
                throw new IOException("Writing records to JDBC failed.", e);
            }
        }
    }
Extract and write record to recordsMap(buffer)

@param row The record to write.
inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/jdbc/src/main/java/org/apache/inlong/sort/jdbc/internal/JdbcMultiBatchingOutputFormat.java