nebula-flink-connector中NebulaSinkFunction无法识别FlinkCDC的Source中Row的RowKind的类型

[fanqiejiang8]

我自己应用程序中的代码如下：
VertexExecutionOptions executionOptions = new VertexExecutionOptions.ExecutionOptionBuilder() .setGraphSpace(ennVertexConfig.getGraphSpace()) .setTag(ennVertexConfig.getTagName()) .setIdIndex(ennVertexConfig.getIndex())//vid .setFields(ennVertexConfig.getFields()) .setWriteMode(WriteModeEnum.INSERT) .setPositions(ennVertexConfig.getPositions()) .setBatchSize(ennVertexConfig.getBatchSize()) .build(); NebulaVertexBatchOutputFormat outputFormat = new NebulaVertexBatchOutputFormat(graphConnectionProvider, metaConnectionProvider, executionOptions);
从这里我们可以看出来，在创建NebulaVertexBatchOutputFormat的时候就把WriteMode给确定了，但是对于FlinkCDC Source的数据，只有在反序列化成Row格式之后才能知道RowKind的类型，所以当使用 FlinkCDC的MysqSqlSource + NebulaSinkFunction的时候是无法处理删除的数据。
我们从源码中也能看出来，下面是源码：
`
@OverRide
public void addToBatch(Row record) {
NebulaEdge edge = converter.createEdge(record, executionOptions.getPolicy());
if (edge == null) {
return;
}
nebulaEdgeList.add(edge);
}

@OverRide
public void executeBatch(Session session) throws IOException {
if (isBatchEmpty()) {
return;
}
NebulaEdges nebulaEdges = new NebulaEdges(executionOptions.getLabel(),
executionOptions.getFields(), nebulaEdgeList, executionOptions.getPolicy(),
executionOptions.getPolicy());
// generate the write ngql statement
String statement = null;
switch (executionOptions.getWriteMode()) {
case INSERT:
statement = nebulaEdges.getInsertStatement();
break;
case UPDATE:
statement = nebulaEdges.getUpdateStatement();
break;
case DELETE:
statement = nebulaEdges.getDeleteStatement();
break;
default:
throw new IllegalArgumentException("write mode is not supported");
}
executeStatement(session, statement);
clearBatch();
}
`
然后我自己测试的结果如下：

[fanqiejiang8]

补充一下：FlinkCDC的MysqSqlSource + NebulaSinkFunction的时候是无法处理删除的数据。