dapr / dapr/components-contrib
pubsub.azure.eventhubs: checkpoint advances regardless of app response, breaking the documented at-least-once guarantee
- Dominant language
- Go
- Stars
- 602
- Forks
- 580
- Avg merge
- 4d 9h
- Merged PRs (30d)
- 6
Description
## Expected Behavior
Per the [pub/sub overview](https://docs.dapr.io/developing-applications/building-blocks/pubsub/pubsub-overview/#at-least-once-guarantee):
> Dapr guarantees at-least-once semantics for message delivery. [...] Even if the message fails to deliver, or your application crashes, Dapr attempts to redeliver the message until successful delivery.
>
> All Dapr pub/sub components support the at-least-once guarantee.
A subscriber that returns `RETRY` (or whose sidecar dies mid-processing) should therefore see the message again after a restart. `pubsub.azure.eventhubs` should not advance the consumer-group checkpoint past a message the application never successfully processed.
This is how the sibling components already behave:
- **Kafka** — [`common/component/kafka/consumer.go#L221-L223`](https://github.com/dapr/components-contrib/blob/ece1f255da3313670729c6ef11c6b65108537a5c/common/component/kafka/consumer.go#L221-L223): `session.MarkMessage()` is guarded on `err == nil`.
- **Service Bus** — per-message `CompleteMessage` / `AbandonMessage`, plus a native DLQ.
## Actual Behavior
The Event Hubs component advances the checkpoint **unconditionally**, ignoring the application's response. The consequence is **at-most-once** delivery across a sidecar restart: the unprocessed event is permanently lost, with no error surfaced to the publisher or the subscriber.
[`processEvents`](https://github.com/dapr/components-contrib/blob/ece1f255da3313670729c6ef11c6b65108537a5c/common/component/azure/eventhubs/eventhubs.go#L399-L422) — the handler's error is discarded in *both* branches, and `UpdateCheckpoint` then runs regardless:
```go
if aeh.metadata.EnableInOrderMessageDelivery {
_ = aeh.handleAsync(subscribeCtx, config.Topic, events, config.Handler) // error dropped
} else {
go aeh.handleAsync(subscribeCtx, config.Topic, events, config.Handler) // fire-and-forget
}
// Checkpointing disabled for CheckPointFrequencyPerPartition == 0
if config.CheckPointFrequencyPerPartition > 0 {
if counter%config.CheckPointFrequencyPerPartition == 0 {
err = partitionClient.UpdateCheckpoint(ctx, events[len(events)-1], nil) // runs unconditionally
```
The error is available at the call site — [`handleAsync`](https://github.com/dapr/components-contrib/blob/ece1f255da3313670729c6ef11c6b65108537a5c/common/component/azure/eventhubs/eventhubs.go#L344-L358) logs it and returns it. It is simply not consulted.
Both delivery modes are affected, to different degrees:
- `enableInOrderMessageDelivery: false` (**default**) — delivery is `go handleAsync(...)`, so the checkpoint advances without waiting for the app at all. The cursor can run arbitrarily far ahead of a stalled subscriber.
- `enableInOrderMessageDelivery: true` — the component waits for the app, which bounds the loss to one batch, but a `RETRY` still checkpoints.
**There is no workaround.** No component or subscription metadata field makes checkpointing conditional on app success, and no Resiliency policy helps — the inbound retry runs entirely inside the handler, and the checkpoint follows whether it succeeded or not. Setting `checkPointFrequencyPerPartition: 0` looks like an escape hatch but is strictly worse: with no checkpoint written, the SDK's start position falls back to `@latest`, so a restart skips everything enqueued while the sidecar was down.
## Steps to Reproduce the Problem
1. Deploy a `pubsub.azure.eventhubs` component (defaults are enough: `enableInOrderMessageDelivery` unset, `checkPointFrequencyPerPartition` unset → 1) with a single-partition hub.
2. Subscribe with an app that returns `RETRY` for every delivery of the topic — e.g. throws unconditionally.
3. Publish message **A**, and wait for the app to log that it received it and begin retrying.
4. Inspect the checkpoint blob for the partition — the sequence number has already advanced past **A**.
5. Restart (or kill) the daprd sidecar while the app is still retrying.
6. Publish message **B**.
7. **Observed:** the app receives **B** only. **A** is never redelivered and is unrecoverable.
**Expected:** **A** is redelivered, because the app never acknowledged it successfully.
## Proposed Fix
Make checkpointing conditional on acknowledgement, opt-in via subscription metadata to preserve today's behavior by default — e.g. `checkpointOnSuccess: true`.
## Release Note
RELEASE NOTE: **FIX** Azure Event Hubs pub/sub: advance the partition checkpoint only after the application acknowledges the message.
Contributor guide
Research direction
Start in common/component/azure/eventhubs/eventhubs.go, reading processEvents and handleAsync at the referenced lines. Compare the checkpoint flow with common/component/kafka/consumer.go and review the subscription metadata handling for the proposed checkpointOnSuccess option. Done means RETRY or interrupted processing does not advance the checkpoint in either delivery mode, while the documented default behavior remains compatible.
Written by the indexing model from the issue text.
Assessment
- Tech stack
- azure, go
- Domain
- backend, distributed-systems
- Issue type
- Bug
- Difficulty
- 4/5
- Estimated time
- 3-5 days
- Activity status
- Active
- Clarity
- Clearly specified
- Newbie friendliness
- 58/100