sql cdc connector 支持kingbaseESV8人大金仓

[jeansitplay]

使用：flink-sql-connector-postgres-cdc-2.2.1
（1）使用pg jdbc驱动去获取kingbase某张数据库表全量数据时，能正常获取到。
（2）后续获取这经表的增量数据时，一直不能成功，报Received message with type 'UNKNOWN' that is unknown to this version of connector, consider upgrading。
（3）查询了下pg解析wal与kingbase解析wal日志不一样。

有哪位同学实现了kingbase的cdc connector吗？

具体日志如下：
2023-04-04 14:25:02,945 INFO io.debezium.connector.postgresql.PostgresConnectorTask [] - user '{USER_NAME}' connected to database '{DB_NAME}' on KingbaseES {VERSION_NO} on x86_64-pc-linux-gnu, compiled by gcc (GCC) 4.1.2 20080704 (Red Hat 4.1.2-46), 64-bit with roles:
role '{USER_NAME}' [superuser: false, replication: true, inherit: false, create role: false, create db: true, can log in: true]
2023-04-04 14:25:02,958 INFO io.debezium.connector.postgresql.connection.PostgresConnection [] - Obtained valid replication slot ReplicationSlot [active=false, latestFlushedLsn=LSN{10/F46AF678}, catalogXmin=23346119]
2023-04-04 14:25:02,959 INFO io.debezium.connector.postgresql.PostgresConnectorTask [] - No previous offset found
2023-04-04 14:25:02,959 INFO io.debezium.connector.postgresql.snapshot.InitialSnapshotter [] - Taking initial snapshot for new datasource
2023-04-04 14:25:02,991 INFO io.debezium.util.Threads [] - Requested thread factory for connector PostgresConnector, id = postgres_cdc_source named = change-event-source-coordinator
2023-04-04 14:25:02,994 INFO io.debezium.util.Threads [] - Creating thread debezium-postgresconnector-postgres_cdc_source-change-event-source-coordinator
2023-04-04 14:25:02,999 INFO io.debezium.pipeline.ChangeEventSourceCoordinator [] - Metrics registered
2023-04-04 14:25:03,000 INFO io.debezium.pipeline.ChangeEventSourceCoordinator [] - Context created
2023-04-04 14:25:03,006 INFO io.debezium.connector.postgresql.snapshot.InitialSnapshotter [] - Taking initial snapshot for new datasource
2023-04-04 14:25:03,006 INFO io.debezium.connector.postgresql.PostgresSnapshotChangeEventSource [] - According to the connector configuration data will be snapshotted
2023-04-04 14:25:03,006 INFO io.debezium.relational.RelationalSnapshotChangeEventSource [] - Snapshot step 1 - Preparing
2023-04-04 14:25:03,171 INFO io.debezium.relational.RelationalSnapshotChangeEventSource [] - Snapshot step 2 - Determining captured tables
2023-04-04 14:25:03,194 INFO io.debezium.relational.RelationalSnapshotChangeEventSource [] - Snapshot step 3 - Locking captured tables [{SCHEMA_NAME}.{TABLE_NAME}]
2023-04-04 14:25:03,195 INFO io.debezium.connector.postgresql.PostgresSnapshotChangeEventSource [] - Waiting a maximum of '10' seconds for each table lock
2023-04-04 14:25:03,245 INFO io.debezium.relational.RelationalSnapshotChangeEventSource [] - Snapshot step 4 - Determining snapshot offset
2023-04-04 14:25:03,246 INFO io.debezium.connector.postgresql.PostgresSnapshotChangeEventSource [] - Creating initial offset context
2023-04-04 14:25:03,263 INFO io.debezium.connector.postgresql.PostgresSnapshotChangeEventSource [] - Read xlogStart at 'LSN{11/73FA7608}' from transaction '26988742'
2023-04-04 14:25:03,275 INFO io.debezium.connector.postgresql.PostgresSnapshotChangeEventSource [] - Read xlogStart at 'LSN{11/73FA7608}' from transaction '26988742'
2023-04-04 14:25:03,276 INFO io.debezium.relational.RelationalSnapshotChangeEventSource [] - Snapshot step 5 - Reading structure of captured tables
2023-04-04 14:25:03,276 INFO io.debezium.connector.postgresql.PostgresSnapshotChangeEventSource [] - Reading structure of schema '{DB_NAME}'
2023-04-04 14:25:03,347 INFO io.debezium.relational.RelationalSnapshotChangeEventSource [] - Snapshot step 6 - Persisting schema history
2023-04-04 14:25:03,347 INFO io.debezium.relational.RelationalSnapshotChangeEventSource [] - Snapshot step 7 - Snapshotting data
2023-04-04 14:25:03,348 INFO io.debezium.relational.RelationalSnapshotChangeEventSource [] - Snapshotting contents of 1 tables while still in transaction
2023-04-04 14:25:03,348 INFO io.debezium.relational.RelationalSnapshotChangeEventSource [] - Exporting data from table '{SCHEMA_NAME}.{TABLE_NAME}' (1 of 1 tables)
2023-04-04 14:25:03,348 INFO io.debezium.relational.RelationalSnapshotChangeEventSource [] - For table '{SCHEMA_NAME}.{TABLE_NAME}' using select statement: 'SELECT * FROM "{SCHEMA_NAME}"."{TABLE_NAME}"'
2023-04-04 14:25:03,978 INFO com.ververica.cdc.debezium.internal.DebeziumChangeFetcher [] - Database snapshot phase can't perform checkpoint, acquired Checkpoint lock.
2023-04-04 14:25:04,920 INFO io.debezium.relational.RelationalSnapshotChangeEventSource [] - Finished exporting 4756 records for table '{SCHEMA_NAME}.{TABLE_NAME'; total duration '00:00:01.571'
2023-04-04 14:25:04,930 INFO io.debezium.pipeline.source.AbstractSnapshotChangeEventSource [] - Snapshot - Final stage
2023-04-04 14:25:04,931 INFO io.debezium.pipeline.ChangeEventSourceCoordinator [] - Snapshot ended with SnapshotResult [status=COMPLETED, offset=PostgresOffsetContext [sourceInfoSchema=Schema{io.debezium.connector.postgresql.Source:STRUCT}, sourceInfo=source_info[server='postgres_cdc_source'db='{DB_NAME}', lsn=LSN{11/73FA7608}, txId=26988742, sequence=[Lio.debezium.connector.postgresql.connection.Lsn;@715a8657, timestamp=2023-04-04T06:25:04.919Z, snapshot=FALSE, schema={SCHEMA_NAME}, table={TABLE_NAME}], partition={server=postgres_cdc_source}, lastSnapshotRecord=true, lastCompletelyProcessedLsn=null, lastCommitLsn=null, transactionContext=TransactionContext [currentTransactionId=null, perTableEventCount={}, totalEventCount=0]]]
2023-04-04 14:25:04,938 INFO io.debezium.pipeline.metrics.StreamingChangeEventSourceMetrics [] - Connected metrics set to 'true'
2023-04-04 14:25:04,939 INFO io.debezium.pipeline.ChangeEventSourceCoordinator [] - Starting streaming
2023-04-04 14:25:04,939 INFO io.debezium.connector.postgresql.PostgresStreamingChangeEventSource [] - Retrieved latest position from stored offset 'LSN{11/73FA7608}'
2023-04-04 14:25:04,940 INFO io.debezium.connector.postgresql.connection.WalPositionLocator [] - Looking for WAL restart position for last commit LSN 'null' and last change LSN 'LSN{11/73FA7608}'
2023-04-04 14:25:05,035 INFO io.debezium.connector.postgresql.connection.PostgresConnection [] - Obtained valid replication slot ReplicationSlot [active=false, latestFlushedLsn=LSN{10/F46AF678}, catalogXmin=23346119]
2023-04-04 14:25:05,040 INFO io.debezium.jdbc.JdbcConnection [] - Connection gracefully closed
2023-04-04 14:25:05,182 INFO io.debezium.connector.postgresql.PostgresSchema [] - REPLICA IDENTITY for '{SCHEMA_NAME}.{TABLE_NAME}' is 'FULL'; UPDATE AND DELETE events will contain the previous values of all the columns
2023-04-04 14:25:05,189 INFO io.debezium.connector.postgresql.PostgresStreamingChangeEventSource [] - Searching for WAL resume position
2023-04-04 14:25:06,591 INFO com.ververica.cdc.debezium.internal.DebeziumChangeFetcher [] - Received record from streaming binlog phase, released checkpoint lock.
2023-04-04 14:29:05,614 WARN io.debezium.connector.postgresql.connection.pgproto.PgProtoMessageDecoder [] - Received message with type 'UNKNOWN' that is unknown to this version of connector, consider upgrading

[Chinafrfq]

有解决方案吗

[141539286]

https://gitee.com/sudashuilzx/debezium-kingbase ，我基于debezium1.5.4实现了一版解析kingbase，你需要修改一下才能集成到最新版本的flinkcdc

[yuana1]

https://gitee.com/zhaoky0918/flink-connector-kingbase-cdc 适配3.0.1版本。参考pgsql的proto格式，逆向了kingbase的。只覆盖了PgProto类和PgProtoColumnValue。