Hacktoberfest 2026: những issue maintainer đã đánh dấu cho tháng Mười, đang mở và phù hợp người mới. Xem issue Hacktoberfest

azure-eventhub migration guide: legacy checkpoint sample writes "None" metadata for partitions that were never checkpointed

Đang mở Phù hợp với người mới
  • Có gì không ổn?
#49,723 2 bình luận 1 reaction 0 người được giao Xem trên GitHub

Maintainer thường phản hồi trong vòng 1 ngày

Chưa có ai nhận issue này.

Đánh giá

Độ khó
2/5
Thời gian dự kiến
1-3 giờ
Mức phù hợp với người mới
78/100
Loại issue
Lỗi
Độ rõ ràng
Đặc tả rõ ràng
Mức độ hoạt động
Sôi nổi
Công nghệ
azure, python
Lĩnh vực
documentation

Hướng nghiên cứu

Mở sdk/eventhub/azure-eventhub/migration_guide.md và xem lại đoạn mã chuyển đổi checkpoint cũ. Kiểm tra cách đoạn mã xử lý các mục thiếu offset hoặc sequence_number, đồng thời xác minh tiền tố sử dụng biến đã được định nghĩa EVENT_HUB_HOSTNAME. Hoàn thành khi ví dụ bỏ qua các mục không thể tạo thành checkpoint và tạo tiền tố blob từ biến đã định nghĩa.

Do mô hình lập chỉ mục viết ra từ nội dung của issue.

Mô tả

Client customer-reported Event Hubs needs-team-attention question Service Attention
Library name

azure-eventhub (migration_guide.md), azure-eventhub-checkpointstoreblob

Describe the bug

The "migrate checkpoint data from the legacy format" snippet at the end of
sdk/eventhub/azure-eventhub/migration_guide.md converts every blob in the
legacy EventProcessorHost container unconditionally:

    # Translate each legacy checkpoint, storing offset and sequence data into correct
    # blob to align with V5 BlobCheckpointStore.
    for checkpoint in legacy_checkpoints:
        metadata = {
            "offset": str(checkpoint["offset"]),
            "sequencenumber": str(checkpoint["sequence_number"]),
        }
        name = "{}{}".format(prefix, checkpoint["partition_id"])
        container_client.upload_blob(name, data="", metadata=metadata)

The legacy (1.x) EventProcessorHost creates a lease blob for every partition
up front, before anything has been checkpointed. In
azure_storage_checkpoint_manager.py (create_lease_if_not_exists_async) a
fresh AzureBlobLease() is serialized with offset = None and
sequence_number = None, so a partition that was never checkpointed has a blob
like:

{"partition_id": "3", "sequence_number": null, "owner": null, "token": null,
 "epoch": 0, "event_processor_context": null, "offset": null}

json.loads turns those into None, and the snippet writes the metadata
offset="None", sequencenumber="None".

The v5 store then can't read the container. BlobCheckpointStore.list_checkpoints
(azure-eventhub-checkpointstoreblob/.../_blobstoragecs.py, and the same line in
the -aio package) does:

                "offset": str(metadata["offset"]),
                "sequence_number": int(metadata["sequencenumber"]),

int("None") raises ValueError. Because list_checkpoints reads all
partitions in one call, a single unconverted partition makes the whole call
fail. EventProcessor catches the exception in its load-balancing loop, logs
"An error occurred while load-balancing and claiming ownership" and retries,
so the consumer stays up but never starts receiving on any partition it
newly claims.

A second, smaller problem in the same snippet: it defines

EVENT_HUB_HOSTNAME = os.environ["EVENT_HUB_HOSTNAME"]  # <your-eventhub-namespace>.servicebus.windows.net

but builds the blob prefix from a different name that is never defined:

    prefix = "{}/{}/{}/checkpoint/".format(
        EVENT_HUB_FULLY_QUALIFIED_NAMESPACE.lower(),

so as written it fails with NameError before any checkpoint is uploaded.

Why it matters

This snippet is for people moving off the 1.x processor, and having some idle
or never-checkpointed partitions is normal (new partitions, low-traffic
partitions, a consumer group that was only partly used). After running the
migration as documented, the v5 processor connects, logs warnings over and
over, and receives nothing. The error points at load balancing rather than at
the migrated blobs, so it is hard to trace back to the migration step.

Suggested fix

Skip legacy entries that have no checkpoint (a v5 processor will just use its
starting_position for those partitions), and use the variable that is
actually defined:

     prefix = "{}/{}/{}/checkpoint/".format(
-        EVENT_HUB_FULLY_QUALIFIED_NAMESPACE.lower(),
+        EVENT_HUB_HOSTNAME.lower(),
         EVENT_HUB_NAME.lower(),
         EVENT_HUB_CONSUMER_GROUP.lower(),
     )

     # Translate each legacy checkpoint, storing offset and sequence data into correct
     # blob to align with V5 BlobCheckpointStore.
     for checkpoint in legacy_checkpoints:
+        # Partitions that were never checkpointed have no offset/sequence number;
+        # skip them so the V5 processor falls back to its starting position.
+        if checkpoint.get("offset") is None or checkpoint.get("sequence_number") is None:
+            continue
         metadata = {
             "offset": str(checkpoint["offset"]),
             "sequencenumber": str(checkpoint["sequence_number"]),
         }

Both fields are skipped together rather than writing only offset (as .NET
does), because the Python BlobCheckpointStore requires sequencenumber to be
present — metadata["sequencenumber"] would raise KeyError otherwise.

Upstream reference

The .NET migration guide had the same unconditional conversion and fixed it in
Azure/azure-sdk-for-net@2597c2265a ("[Event Hubs] Checkpoint Migration Sample
Tweak", Azure/azure-sdk-for-net#29129, 2022-06-06) — legacy entries with no
offset are skipped when reading, and a missing sequence number is no longer
written as a value:

--- a/sdk/eventhub/Azure.Messaging.EventHubs/MigrationGuide.md
+++ b/sdk/eventhub/Azure.Messaging.EventHubs/MigrationGuide.md
@@ -818,10 +818,14 @@ foreach (var checkpoint in legacyCheckpoints)
 {
     var metadata = new Dictionary<string, string>()
     {
-        { offsetKey, checkpoint.Offset.ToString(CultureInfo.InvariantCulture) },
-        { sequenceKey, checkpoint.SequenceNumber.ToString(CultureInfo.InvariantCulture) }
+        { offsetKey, checkpoint.Offset.ToString(CultureInfo.InvariantCulture) }
     };
 
+    if (checkpoint.SequenceNumber.HasValue)
+    {
+        metadata[sequenceKey] = checkpoint.SequenceNumber.Value.ToString(CultureInfo.InvariantCulture);
+    }
+
     BlobClient blobClient = storageClient.GetBlobClient($"{ prefix }{ checkpoint.PartitionId }");
 
     using var content = new MemoryStream(Array.Empty<byte>());
@@ -857,7 +861,11 @@ private async Task<List<MigrationCheckpoint>> ReadLegacyCheckpoints(
         await (storageClient.GetBlobClient(blobItem.Name)).DownloadToAsync(blobContentStream);
 
         var checkpoint = JsonSerializer.Deserialize<MigrationCheckpoint>(Encoding.UTF8.GetString(blobContentStream.ToArray()));
-        checkpoints.Add(checkpoint);
+
+        if (!string.IsNullOrEmpty(checkpoint.Offset))
+        {
+            checkpoints.Add(checkpoint);
+        }
     }
 
     return checkpoints;

The idea carries over directly — don't turn an empty legacy lease into a
checkpoint. The only Python-specific difference is that the v5 store here
needs both fields, so the entry should be skipped whenever either one is
missing.

I can submit a PR for this.

Ngôn ngữ chính
Python
Star
5.6k
Fork
3.4k
Merge trung bình
2 ngày 14 giờ
Pull request đã merge (30 ngày)
200

Chuẩn bị môi trường

Mở trong Codespaces

Khởi chạy dev container của dự án ngay trên trình duyệt, bằng tài khoản GitHub của bạn.

Bắt đầu từ đâu

  1. Đọc hết issue, rồi đọc hướng dẫn đóng góp của dự án.
  2. Bình luận trên issue rằng bạn sẽ nhận — tránh hai người làm cùng một việc.
  3. Fork repository và làm thay đổi trên một nhánh.
  4. Mở pull request có tham chiếu số hiệu của issue.

Issue khác của Azure/azure-sdk-for-python

Tất cả issue của Azure/azure-sdk-for-python

Issue tương tự

Thêm issue về Python

Nhận issue mới trong hộp thư của bạn

Bản tóm tắt ngắn những issue GitHub phù hợp với người mới.