Got error org.apache.kafka.common.InvalidRecordException: Compacted topic cannot accept message without key in topic partition

[Thanh Thảo Huỳnh]

Hi,

I'm using Debezium 2.0.0.Beta1. This is my connector config:

connector.class: io.debezium.connector.postgresql.PostgresConnector,

database.user: test,

database.dbname: test,

topic.creation.default.compression.type: gzip,

topic.creation.default.partitions: 3,

topic.creation.default.cleanup.policy: compact,

slot.name: test,

tasks.max: 1,

database.server.name: debezium_test,

heartbeat.interval.ms: 600000,

database.port: 5432,

plugin.name: pgoutput,

value.converter.schema.registry.url: http://schema-registry:8081,

database.hostname: ***,

database.password: ***,

topic.creation.default.replication.factor: 1,

signal.data.collection: debezium.signal_test,

name: test,

table.include.list: public.test,debezium.signal_test,

value.converter: io.confluent.connect.avro.AvroConverter,

key.converter: io.confluent.connect.avro.AvroConverter,

key.converter.schema.registry.url: http://schema-registry:8081,

snapshot.mode: initial

After creating the connector, I got this error org.apache.kafka.common.InvalidRecordException: Compacted topic cannot accept message without key in topic partition debezium_test.debezium.signal_test-0

How can I fix it?
Thanks.

[Ragul Balaji]

Debezium sends null keys when the table does not have a primary key or unique key .

https://debezium.io/documentation/reference/2.0/connectors/postgresql.html#postgresql-change-events-key

If you feel that compaction is not required, you can use the below property
topic.creation.default.cleanup.policy: delete

--
You received this message because you are subscribed to the Google Groups "debezium" group.
To unsubscribe from this group and stop receiving emails from it, send an email to debezium+u...@googlegroups.com.
To view this discussion on the web visit https://groups.google.com/d/msgid/debezium/b7bde3a9-41cb-48c9-ba1a-6ed476474a05n%40googlegroups.com.

[Thanh Thảo Huỳnh]

This is my table schema:

CREATE TABLE debezium.signal_test (id TEXT PRIMARY KEY, type TEXT NOT NULL, data TEXT NULL);

Do I miss anything?

[jiri.p...@gmail.com]

Hi,

could you please share he content of the signal table?

J.

[Thanh Thảo Huỳnh]

There is no record in this table. I just created the connector and got this error.

[jiri.p...@gmail.com]

PLease share the full Connect log then.

J.

[Thanh Thảo Huỳnh]

2022-09-05 07:27:32,716 ERROR || WorkerSourceTask{id=test-0} Task threw an uncaught and unrecoverable exception. Task is being killed and will not recover until manually restarted [org.apache.kafka.connect.runtime.WorkerTask]

org.apache.kafka.connect.errors.ConnectException: Unrecoverable exception from producer send callback

at org.apache.kafka.connect.runtime.WorkerSourceTask.maybeThrowProducerSendException(WorkerSourceTask.java:291)

at org.apache.kafka.connect.runtime.WorkerSourceTask.execute(WorkerSourceTask.java:245)

at org.apache.kafka.connect.runtime.WorkerTask.doRun(WorkerTask.java:188)

at org.apache.kafka.connect.runtime.WorkerTask.run(WorkerTask.java:243)

at java.base/java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:515)

at java.base/java.util.concurrent.FutureTask.run(FutureTask.java:264)

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:829)

Caused by: org.apache.kafka.common.InvalidRecordException: Compacted topic cannot accept message without key in topic partition debezium_test.debezium.signal_test-1.

2022-09-05 07:27:32,717 INFO || Stopping down connector [io.debezium.connector.common.BaseSourceTask]

2022-09-05 07:27:33,102 INFO Postgres|debezium_test|streaming Connection gracefully closed [io.debezium.jdbc.JdbcConnection]

2022-09-05 07:27:33,121 INFO Postgres|debezium_test|streaming Connection gracefully closed [io.debezium.jdbc.JdbcConnection]

2022-09-05 07:27:33,121 INFO Postgres|debezium_test|streaming Finished streaming [io.debezium.pipeline.ChangeEventSourceCoordinator]

2022-09-05 07:27:33,121 INFO Postgres|debezium_test|streaming Connected metrics set to 'false' [io.debezium.pipeline.ChangeEventSourceCoordinator]

Here you are. Thank you.

[jiri.p...@gmail.com]

This is only exception I need the full log from the start of the connector

J.

[jiri.p...@gmail.com]

Hi,

would you be able to share the full log? I suspect there are events in the replication slot already that contain DMLs for signalling table with PK set to null.

J.

[Thanh Thảo Huỳnh]

Sorry I missed your email.
This is the full log https://drive.google.com/file/d/1RvuFTB1WqKZo85TOgHWr95q7qJoGPaf7/view?usp=sharing

Thank you so much.

[jiri.p...@gmail.com]

So it is as I expected, the table can be empty yet there are signalling table related records in the replication slot. As the DDL does not define the PK as NOT NULL I suppose there was and INSERT of row with NULL PK.

I'd reommend to change the topic to not compacted to process the data from the slot and then re-enable compaction if needed. Also I'd alter the table to make sure only non-null PK can be inserted.

J.

[Thanh Thảo Huỳnh]

Thanks for your support!

On Wed, Sep 7, 2022 at 5:54 PM Thanh Thảo Huỳnh <thao...@gmail.com> wrote:

Sorry I missed your email.

I attached the log below.

You received this message because you are subscribed to a topic in the Google Groups "debezium" group.
To unsubscribe from this topic, visit https://groups.google.com/d/topic/debezium/P_cUGYqSpX0/unsubscribe.
To unsubscribe from this group and all its topics, send an email to debezium+u...@googlegroups.com.
To view this discussion on the web visit https://groups.google.com/d/msgid/debezium/f5d41547-9011-487d-88fd-8cad2d18c8d1n%40googlegroups.com.