[azeventhubs] Adding a section to the migration guide that shows how to migrate from older checkpoints to a newer checkpoint store Azure#20297

Adding in a migration guide section that shows how to migrate We want to make it easy for customers using the previous `azure-event-hubs-go` package to use this package (`github.com/Azure/azure-sdk-for-go/sdk/messaging/azeventhubs`).

Fixes Azure#19849

Author: richardpark-msft

sdk/messaging/azeventhubs/example_checkpoint_migration_test.go

@@ -0,0 +1,151 @@
+// Copyright (c) Microsoft Corporation. All rights reserved.
+// Licensed under the MIT License.
+
+package azeventhubs_test
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"os"
+	"strconv"
+
+	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azeventhubs"
+	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azeventhubs/checkpoints"
+	"github.com/Azure/azure-sdk-for-go/sdk/storage/azblob"
+	"github.com/Azure/azure-sdk-for-go/sdk/storage/azblob/container"
+)
+
+type LegacyCheckpoint struct {
+	PartitionID string `json:"partitionID"`
+	Epoch       int    `json:"epoch"`
+	Owner       string `json:"owner"`
+	Checkpoint  struct {
+		Offset         string `json:"offset"`
+		SequenceNumber int64  `json:"sequenceNumber"`
+		EnqueueTime    string `json:"enqueueTime"` // ": "0001-01-01T00:00:00Z"
+	} `json:"checkpoint"`
+}
+
+// Shows how to migrate from the older `github.com/Azure/azure-event-hubs-go` checkpointer to to
+// the format used by this package, `github.com/Azure/azure-sdk-for-go/sdk/messaging/azeventhubs/checkpoints/BlobStore`
+//
+// NOTE: This example is not safe to run while either the old or new checkpoint store is in-use as it doesn't
+// respect locking or ownership.
+func Example_migrateCheckpoints() {
+	// Azure Event Hubs connection string. You can get this from the Azure Portal.
+	// For example: youreventhub.servicebus.windows.net
+	var EventHubNamespace = os.Getenv("EVENTHUB_NAMESPACE")
+
+	// Name of your Event Hub that these checkpoints reference.
+	var EventHubName = os.Getenv("EVENTHUB_NAME")
+
+	// Name of your Event Hub consumer group
+	// Example: $Default
+	var EventHubConsumerGroup = os.Getenv("EVENTHUB_CONSUMER_GROUP")
+
+	// Azure Storage account connection string. You can get this from the Azure Portal.
+	// For example: DefaultEndpointsProtocol=https;AccountName=accountname;AccountKey=account-key;EndpointSuffix=core.windows.net
+	var StorageConnectionString = os.Getenv("STORAGE_CONNECTION_STRING")
+
+	// Optional: If you used `eventhub.WithPrefixInBlobPath()` configuration option for your Event Processor Host
+	// then you'll need to set this value.
+	//
+	// NOTE: This is no longer needed with the new checkpoint store as it automatically makes the path unique
+	// for each combination of eventhub + hubname + consumergroup + partition.
+	var BlobPrefix = os.Getenv("OLD_STORAGE_BLOB_PREFIX")
+
+	// Name of the checkpoint store's Azure Storage container.
+	var OldStorageContainerName = os.Getenv("OLD_STORAGE_CONTAINER_NAME")
+
+	// Name of the Azure Storage container to place new checkpoints in.
+	var NewStorageContainerName = os.Getenv("NEW_STORAGE_CONTAINER_NAME")
+
+	if EventHubNamespace == "" || EventHubName == "" || EventHubConsumerGroup == "" ||
+		StorageConnectionString == "" || OldStorageContainerName == "" || NewStorageContainerName == "" {
+		fmt.Printf("Skipping migration, missing parameters\n")
+		return
+	}
+
+	blobClient, err := azblob.NewClientFromConnectionString(StorageConnectionString, nil)
+
+	if err != nil {
+		panic(err)
+	}
+
+	oldCheckpoints, err := loadOldCheckpoints(blobClient, OldStorageContainerName, BlobPrefix)
+
+	if err != nil {
+		panic(err)
+	}
+
+	newCheckpointStore, err := checkpoints.NewBlobStore(blobClient.ServiceClient().NewContainerClient(NewStorageContainerName), nil)
+
+	if err != nil {
+		panic(err)
+	}
+
+	for _, oldCheckpoint := range oldCheckpoints {
+		newCheckpoint := azeventhubs.Checkpoint{
+			ConsumerGroup:           EventHubConsumerGroup,
+			EventHubName:            EventHubName,
+			FullyQualifiedNamespace: EventHubNamespace,
+			PartitionID:             oldCheckpoint.PartitionID,
+		}
+
+		offset, err := strconv.ParseInt(oldCheckpoint.Checkpoint.Offset, 10, 64)
+
+		if err != nil {
+			panic(err)
+		}
+
+		newCheckpoint.Offset = &offset
+		newCheckpoint.SequenceNumber = &oldCheckpoint.Checkpoint.SequenceNumber
+
+		if err := newCheckpointStore.UpdateCheckpoint(context.Background(), newCheckpoint, nil); err != nil {
+			panic(err)
+		}
+	}
+}
+
+func loadOldCheckpoints(blobClient *azblob.Client, containerName string, customBlobPrefix string) ([]*LegacyCheckpoint, error) {
+	blobPrefix := &customBlobPrefix
+
+	if customBlobPrefix == "" {
+		blobPrefix = nil
+	}
+
+	pager := blobClient.NewListBlobsFlatPager(containerName, &container.ListBlobsFlatOptions{
+		Prefix: blobPrefix,
+	})
+
+	var checkpoints []*LegacyCheckpoint
+
+	for pager.More() {
+		page, err := pager.NextPage(context.Background())
+
+		if err != nil {
+			return nil, err
+		}
+
+		for _, item := range page.Segment.BlobItems {
+			buff := [4000]byte{}
+
+			len, err := blobClient.DownloadBuffer(context.Background(), containerName, *item.Name, buff[:], nil)
+
+			if err != nil {
+				return nil, err
+			}
+
+			var legacyCheckpoint *LegacyCheckpoint
+
+			if err := json.Unmarshal(buff[0:len], &legacyCheckpoint); err != nil {
+				return nil, err
+			}
+
+			checkpoints = append(checkpoints, legacyCheckpoint)
+		}
+	}
+
+	return checkpoints, nil
+}

sdk/messaging/azeventhubs/example_consuming_events_test.go

@@ -16,8 +16,7 @@ import (
 	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azeventhubs"
 )
 
-// Example_consumingEventsUsingConsumerClient shows how to start consuming events in partitions
-// in an Event Hub.
+// Shows how to start consuming events in partitions in an Event Hub.
 //
 // If you have an Azure Storage account you can use the [Processor] type instead, which will handle
 // distributing partitions between multiple consumers. See example_processor_test.go for usage of

sdk/messaging/azeventhubs/example_processor_test.go

@@ -16,7 +16,7 @@ import (
 	"github.com/Azure/azure-sdk-for-go/sdk/storage/azblob/container"
 )
 
-// Example_consumingEventsUsingProcessor shows how to use the [Processor] type.
+// Shows how to use the [Processor] type.
 //
 // The Processor type acts as a load balancer, ensuring that partitions are divided up amongst
 // active Processor instances. You provide it with a [ConsumerClient] as well as a [CheckpointStore].

sdk/messaging/azeventhubs/migrationguide.md

@@ -101,3 +101,6 @@ producerClient.GetEventHubProperties(context.TODO(), nil)
 producerClient.GetPartitionProperties(context.TODO(), "partition-id", nil)
 ```
 
+## Migrating from a previous checkpoint store
+
+See here for an example: [link](https://github.com/Azure/azure-sdk-for-go/blob/main/sdk/messaging/azeventhubs/example_checkpoint_migration_test.go)