debezium-connector-kingbase对接人大金仓kes v9的cdc增量信息的bug

[pinis123]

你好
我采用你的debezium-connector-kingbase项目，对接人大金仓kes v9的cdc增量信息

实测发现，只能正常处理op=r 也就是快照，对于insert插入操作就报错了，报错：

2025-06-11 15:27:37,679 ERROR  ||  Error while trying to run connector class 'io.debezium.connector.postgresqlkes.PostgresConnector'   [io.debezium.embedded.EmbeddedEngine]
org.apache.kafka.connect.errors.ConnectException: An exception occurred in the change event producer. This connector will be stopped.
	at io.debezium.pipeline.ErrorHandler.setProducerThrowable(ErrorHandler.java:50)
	at io.debezium.connector.postgresqlkes.PostgresStreamingChangeEventSource.execute(PostgresStreamingChangeEventSource.java:181)
	at io.debezium.connector.postgresqlkes.PostgresStreamingChangeEventSource.execute(PostgresStreamingChangeEventSource.java:42)
	at io.debezium.pipeline.ChangeEventSourceCoordinator.streamEvents(ChangeEventSourceCoordinator.java:174)
	at io.debezium.pipeline.ChangeEventSourceCoordinator.executeChangeEventSources(ChangeEventSourceCoordinator.java:141)
	at io.debezium.pipeline.ChangeEventSourceCoordinator.lambda$start$0(ChangeEventSourceCoordinator.java:109)
	at java.base/java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:515)
	at java.base/java.util.concurrent.FutureTask.run$$$capture(FutureTask.java:264)
	at java.base/java.util.concurrent.FutureTask.run(FutureTask.java)
	at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128)
	at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628)
	at java.base/java.lang.Thread.run(Thread.java:834)
Caused by: org.apache.kafka.connect.errors.ConnectException: com.google.protobuf.InvalidProtocolBufferException: While parsing a protocol message, the input ended unexpectedly in the middle of a field.  This could mean either that the input has been truncated or that an embedded message misreported its own length.
	at io.debezium.connector.postgresqlkes.connection.pgproto.PgProtoMessageDecoder.processNotEmptyMessage(PgProtoMessageDecoder.java:71)
	at io.debezium.connector.postgresqlkes.connection.AbstractMessageDecoder.processMessage(AbstractMessageDecoder.java:33)
	at io.debezium.connector.postgresqlkes.connection.PostgresReplicationConnection$1.deserializeMessages(PostgresReplicationConnection.java:533)
	at io.debezium.connector.postgresqlkes.connection.PostgresReplicationConnection$1.readPending(PostgresReplicationConnection.java:525)
	at io.debezium.connector.postgresqlkes.PostgresStreamingChangeEventSource.searchWalPosition(PostgresStreamingChangeEventSource.java:327)
	at io.debezium.connector.postgresqlkes.PostgresStreamingChangeEventSource.execute(PostgresStreamingChangeEventSource.java:162)
	... 10 common frames omitted
Caused by: com.google.protobuf.InvalidProtocolBufferException: While parsing a protocol message, the input ended unexpectedly in the middle of a field.  This could mean either that the input has been truncated or that an embedded message misreported its own length.
	at com.google.protobuf.InvalidProtocolBufferException.truncatedMessage(InvalidProtocolBufferException.java:107)
	at com.google.protobuf.CodedInputStream$ArrayDecoder.readRawLittleEndian32(CodedInputStream.java:1144)
	at com.google.protobuf.CodedInputStream$ArrayDecoder.readFixed32(CodedInputStream.java:783)
	at com.google.protobuf.UnknownFieldSet$Builder.mergeFieldFrom(UnknownFieldSet.java:524)
	at com.google.protobuf.UnknownFieldSet$Builder.mergeFrom(UnknownFieldSet.java:491)
	at com.google.protobuf.UnknownFieldSet$Builder.mergeFrom(UnknownFieldSet.java:604)
	at com.google.protobuf.UnknownFieldSet$Builder.mergeFrom(UnknownFieldSet.java:304)
	at com.google.protobuf.CodedInputStream$ArrayDecoder.readGroup(CodedInputStream.java:837)
	at com.google.protobuf.UnknownFieldSet$Builder.mergeFieldFrom(UnknownFieldSet.java:518)
	at com.google.protobuf.GeneratedMessageV3$Builder.parseUnknownField(GeneratedMessageV3.java:863)
	at io.debezium.connector.postgresql.proto.PgProto$Point$Builder.mergeFrom(PgProto.java:613)
	at io.debezium.connector.postgresql.proto.PgProto$Point$Builder.mergeFrom(PgProto.java:445)
	at com.google.protobuf.CodedInputStream$ArrayDecoder.readMessage(CodedInputStream.java:872)
	at io.debezium.connector.postgresql.proto.PgProto$DatumMessage$Builder.mergeFrom(PgProto.java:1941)
	at io.debezium.connector.postgresql.proto.PgProto$DatumMessage$1.parsePartialFrom(PgProto.java:2670)
	at io.debezium.connector.postgresql.proto.PgProto$DatumMessage$1.parsePartialFrom(PgProto.java:2662)
	at com.google.protobuf.CodedInputStream$ArrayDecoder.readMessage(CodedInputStream.java:889)
	at io.debezium.connector.postgresql.proto.PgProto$RowMessage$Builder.mergeFrom(PgProto.java:4417)
	at io.debezium.connector.postgresql.proto.PgProto$RowMessage$1.parsePartialFrom(PgProto.java:5417)
	at io.debezium.connector.postgresql.proto.PgProto$RowMessage$1.parsePartialFrom(PgProto.java:5409)
	at com.google.protobuf.AbstractParser.parsePartialFrom(AbstractParser.java:158)
	at com.google.protobuf.AbstractParser.parseFrom(AbstractParser.java:191)
	at com.google.protobuf.AbstractParser.parseFrom(AbstractParser.java:203)
	at com.google.protobuf.AbstractParser.parseFrom(AbstractParser.java:208)
	at com.google.protobuf.AbstractParser.parseFrom(AbstractParser.java:48)
	at io.debezium.connector.postgresql.proto.PgProto$RowMessage.parseFrom(PgProto.java:3994)
	at io.debezium.connector.postgresqlkes.connection.pgproto.PgProtoMessageDecoder.processNotEmptyMessage(PgProtoMessageDecoder.java:53)
	... 15 common frames omitted。。。。。。

看上去和

<dependency>
            <groupId>com.google.protobuf</groupId>
            <artifactId>protobuf-java</artifactId>
            <version>3.3.0</version>
        </dependency>

的具体版本有关系，切换protobuf-java不同的版本往往会有不同的错误。。。

当然了，对接人大kes ，采用的底层逻辑解码插件是decoderbufs。。。这个你肯定知道

我的问题是，你们具体实测过吗？在人大金仓kes v8/v9上，具体实战测试过吗？抑或，你们是否遭遇相关的问题？

[pinis123]

@baisui1981

[pinis123]

@baisui1981 时间紧，可以尽快看一下吗？谢谢

[pinis123]

如，当protobuf-java采用3.8.0以后，报：

2025-06-11 16:04:42,535 ERROR  ||  Error while trying to run connector class 'io.debezium.connector.postgresqlkes.PostgresConnector'   [io.debezium.embedded.EmbeddedEngine]
org.apache.kafka.connect.errors.ConnectException: An exception occurred in the change event producer. This connector will be stopped.
	at io.debezium.pipeline.ErrorHandler.setProducerThrowable(ErrorHandler.java:50)
	at io.debezium.connector.postgresqlkes.PostgresStreamingChangeEventSource.execute(PostgresStreamingChangeEventSource.java:181)
	at io.debezium.connector.postgresqlkes.PostgresStreamingChangeEventSource.execute(PostgresStreamingChangeEventSource.java:42)
	at io.debezium.pipeline.ChangeEventSourceCoordinator.streamEvents(ChangeEventSourceCoordinator.java:174)
	at io.debezium.pipeline.ChangeEventSourceCoordinator.executeChangeEventSources(ChangeEventSourceCoordinator.java:141)
	at io.debezium.pipeline.ChangeEventSourceCoordinator.lambda$start$0(ChangeEventSourceCoordinator.java:109)
	at java.base/java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:515)
	at java.base/java.util.concurrent.FutureTask.run$$$capture(FutureTask.java:264)
	at java.base/java.util.concurrent.FutureTask.run(FutureTask.java)
	at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128)
	at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628)
	at java.base/java.lang.Thread.run(Thread.java:834)
Caused by: java.lang.NoSuchMethodError: com.google.protobuf.GeneratedMessageV3$Builder.parseUnknownField(Lcom/google/protobuf/CodedInputStream;Lcom/google/protobuf/ExtensionRegistryLite;I)Z
	at io.debezium.connector.postgresql.proto.PgProto$RowMessage$Builder.mergeFrom(PgProto.java:4442)
	at io.debezium.connector.postgresql.proto.PgProto$RowMessage$1.parsePartialFrom(PgProto.java:5417)
	at io.debezium.connector.postgresql.proto.PgProto$RowMessage$1.parsePartialFrom(PgProto.java:5409)
	at com.google.protobuf.AbstractParser.parsePartialFrom(AbstractParser.java:158)
	at com.google.protobuf.AbstractParser.parseFrom(AbstractParser.java:191)
	at com.google.protobuf.AbstractParser.parseFrom(AbstractParser.java:203)
	at com.google.protobuf.AbstractParser.parseFrom(AbstractParser.java:208)
	at com.google.protobuf.AbstractParser.parseFrom(AbstractParser.java:48)
	at io.debezium.connector.postgresql.proto.PgProto$RowMessage.parseFrom(PgProto.java:3994)
	at io.debezium.connector.postgresqlkes.connection.pgproto.PgProtoMessageDecoder.processNotEmptyMessage(PgProtoMessageDecoder.java:53)
	at io.debezium.connector.postgresqlkes.connection.AbstractMessageDecoder.processMessage(AbstractMessageDecoder.java:33)
	at io.debezium.connector.postgresqlkes.connection.PostgresReplicationConnection$1.deserializeMessages(PostgresReplicationConnection.java:533)
	at io.debezium.connector.postgresqlkes.connection.PostgresReplicationConnection$1.readPending(PostgresReplicationConnection.java:525)
	at io.debezium.connector.postgresqlkes.PostgresStreamingChangeEventSource.searchWalPosition(PostgresStreamingChangeEventSource.java:327)
	at io.debezium.connector.postgresqlkes.PostgresStreamingChangeEventSource.execute(PostgresStreamingChangeEventSource.java:162)
	... 10 common frames omitted

[pinis123]

@baisui1981 现在，最新的报错日志：

2025-06-12 12:07:18,686 ERROR  ||  Error while trying to run connector class 'io.debezium.connector.postgresql.PostgresConnector'   [io.debezium.embedded.EmbeddedEngine]
org.apache.kafka.connect.errors.ConnectException: An exception occurred in the change event producer. This connector will be stopped.
	at io.debezium.pipeline.ErrorHandler.setProducerThrowable(ErrorHandler.java:50)
	at io.debezium.connector.postgresql.PostgresStreamingChangeEventSource.execute(PostgresStreamingChangeEventSource.java:181)
	at io.debezium.connector.postgresql.PostgresStreamingChangeEventSource.execute(PostgresStreamingChangeEventSource.java:42)
	at io.debezium.pipeline.ChangeEventSourceCoordinator.streamEvents(ChangeEventSourceCoordinator.java:174)
	at io.debezium.pipeline.ChangeEventSourceCoordinator.executeChangeEventSources(ChangeEventSourceCoordinator.java:141)
	at io.debezium.pipeline.ChangeEventSourceCoordinator.lambda$start$0(ChangeEventSourceCoordinator.java:109)
	at java.base/java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:515)
	at java.base/java.util.concurrent.FutureTask.run$$$capture(FutureTask.java:264)
	at java.base/java.util.concurrent.FutureTask.run(FutureTask.java)
	at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128)
	at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628)
	at java.base/java.lang.Thread.run(Thread.java:834)
Caused by: org.apache.kafka.connect.errors.ConnectException: com.google.protobuf.InvalidProtocolBufferException: Protocol message end-group tag did not match expected tag.
	at io.debezium.connector.postgresql.connection.pgproto.PgProtoMessageDecoder.processNotEmptyMessage(PgProtoMessageDecoder.java:71)
	at io.debezium.connector.postgresql.connection.AbstractMessageDecoder.processMessage(AbstractMessageDecoder.java:33)
	at io.debezium.connector.postgresql.connection.PostgresReplicationConnection$1.deserializeMessages(PostgresReplicationConnection.java:533)
	at io.debezium.connector.postgresql.connection.PostgresReplicationConnection$1.readPending(PostgresReplicationConnection.java:525)
	at io.debezium.connector.postgresql.PostgresStreamingChangeEventSource.processMessages(PostgresStreamingChangeEventSource.java:216)
	at io.debezium.connector.postgresql.PostgresStreamingChangeEventSource.execute(PostgresStreamingChangeEventSource.java:178)
	... 10 common frames omitted
Caused by: com.google.protobuf.InvalidProtocolBufferException: Protocol message end-group tag did not match expected tag.
	at com.google.protobuf.InvalidProtocolBufferException.invalidEndTag(InvalidProtocolBufferException.java:129)
	at com.google.protobuf.CodedInputStream$ArrayDecoder.checkLastTagWas(CodedInputStream.java:641)
	at com.google.protobuf.CodedInputStream$ArrayDecoder.readGroup(CodedInputStream.java:838)
	at com.google.protobuf.UnknownFieldSet$Builder.mergeFieldFrom(UnknownFieldSet.java:518)
	at com.google.protobuf.UnknownFieldSet$Builder.mergeFrom(UnknownFieldSet.java:491)
	at com.google.protobuf.UnknownFieldSet$Builder.mergeFrom(UnknownFieldSet.java:604)
	at com.google.protobuf.UnknownFieldSet$Builder.mergeFrom(UnknownFieldSet.java:304)
	at com.google.protobuf.CodedInputStream$ArrayDecoder.readGroup(CodedInputStream.java:837)
	at com.google.protobuf.UnknownFieldSet$Builder.mergeFieldFrom(UnknownFieldSet.java:518)
	at com.google.protobuf.GeneratedMessageV3$Builder.parseUnknownField(GeneratedMessageV3.java:863)
	at io.debezium.connector.postgresql.proto.PgProto$Point$Builder.mergeFrom(PgProto.java:613)
	at io.debezium.connector.postgresql.proto.PgProto$Point$Builder.mergeFrom(PgProto.java:445)
	at com.google.protobuf.CodedInputStream$ArrayDecoder.readMessage(CodedInputStream.java:872)
	at io.debezium.connector.postgresql.proto.PgProto$DatumMessage$Builder.mergeFrom(PgProto.java:1941)
	at io.debezium.connector.postgresql.proto.PgProto$DatumMessage$1.parsePartialFrom(PgProto.java:2670)
	at io.debezium.connector.postgresql.proto.PgProto$DatumMessage$1.parsePartialFrom(PgProto.java:2662)
	at com.google.protobuf.CodedInputStream$ArrayDecoder.readMessage(CodedInputStream.java:889)
	at io.debezium.connector.postgresql.proto.PgProto$RowMessage$Builder.mergeFrom(PgProto.java:4417)
	at io.debezium.connector.postgresql.proto.PgProto$RowMessage$1.parsePartialFrom(PgProto.java:5417)
	at io.debezium.connector.postgresql.proto.PgProto$RowMessage$1.parsePartialFrom(PgProto.java:5409)
	at com.google.protobuf.AbstractParser.parsePartialFrom(AbstractParser.java:158)
	at com.google.protobuf.AbstractParser.parseFrom(AbstractParser.java:191)
	at com.google.protobuf.AbstractParser.parseFrom(AbstractParser.java:203)
	at com.google.protobuf.AbstractParser.parseFrom(AbstractParser.java:208)
	at com.google.protobuf.AbstractParser.parseFrom(AbstractParser.java:48)
	at io.debezium.connector.postgresql.proto.PgProto$RowMessage.parseFrom(PgProto.java:3994)
	at io.debezium.connector.postgresql.connection.pgproto.PgProtoMessageDecoder.processNotEmptyMessage(PgProtoMessageDecoder.java:53)
	... 15 common frames omitted。。。

看到了，请回复一下

[baisui1981]

我们本地是使用人大金仓的docker 版本测试的，使用的是：

docker run -tid --privileged -p 4321:54321 -v /root/kingbase/data_mysql:/home/kingbase/userdata/ -e ENABLE_CI=yes -e NEED_START=yes -e DB_USER=kingbase -e DB_PASSWORD=123456 -e DB_MODE=mysql --name kingbase_mysql kingbase_v009r001c002b0014_single_x86:v1

按照docker image 应该是 v9版本，能否提供提供：

1. 建表的ddl，
2. 该表的insert语句，有代表性的一条即可
3. 使用db model（mysql？oracle？）

[pinis123]

@baisui1981 > 我们本地是使用人大金仓的docker 版本测试的，使用的是：

docker run -tid --privileged -p 4321:54321 -v /root/kingbase/data_mysql:/home/kingbase/userdata/ -e ENABLE_CI=yes -e NEED_START=yes -e DB_USER=kingbase -e DB_PASSWORD=123456 -e DB_MODE=mysql --name kingbase_mysql kingbase_v009r001c002b0014_single_x86:v1

按照docker image 应该是 v9版本，能否提供提供：

1. 建表的ddl，
2. 该表的insert语句，有代表性的一条即可
3. 使用db model（mysql？oracle？）

我看你用的是兼容mysql模式吗？我用的兼容pg模式。还有，我把代码下下来，这段代码的url_pattern是红框里的部分，截图是

。红框下面的是我改的。还有，我看应该是你修改的这个文件，截图是

。我跟踪调试了一下代码，我看逻辑上，并没有使用这个pg_logicaldec.proto文件。这些疑问，可以解答一下吗？建表和insert，是我自己弄的。ddl是：-- public.users3 definition

-- Drop table

-- DROP TABLE public.users3;

CREATE TABLE public.users3 (
id serial DEFAULT nextval('users3_id_seq'::regclass) NOT NULL,
"name" varchar(50),
email varchar(100),
CONSTRAINT users3_pkey PRIMARY KEY (id)
);
insert就是INSERT INTO public.users3 ( name, email) VALUES( 'daniu', 'daniu');
这肯定没问题。

可以加个qq沟通一下吗？我的qq是729672518。

[pinis123]

我们本地是使用人大金仓的docker 版本测试的，使用的是：

docker run -tid --privileged -p 4321:54321 -v /root/kingbase/data_mysql:/home/kingbase/userdata/ -e ENABLE_CI=yes -e NEED_START=yes -e DB_USER=kingbase -e DB_PASSWORD=123456 -e DB_MODE=mysql --name kingbase_mysql kingbase_v009r001c002b0014_single_x86:v1

按照docker image 应该是 v9版本，能否提供提供：

1. 建表的ddl，
2. 该表的insert语句，有代表性的一条即可
3. 使用db model（mysql？oracle？）

@baisui1981 我用的kes 版本 ，截图是：

。兼容postgresql模式

[baisui1981]

ok，我本地回归一下试试

[pinis123]

ok，我本地回归一下试试

@baisui1981 可以加一下qq联系方式吗？我的qq是：729672518，或者你把你的qq发我

[baisui1981]

ok，我本地回归一下试试

@baisui1981 可以加一下qq联系方式吗？我的qq是：729672518，或者你把你的qq发我

通过你的qq号已经加了你的微信。另外，你是否用的是 https://github.com/qlangtech/debezium/blob/tis.v1.9.8.Final/debezium-connector-kingbase/src/main/proto/pg_logicaldec.proto 这份协议文件？

[pinis123]

ok，我本地回归一下试试

@baisui1981 可以加一下qq联系方式吗？我的qq是：729672518，或者你把你的qq发我

通过你的qq号已经加了你的微信。另外，你是否用的是 https://github.com/qlangtech/debezium/blob/tis.v1.9.8.Final/debezium-connector-kingbase/src/main/proto/pg_logicaldec.proto 这份协议文件？

@baisui1981 我是下载的这个代码，但我跟踪调试代码以后，我的判断是代码逻辑和流程上没使用这份协议文件（https://github.com/qlangtech/debezium/blob/tis.v1.9.8.Final/debezium-connector-kingbase/src/main/proto/pg_logicaldec.proto ），或者是我没有使用对？我下载了代码以后，我可以把我的代码发你，你看一下，我看一下微信

[zhaotiemao]

遇到同样的问题了，能否给个联系方式请教一下

[zhaotiemao]

我们的应用场景：需要捕获kingbase的数据变动，转换一下格式，实时投送到kafka里给后面的业务消费