bkbase-plugin/flink-cdc-connectors

CDC Connectors for Apache Flink®

flink-cdc-connectors's Issues

update debezium version to v1.5.4-bkbase.1

Is your feature request related to a problem? Please describe.
A clear and concise description of what the problem is. Ex. I'm always frustrated when [...]

Describe the solution you'd like
A clear and concise description of what you want to happen.

Describe alternatives you've considered
A clear and concise description of any alternative solutions or features you've considered.

change io.debezium to com.tencent.bk.base.datahub in fat jar

Is your feature request related to a problem? Please describe.
A clear and concise description of what the problem is. Ex. I'm always frustrated when [...]

Describe the solution you'd like
A clear and concise description of what you want to happen.

Describe alternatives you've considered
A clear and concise description of any alternative solutions or features you've considered.

Use SampleBucketSplitStrategy for shard collection with hashed keys

For collection with hashed keys as shard key, the record counts in chunks are varied, some may contains millions of records, other may contains a few thousands.

This would lead to OOM exception during snapshot phrase if a chunk contains millions of records.

This feature use SampleBucketSplitStrategy instead of ShardedSplitStrategy to avoid this OOM situation.

Allow mongo ARRAY to be converted to string type in Flink

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

release 2.2.1-bkbase.1

Is your feature request related to a problem? Please describe.
A clear and concise description of what the problem is. Ex. I'm always frustrated when [...]

Describe the solution you'd like
A clear and concise description of what you want to happen.

Describe alternatives you've considered
A clear and concise description of any alternative solutions or features you've considered.

allow mongodb document field defined as string type in flink

Is your feature request related to a problem? Please describe.
Sometimes it's unable to define the mongodb collection schema as a table, the schema changes. We want some nested document to be a string for later parse and calculate, but now you need to define the exactly field type in flink.

Describe the solution you'd like
Allow mapping a nested JSON document to flink string type.

2022-07-15 17:50:37,770 WARN  org.apache.flink.runtime.taskmanager.Task                    [] - Source: mongodb source dynamic.dynamic.Node -> Map -> IcebergStreamWriter (1/1)#3 (36334134ff691818fe716011e049fe09) switched from RUNNING to FAILED with failure cause: java.lang.IllegalArgumentException: Unable to convert to string from unexpected value '{"metadata": {"creationTimestamp": "2022-03-17T08:31:27Z", "labels": {"beta.kubernetes.io/os": "linux", "failure-domain.beta.kubernetes.io/region": "nj"}}}' of type DOCUMENT
        at com.ververica.cdc.connectors.mongodb.table.MongoDBConnectorDeserializationSchema.convertToString(MongoDBConnectorDeserializationSchema.java:629)
        at com.ververica.cdc.connectors.mongodb.table.MongoDBConnectorDeserializationSchema.lambda$wrapIntoNullableConverter$81b05d5e$1(MongoDBConnectorDeserializationSchema.java:764)
        at com.ververica.cdc.connectors.mongodb.table.MongoDBConnectorDeserializationSchema.convertField(MongoDBConnectorDeserializationSchema.java:751)
        at com.ververica.cdc.connectors.mongodb.table.MongoDBConnectorDeserializationSchema.lambda$createRowConverter$8ca31a0b$1(MongoDBConnectorDeserializationSchema.java:688)
        at com.ververica.cdc.connectors.mongodb.table.MongoDBConnectorDeserializationSchema.lambda$wrapIntoNullableConverter$81b05d5e$1(MongoDBConnectorDeserializationSchema.java:764)
        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)

[mongodb] Add lookup option to IncrementalSource for mongodb

Search before asking

• I searched in the issues and found nothing similar.

Motivation

this option is missing

Solution

No response

Alternatives

No response

Anything else?

No response

Are you willing to submit a PR?

• I'm willing to submit a PR!

release 2.2.1-bkbase.2

new release

mongodb cdc hotfix

some hotfix for mongodb cdc in 2.3.0

change groupId to com.tencent.bk.base.datahub

Is your feature request related to a problem? Please describe.
A clear and concise description of what the problem is. Ex. I'm always frustrated when [...]

Describe the solution you'd like
A clear and concise description of what you want to happen.

Describe alternatives you've considered
A clear and concise description of any alternative solutions or features you've considered.

pick 2.2.1 features to 2.3.0

upgrade to 2.3.0, and cherry-pick features in 2.2.1

split big mongodb chunks into many small chunks

Is your feature request related to a problem? Please describe.
A clear and concise description of what the problem is. Ex. I'm always frustrated when [...]

Describe the solution you'd like
A clear and concise description of what you want to happen.

Describe alternatives you've considered
A clear and concise description of any alternative solutions or features you've considered.

Remove bkbase debuzium dependency

as in 2.30, flink cdc mysql package copied source code of MySqlStreamingChangeEventSource from debuzium to fix some issue.