private String serializeDelete(SeaTunnelRow row) {
        String key = keyExtractor.apply(row);
        Map<String, String> deleteMetadata = createMetadata(row, key);
        String deleteMetadataStr;
        try {
            deleteMetadataStr = objectMapper.writeValueAsString(deleteMetadata);
        } catch (JsonProcessingException e) {
            throw CommonError.jsonOperationError(
                    "Elasticsearch", "deleteMetadata:" + deleteMetadata.toString(), e);
        }

        /**
         * format example: { "delete" : {"_index" : "${your_index}", "_id" : "${your_document_id}"}
         * }
         */
        return new StringBuilder()
                .append("{ \"delete\" :")
                .append(deleteMetadataStr)
                .append(" }")
                .toString();
    }
@Test
    public void testSerializeDelete() {
        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.DELETE);

        String expected = "{ \"delete\" :{\"_index\":\"" + index + "\",\"_id\":\"" + id + "\"} }";

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