private String serializeUpsert(SeaTunnelRow row) {
        String key = keyExtractor.apply(row);
        Map<String, Object> document = toDocumentMap(row, seaTunnelRowType);
        String documentStr;

        try {
            documentStr = objectMapper.writeValueAsString(document);
        } catch (JsonProcessingException e) {
            throw CommonError.jsonOperationError(
                    "Elasticsearch", "document:" + document.toString(), e);
        }

        if (key != null) {
            Map<String, String> upsertMetadata = createMetadata(row, key);
            String upsertMetadataStr;
            try {
                upsertMetadataStr = objectMapper.writeValueAsString(upsertMetadata);
            } catch (JsonProcessingException e) {
                throw CommonError.jsonOperationError(
                        "Elasticsearch", "upsertMetadata:" + upsertMetadata.toString(), e);
            }

            /**
             * format example: { "update" : {"_index" : "${your_index}", "_id" :
             * "${your_document_id}"} }\n { "doc" : ${your_document_json}, "doc_as_upsert" : true }
             */
            return new StringBuilder()
                    .append("{ \"update\" :")
                    .append(upsertMetadataStr)
                    .append(" }")
                    .append("\n")
                    .append("{ \"doc\" :")
                    .append(documentStr)
                    .append(", \"doc_as_upsert\" : true }")
                    .toString();
        }

        Map<String, String> indexMetadata = createMetadata(row);
        String indexMetadataStr;
        try {
            indexMetadataStr = objectMapper.writeValueAsString(indexMetadata);
        } catch (JsonProcessingException e) {
            throw CommonError.jsonOperationError(
                    "Elasticsearch", "indexMetadata:" + indexMetadata.toString(), e);
        }

        /**
         * format example: { "index" : {"_index" : "${your_index}", "_id" : "${your_document_id}"}
         * }\n ${your_document_json}
         */
        return new StringBuilder()
                .append("{ \"index\" :")
                .append(indexMetadataStr)
                .append(" }")
                .append("\n")
                .append(documentStr)
                .toString();
    }
@Test
    public void testSerializeUpsert() {
        String index = "st_index";
        String primaryKey = "id";
        Map<String, Object> confMap = new HashMap<>();
        confMap.put(SinkConfig.INDEX.key(), index);
        confMap.put(SinkConfig.PRIMARY_KEYS.key(), Arrays.asList(primaryKey));

        ReadonlyConfig pluginConf = ReadonlyConfig.fromMap(confMap);
        ElasticsearchClusterInfo clusterInfo =
                ElasticsearchClusterInfo.builder().clusterVersion("8.0.0").build();
        IndexInfo indexInfo = new IndexInfo(index, pluginConf);
        SeaTunnelRowType schema =
                new SeaTunnelRowType(
                        new String[] {primaryKey, "name"},
                        new SeaTunnelDataType[] {STRING_TYPE, STRING_TYPE});

        final ElasticsearchRowSerializer serializer =
                new ElasticsearchRowSerializer(clusterInfo, indexInfo, schema);

        String id = "0001";
        String name = "jack";
        SeaTunnelRow row = new SeaTunnelRow(new Object[] {id, name});
        row.setRowKind(RowKind.UPDATE_AFTER);

        String expected =
                "{ \"update\" :{\"_index\":\""
                        + index
                        + "\",\"_id\":\""
                        + id
                        + "\"} }\n"
                        + "{ \"doc\" :{\"name\":\""
                        + name
                        + "\",\"id\":\""
                        + id
                        + "\"}, \"doc_as_upsert\" : true }";

        String upsertStr = serializer.serializeRow(row);
        Assertions.assertEquals(expected, upsertStr);
    }