[mongodb] Allow mongo ARRAY to be converted to string type in Flink

[vanliu-tx]

failed to convert BsonArray to String type in Flink

2022-08-11 17:57:06.023 [Source: mongodb source datamanager.bcsdatamanager_cluster (1/1)#0] WARN  org.apache.flink.runtime.taskmanager.Task  - Source: mongodb source datamanager.bcsdatamanager_cluster (1/1)#0 (310561d264d8435431a809240a2e1802) switched from RUNNING to FAILED with failure cause: java.lang.IllegalArgumentException: Unable to convert to string from unexpected value 'BsonArray{values=[{"index": 30, "time": {"$date": "2022-04-29T06:30:00Z"}, "nodecount": 0, "availablenodecount": 0, "minnode": {"name": "MinNode", "metricname": "MinNode", "value": 0.0, "period": "2022-04-29 14:30:00 +0800 CST", "xxx_nounkeyedliteral": {}, "xxx_unrecognized": null, "xxx_sizecache": 0}, "maxnode": {"name": "MaxNode", "metricname": "MaxNode", "value": 0.0, "period": "2022-04-29 14:30:00 +0800 CST", "xxx_nounkeyedliteral": {}, "xxx_unrecognized": null, "xxx_sizecache": 0}, "nodequantile": [], "minusagenode": "", "totalcpu": 4272.0, "totalmemory": 11978108223488, "totalloadcpu": 83.1133370062609, "totalloadmemory": 2845860257792, "avgloadcpu": 0.0, "avgloadmemory": 0, "cpuusage": 0.019455369149405644, "memoryusage": 0.23758845760063532, "workloadcount": 136, "instancecount": 1774, "mininstance": {"name": "MinInstance", "metricname": "MinInstance", "value": 1774.0, "period": "2022-04-29 14:30:00 +0800 CST", "xxx_nounkeyedliteral": {}, "xxx_unrecognized": null, "xxx_sizecache": 0}, "maxinstance": {"name": "MaxInstance", "metricname": "MaxInstance", "value": 1774.0, "period": "2022-04-29 14:30:00 +0800 CST", "xxx_nounkeyedliteral": {}, "xxx_unrecognized": null, "xxx_sizecache": 0}, "cpurequest": 3334.8, "memoryrequest": 6358271262720}, {"index": 50, "time": {"$date": "2022-04-29T06:50:00Z"}, "nodecount": 0, "availablenodecount": 0, "minnode": {"name": "MinNode", "metricname": "MinNode", "value": 0.0, "period": "2022-04-29 14:50:00 +0800 CST", "xxx_nounkeyedliteral": {}, "xxx_unrecognized": null, "xxx_sizecache": 0}, "maxnode": {"name": "MaxNode", "metricname": "MaxNode", "value": 0.0, "period": "2022-04-29 14:50:00 +0800 CST", "xxx_nounkeyedliteral": {}, "xxx_unrecognized": null, "xxx_sizecache": 0}, "nodequantile": [], "minusagenode": "", "totalcpu": 4272.0, "totalmemory": 11978108223488, "totalloadcpu": 84.36179384860675, "totalloadmemory": 2845803163648, "avgloadcpu": 0.0, "avgloadmemory": 0, "cpuusage": 0.019747610919617686, "memoryusage": 0.23758369105963112, "workloadcount": 136, "instancecount": 1774, "mininstance": {"name": "MinInstance", "metricname": "MinInstance", "value": 1774.0, "period": "2022-04-29 14:50:00 +0800 CST", "xxx_nounkeyedliteral": {}, "xxx_unrecognized": null, "xxx_sizecache": 0}, "maxinstance": {"name": "MaxInstance", "metricname": "MaxInstance", "value": 1774.0, "period": "2022-04-29 14:50:00 +0800 CST", "xxx_nounkeyedliteral": {}, "xxx_unrecognized": null, "xxx_sizecache": 0}, "cpurequest": 3334.8, "memoryrequest": 6358271262720}]}' of type ARRAY
        at com.ververica.cdc.connectors.mongodb.table.MongoDBConnectorDeserializationSchema.convertToString(MongoDBConnectorDeserializationSchema.java:627)
        at com.ververica.cdc.connectors.mongodb.table.MongoDBConnectorDeserializationSchema.lambda$wrapIntoNullableConverter$81b05d5e$1(MongoDBConnectorDeserializationSchema.java:762)
        at com.ververica.cdc.connectors.mongodb.table.MongoDBConnectorDeserializationSchema.convertField(MongoDBConnectorDeserializationSchema.java:749)
        at com.ververica.cdc.connectors.mongodb.table.MongoDBConnectorDeserializationSchema.lambda$createRowConverter$8ca31a0b$1(MongoDBConnectorDeserializationSchema.java:686)
        at com.ververica.cdc.connectors.mongodb.table.MongoDBConnectorDeserializationSchema.lambda$wrapIntoNullableConverter$81b05d5e$1(MongoDBConnectorDeserializationSchema.java:762)
        at com.ververica.cdc.connectors.mongodb.table.MongoDBConnectorDeserializationSchema.extractRowData(MongoDBConnectorDeserializationSchema.java:167)
        at com.ververica.cdc.connectors.mongodb.table.MongoDBConnectorDeserializationSchema.deserialize(MongoDBConnectorDeserializationSchema.java:131)
        at com.ververica.cdc.debezium.internal.DebeziumChangeFetcher.handleBatch(DebeziumChangeFetcher.java:229)
        at com.ververica.cdc.debezium.internal.DebeziumChangeFetcher.runFetchLoop(DebeziumChangeFetcher.java:151)
        at com.ververica.cdc.debezium.DebeziumSourceFunction.run(DebeziumSourceFunction.java:446)
        at org.apache.flink.streaming.api.operators.StreamSource.run(StreamSource.java:110)
        at org.apache.flink.streaming.api.operators.StreamSource.run(StreamSource.java:67)
        at org.apache.flink.streaming.runtime.tasks.SourceStreamTask$LegacySourceFunctionThread.run(SourceStreamTask.java:323)

[vanliu-tx]

@Jiabao-Sun @leonardBang could you help to review this?

[Jiabao-Sun]

Thanks @vanliu-tx, I left a few comments.

[vanliu-tx]

Thanks @vanliu-tx, I left a few comments.

ok, thanks for your review.

[Jiabao-Sun]

Thanks @vanliu-tx, LGTM.

[ShawshankLin]

请问这个哪个版本会有支持呢？