[Event Hubs] Checkpoint Migration Sample Tweak (Azure#29129)

The focus of these changes is to tweak the checkpoint migration sample to
account for the potential that a blob exists for ownership purposes but
does not have valid checkpoint data.  Previously, the sample would emit
a T2 checkpoint for these will potentially invalid data.

Author: jsquire

sdk/eventhub/Azure.Messaging.EventHubs.Processor/tests/Snippets/MigrationGuideSnippetsLiveTests.cs

@@ -100,10 +100,14 @@ public async Task MigrateCheckpoints()
             {
                 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>());
@@ -140,6 +144,14 @@ private IEnumerable<MigrationCheckpoint> ReadFakeLegacyCheckpoints(string fakeCo
                 ""Epoch"":1,
                 ""Offset"":""78"",
                 ""SequenceNumber"":39
+            },
+            {
+                ""PartitionId"":""3"",
+                ""Owner"":""eecd42df-a253-49d1-bb04-e5f00c106cfc"",
+                ""Token"":""6271aadb-801f-4ec7-a011-a008808a656c"",
+                ""Epoch"":1,
+                ""Offset"":""123"",
+                ""SequenceNumber"":null
             }]";
 
             return JsonSerializer.Deserialize<MigrationCheckpoint[]>(checkpointJson);
@@ -172,7 +184,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;
@@ -186,7 +202,7 @@ private class MigrationCheckpoint
         {
             public string PartitionId { get; set; }
             public string Offset { get; set; }
-            public long SequenceNumber { get; set; }
+            public long? SequenceNumber { get; set; }
         }
 
         #endregion

sdk/eventhub/Azure.Messaging.EventHubs/MigrationGuide.md

@@ -761,7 +761,7 @@ private class MigrationCheckpoint
 {
     public string PartitionId { get; set; }
     public string Offset { get; set; }
-    public long SequenceNumber { get; set; }
+    public long? SequenceNumber { get; set; }
 }
 ```
 
@@ -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;