dapr / dapr/components-contrib

pubsub.azure.eventhubs: checkpoint advances regardless of app response, breaking the documented at-least-once guarantee

Open
#4,561 1 comment 0 reactions 0 assignees Claimed by @gargsajal9 View on GitHub
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

Open the contributing 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

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.