@Override
    public String serializeRow(SeaTunnelRow row) {
        switch (row.getRowKind()) {
            case INSERT:
            case UPDATE_AFTER:
                return serializeUpsert(row);
            case UPDATE_BEFORE:
            case DELETE:
                return serializeDelete(row);
            default:
                throw new ElasticsearchConnectorException(
                        CommonErrorCodeDeprecated.UNSUPPORTED_OPERATION,
                        "Unsupported write row kind: " + row.getRowKind());
        }
    }
@Test
    public void testSerializeUpsertWithoutKey() {
        String index = "st_index";
        Map<String, Object> confMap = new HashMap<>();
        confMap.put(SinkConfig.INDEX.key(), index);

        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[] {"id", "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 =
                "{ \"index\" :{\"_index\":\""
                        + index
                        + "\"} }\n"
                        + "{\"name\":\""
                        + name
                        + "\",\"id\":\""
                        + id
                        + "\"}";

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