Skip to content

#46752: Fix missing offset assignment logic in checkpoint store #46753

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
kamil-di wants to merge 1 commit into
base: main
Choose a base branch
from
Open

#46752: Fix missing offset assignment logic in checkpoint store #46753

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 2 additions & 1 deletion sdk/eventhubs/azure-messaging-eventhubs-checkpointstore-blob/CHANGELOG.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -8,6 +8,8 @@

### Bugs Fixed

- Fixed `updateCheckpoint` to properly handle checkpoints when `offsetString` is available but `offset` is null, preventing potential data loss during checkpoint updates.

### Other Changes

## 1.21.2 (2025-10-27)
Expand Down Expand Up @@ -617,4 +619,3 @@ store checkpoints and balance partition load among all instances of Event Proces

- Initial offset provider for each partition is not implemented.
- Interoperability with Event Processors of other language SDKs like Python is not supported.

15 changes: 12 additions & 3 deletions ...src/main/java/com/azure/messaging/eventhubs/checkpointstore/blob/BlobCheckpointStore.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -243,9 +243,12 @@ private Mono<PartitionOwnership> updateOwnershipETag(Response<?> response, Parti
*/
@Override
public Mono<Void> updateCheckpoint(Checkpoint checkpoint) {
if (checkpoint == null || (checkpoint.getSequenceNumber() == null && checkpoint.getOffset() == null)) {
if (checkpoint == null
|| (checkpoint.getSequenceNumber() == null
&& checkpoint.getOffset() == null
&& CoreUtils.isNullOrEmpty(checkpoint.getOffsetString()))) {
throw LOGGER.logExceptionAsWarning(Exceptions.propagate(new IllegalStateException(
"Both sequence number and offset cannot be null when updating a checkpoint")));
"At least one of sequence number or offset information (offset or offsetString) must be provided when updating a checkpoint")));
}

String partitionId = checkpoint.getPartitionId();
Expand All @@ -259,8 +262,14 @@ public Mono<Void> updateCheckpoint(Checkpoint checkpoint) {
String sequenceNumber
= checkpoint.getSequenceNumber() == null ? null : String.valueOf(checkpoint.getSequenceNumber());

String offset;
if (CoreUtils.isNullOrEmpty(checkpoint.getOffsetString())) {
offset = Objects.toString(checkpoint.getOffset(), null);
} else {
offset = checkpoint.getOffsetString();
}
metadata.put(SEQUENCE_NUMBER, sequenceNumber);
metadata.put(OFFSET, checkpoint.getOffsetString());
metadata.put(OFFSET, offset);
BlobAsyncClient blobAsyncClient = blobClients.get(blobName);

return blobAsyncClient.exists().flatMap(exists -> {
Expand Down