Aiven-Open / Aiven-Open/amqp-connector-for-apache-kafka

RabbitMQ messages are acknowledged before Kafka producer confirmation, causing data loss

Offen
#37 1 Kommentar 0 Reaktionen 0 zugewiesene Personen Auf GitHub ansehen
Vorherrschende Sprache
Java
Sterne
0
Forks
1
Ø Merge
3 Std. 19 Min.
Gemergte PRs (30 T.)
4

Beschreibung

# What happened?

The source connector acknowledges RabbitMQ deliveries before Kafka has persisted the corresponding records.

`AmqpFragment#getReceiver` opens a ProtonJ2 receiver with default options:

```java
connection.openReceiver(address)
```

With ProtonJ2 1.1.0, the defaults are `autoAccept=true` and `autoSettle=true`. As a result, `AmqpSourceData#getNativeItemIterator` calling `receiver.tryReceive()` schedules an `Accepted` and settled disposition before the delivery is returned. This happens before message extraction, `SourceRecord` creation, `SourceTask#poll`, and Kafka producer acknowledgement.

Neither `AmqpSourceTask` nor `framework-for-source-connectors` 0.3.0 handles `SourceTask#commitRecord`, so the RabbitMQ acknowledgement is not tied to Kafka delivery success.

This creates an at-most-once handoff and can permanently lose messages when:

- The worker stops or crashes after `tryReceive()` but before Kafka producer acknowledgement.
- The Kafka producer, converter, or serializer fails.
- Records are still buffered by the source task when it stops.
- `AmqpExtractor` catches a `ClientException` or `IOException` and returns an empty result.

Relevant locations at commit `be44dce`:

- `common/src/main/java/io/aiven/kafka/connect/amqp/common/config/AmqpFragment.java:167-170`
- `source/src/main/java/io/aiven/kafka/connect/amqp/source/AmqpSourceData.java:82-102`
- `source/src/main/java/io/aiven/kafka/connect/amqp/source/AmqpSourceTask.java:33-72`
- `source/src/main/java/io/aiven/kafka/connect/amqp/source/extractor/AmqpExtractor.java:107-124`

Minimal reproduction:

1. Publish one message to RabbitMQ.
2. Poll it through `AmqpSourceTask`.
3. Stop the task before invoking any commit callback.
4. Connect a new receiver to the same RabbitMQ address.

The message is not redelivered. In an A/B integration test, changing only the receiver to `autoAccept(false)` caused the same message ID to be redelivered, confirming the cause.

# What did you expect to happen?

RabbitMQ deliveries should remain unsettled until Kafka confirms that the corresponding record has been persisted. If the task stops or fails before that confirmation, RabbitMQ should redeliver the message.

The connector should provide at-least-once delivery. A failure between Kafka acknowledgement and RabbitMQ settlement may cause a duplicate, but it must not cause data loss.

# What else do we need to know?

Suggested fix:

- Open the receiver with `autoAccept(false)` and use manual settlement.
- Track pending source-record-to-delivery mappings.
- Call `Delivery.accept()` from `commitRecord(record, metadata)` only after a successful Kafka write.
- Release pending deliveries on failure and shutdown.
- Do not acknowledge `metadata == null` without an explicit filter, reject, or DLQ policy; it can also represent a tolerated producer failure.
- Bound unsettled deliveries using manual credit or equivalent backpressure.
- Propagate extraction errors or route them through an explicit reject/DLQ policy instead of returning an empty stream.

Acceptance criteria:

- An uncommitted delivery is redelivered after task restart.
- A successfully committed delivery is not redelivered.
- Producer and extraction failures do not remove the RabbitMQ message.
- The number of outstanding unsettled deliveries remains bounded.

Observed with the current `main` branch at `be44dce`, ProtonJ2 1.1.0, and RabbitMQ 4.2.x.

Beitragsleitfaden

Beitragsleitfaden öffnen

Bewertung

Dieses Issue wurde noch nicht bewertet.

Neue Issues direkt in Ihr Postfach

Eine kurze Übersicht über anfängerfreundliche GitHub-Issues.