debezium / debezium/dbz

Handle poison messages from source [DBZ-7005]

Open
#902 0 comments 0 reactions 0 assignees View on GitHub
component/core-library migrated-from-jira type/enhancement
Dominant language
HTML
Stars
6
Forks
8
Avg merge
2d 19h
Merged PRs (30d)
1

Description

Migrated from [DBZ-7005](https://issues.redhat.com/browse/DBZ-7005)

h1. *Feature request or enhancement*
h2. *Which use case/requirement will be addressed by the proposed feature?*

This follows from a conversation in the [community chat|[https://debezium.zulipchat.com/#narrow/stream/348106-community-mongodb/topic/DeadLetterQueue.20Doesn't.20Work.20With.20Debezium.20Source.20Connector].|https://debezium.zulipchat.com/#narrow/stream/348106-community-mongodb/topic/DeadLetterQueue.20Doesn't.20Work.20With.20Debezium.20Source.20Connector%5D.]

Playing with the mongodb-outbox example, I ran into a situation where the a record written to the outbox table resulted in a non-retriable error in the Kafka producer, causing the connector to stop, and had no way to skip over the poison record.

In this particular case, the connector was configured to set the Kafka timestamp header from a field in the outbox document, and inserting a document with a negative timestamp would result in the following:

{{
[2023-10-03 14:53:03,346] ERROR [outbox-connector|task-0] WorkerSourceTask\\{id=outbox-connector-0} Task threw an uncaught and unrecoverable exception. Task is being killed and will not recover until manually restarte
d (org.apache.kafka.connect.runtime.WorkerTask:210)
org.apache.kafka.common.InvalidRecordException: Invalid record timestamp -9999999999
[2023-10-03 14:53:03,347] INFO [outbox-connector|task-0] Stopping down connector (io.debezium.connector.common.BaseSourceTask:265)
[2023-10-03 14:53:03,968] INFO [outbox-connector|task-0] Closing all connections to mongodb://mongodb:27017/?replicaSet=rs0 (io.debezium.connector.mongodb.ConnectionContext:119)
[2023-10-03 14:53:03,974] INFO [outbox-connector|task-0] Finished streaming (io.debezium.pipeline.ChangeEventSourceCoordinator:175)}}

It's pretty easy to reproduce -- run the example here https://github.com/debezium/debezium-examples/tree/main/mongodb-outbox

{{then execute this:

docker compose exec mongodb bash -c 'mongo -u $MONGODB_USER -p $MONGODB_PASSWORD --authenticationDatabase admin inventory'db.outboxevent.insert(\\{ aggregateid : "123", aggregatetype : "Order", type : "OrderCreated", timestamp: NumberLong(-999), payload : {} })}}
h2. *Implementation ideas (optional)*

The connector could provide some configuration, API, or utility to facilitate skipping records.

Alternatively, revisiting [KIP-298|https://cwiki.apache.org/confluence/display/KAFKA/KIP-298%3A+Error+Handling+in+Connect] could be in order--the assumption made in the first rejected alternative of correcting records in the handler is that the "user can fix the data at the source", but in the case of reading oplogs (or similar), that might not be feasible.

Contributor guide

Open the contributing guide

Research direction

Start with the mongodb-outbox example linked in the issue and reproduce the failure using the provided docker compose command with the negative timestamp. Read the resulting Kafka Connect and Debezium task error behavior; done means providing a defined way for the source connector to skip or otherwise handle the poison record without stopping.

Written by the indexing model from the issue text.

Assessment

Tech stack
docker, kafka, mongodb
Domain
backend, databases, distributed-systems
Issue type
Feature
Difficulty
5/5
Estimated time
Over a week
Activity status
Stale
Clarity
Mostly clear
Newbie friendliness
32/100

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.