Repository: Azure/azure-sdk-for-go
Pull request #24489 (2025-04-23 02:28:28)

Updating CODEOWNERS with new owners for azopenai


Patch:
diff --git a/.github/CODEOWNERS b/.github/CODEOWNERS
index b8022de1d9..c84258d9d5 100644
--- a/.github/CODEOWNERS
+++ b/.github/CODEOWNERS
@@ -27,6 +27,11 @@
 # PRLabel: %OpenAI
 /sdk/ai                       @richardpark-msft @jhendrixMSFT @chlowell
 
+# AzureSDKOwners: @richardpark-msft
+# ServiceLabel: %OpenAI
+# PRLabel: %OpenAI
+/sdk/ai/azopenai                       @richardpark-msft @achauhan-scc @kingernupur @glecaros
+
 # AzureSDKOwners: @jhendrixMSFT
 # ServiceLabel: %Internal
 # PRLabel: %Internal
@@ -89,8 +94,9 @@
 /sdk/resourcemanager/         @ArcturusZhang @lirenhe @tadelesh @JiaqiZhang-Dev
 
 # AzureSDKOwners: @jhendrixMSFT
-# ServiceOwner: @tanyasethi-msft
-# ServiceLabel: %Azure.Core
+# ServiceOwners: @tanyasethi-msft
+# ServiceLabel: %Storage
+
 # PRLabel: %Storage
 /sdk/storage/                 @souravgupta-msft @jhendrixMSFT @gapra-msft @vibhansa-msft @tanyasethi-msft
 
diff --git a/.github/workflows/post-apiview.yml b/.github/workflows/post-apiview.yml
new file mode 100644
index 0000000000..4306a63ace
--- /dev/null
+++ b/.github/workflows/post-apiview.yml
@@ -0,0 +1,29 @@
+name: After APIView
+
+on:
+  check_run:
+    types: [completed]
+    
+permissions:
+  pull-requests: write
+  contents: read
+
+jobs:
+  post-apiview:
+    name: After APIView
+    runs-on: ubuntu-24.04
+    if: |
+      github.event.check_run.check_suite.app.name == 'Azure Pipelines' && (
+      contains(github.event.check_run.name, 'Build Build') ||
+      contains(github.event.check_run.name, 'Build Analyze') )
+    steps:
+      - name: Checkout
+        uses: actions/checkout@v4
+        with:
+          sparse-checkout: 'eng/common'
+    
+      - name: Create APIView Comment on PR
+        run: |
+          . "eng/common/scripts/Helpers/ApiView-Helpers.ps1"
+          Set-ApiViewCommentForRelatedIssues -HeadCommitish ${{ github.event.check_run.head_sha }} -AuthToken ${{ secrets.GITHUB_TOKEN }}
+        shell: pwsh
diff --git a/sdk/data/azcosmos/cosmos_container_query_engine.go b/sdk/data/azcosmos/cosmos_container_query_engine.go
index 62d4bc009b..0b0711989d 100644
--- a/sdk/data/azcosmos/cosmos_container_query_engine.go
+++ b/sdk/data/azcosmos/cosmos_container_query_engine.go
@@ -122,7 +122,7 @@ func (c *ContainerClient) executeQueryWithEngine(queryEngine queryengine.QueryEn
 					}, nil
 				}
 				// Fetch more data from the pipeline
-				items, requests, err := queryPipeline.NextBatch()
+				result, err := queryPipeline.Run()
 				if err != nil {
 					queryPipeline.Close()
 					return QueryItemsResponse{}, err
@@ -130,16 +130,16 @@ func (c *ContainerClient) executeQueryWithEngine(queryEngine queryengine.QueryEn
 
 				// If we got items, we can return them, and we should do so now, to avoid making unnecessary requests.
 				// Even if there are requests in the queue, the pipeline should return the same requests again on the next call to NextBatch.
-				if len(items) > 0 {
+				if len(result.Items) > 0 {
 					return QueryItemsResponse{
 						Response: lastResponse,
-						Items:    items,
+						Items:    result.Items,
 					}, nil
 				}
 
 				// If we didn't have any items to return, we need to make requests for the items in the queue.
 				// If there are no requests, the pipeline should return true for IsComplete, so we'll stop on the next iteration.
-				for _, request := range requests {
+				for _, request := range result.Requests {
 					// Make the single-partition query request
 					qryRequest := queryRequest(request) // Cast to our type, which has toHeaders defined on it.
 					azResponse, err := c.database.client.sendQueryRequest(
diff --git a/sdk/data/azcosmos/internal/mock_query_engine.go b/sdk/data/azcosmos/internal/mock_query_engine.go
index 0e28bcb775..1d98f96703 100644
--- a/sdk/data/azcosmos/internal/mock_query_engine.go
+++ b/sdk/data/azcosmos/internal/mock_query_engine.go
@@ -139,9 +139,9 @@ func (m *MockQueryPipeline) IsComplete() bool {
 }
 
 // NextBatch returns the next batch of items from the pipeline, as well as any requests needed to collect more data.
-func (m *MockQueryPipeline) NextBatch() ([][]byte, []queryengine.QueryRequest, error) {
+func (m *MockQueryPipeline) Run() (*queryengine.PipelineResult, error) {
 	if m.IsClosed {
-		return nil, nil, fmt.Errorf("pipeline is closed")
+		return nil, fmt.Errorf("pipeline is closed")
 	}
 
 	items := make([][]byte, 0)
@@ -154,7 +154,11 @@ func (m *MockQueryPipeline) NextBatch() ([][]byte, []queryengine.QueryRequest, e
 		for i := range m.partitionState {
 			// If any partition hasn't started yet, we can't return any items.
 			if !m.partitionState[i].started {
-				return nil, m.getRequests(), nil
+				return &queryengine.PipelineResult{
+					IsCompleted: false,
+					Items:       nil,
+					Requests:    m.getRequests(),
+				}, nil
 			}
 
 			if m.partitionState[i].IsExhausted() {
@@ -175,7 +179,7 @@ func (m *MockQueryPipeline) NextBatch() ([][]byte, []queryengine.QueryRequest, e
 			// Add the item to the result set and remove it from the queue.
 			item, err := lowestPartition.PopItem()
 			if err != nil {
-				return nil, nil, err
+				return nil, err
 			}
 			items = append(items, item)
 		}
@@ -190,7 +194,11 @@ func (m *MockQueryPipeline) NextBatch() ([][]byte, []queryengine.QueryRequest, e
 		m.completed = true
 	}
 
-	return items, requests, nil
+	return &queryengine.PipelineResult{
+		IsCompleted: m.completed,
+		Items:       items,
+		Requests:    requests,
+	}, nil
 }
 
 // getRequests returns a list of all the QueryRequests that are needed to get the next batch of items.
diff --git a/sdk/data/azcosmos/unstable/queryengine/cosmos_query_engine.go b/sdk/data/azcosmos/unstable/queryengine/cosmos_query_engine.go
index 706493c00a..8750c0a4d9 100644
--- a/sdk/data/azcosmos/unstable/queryengine/cosmos_query_engine.go
+++ b/sdk/data/azcosmos/unstable/queryengine/cosmos_query_engine.go
@@ -56,9 +56,8 @@ type QueryPipeline interface {
 	Query() string
 	// IsComplete gets a boolean indicating if the pipeline has concluded
 	IsComplete() bool
-	// NextBatch gets the next batch of items, which will be empty if there are no more items in the buffer, and the next set of QueryRequests which must be fulfilled, which will be empty if there are no more requests.
-	// If both the items and requests are empty, the pipeline has concluded.
-	NextBatch() ([][]byte, []QueryRequest, error)
+	// Run executes a single turn of the pipeline, yielding a PipelineResult containing the items and requests for more data.
+	Run() (*PipelineResult, error)
 	// ProvideData provides more data for a given partition key range ID, using data retrieved from the server in response to making a DataRequest.
 	ProvideData(data QueryResult) error
 	// Close frees the resources associated with the pipeline.
diff --git a/sdk/messaging/azeventhubs/CHANGELOG.md b/sdk/messaging/azeventhubs/CHANGELOG.md
index c8f7c219df..c24c73e65b 100644
--- a/sdk/messaging/azeventhubs/CHANGELOG.md
+++ b/sdk/messaging/azeventhubs/CHANGELOG.md
@@ -1,11 +1,16 @@
 # Release History
 
-## 1.3.3-beta.1 (Unreleased)
+## 2.0.0-beta.1 (Unreleased)
 
 ### Features Added
 
+- Support for Event Hubs Geo-Replication.
+
 ### Breaking Changes
 
+- This package is compatible with azeventhubs v1, with one difference - Checkpoint.Offset and ReceivedEventData.Offset's type have been changed to a string (from an integer). 
+  This change does NOT affect any stored checkpoints. Most customers will be unaffected by this change and can safely upgrade.
+
 ### Bugs Fixed
 
 ### Other Changes
diff --git a/sdk/messaging/azeventhubs/checkpoint_store.go b/sdk/messaging/azeventhubs/checkpoint_store.go
index 83c1c3e54f..7623fa433c 100644
--- a/sdk/messaging/azeventhubs/checkpoint_store.go
+++ b/sdk/messaging/azeventhubs/checkpoint_store.go
@@ -45,8 +45,8 @@ type Checkpoint struct {
 	FullyQualifiedNamespace string
 	PartitionID             string
 
-	Offset         *int64 // the last succesfully processed Offset.
-	SequenceNumber *int64 // the last succesfully processed SequenceNumber.
+	Offset         *string // the last succesfully processed Offset.
+	SequenceNumber *int64  // the last succesfully processed SequenceNumber.
 }
 
 // ListCheckpointsOptions contains optional parameters for the ListCheckpoints function
diff --git a/sdk/messaging/azeventhubs/checkpoints/blob_store.go b/sdk/messaging/azeventhubs/checkpoints/blob_store.go
index ece760739e..7e1e7a8712 100644
--- a/sdk/messaging/azeventhubs/checkpoints/blob_store.go
+++ b/sdk/messaging/azeventhubs/checkpoints/blob_store.go
@@ -16,7 +16,7 @@ import (
 	"github.com/Azure/azure-sdk-for-go/sdk/azcore/streaming"
 	"github.com/Azure/azure-sdk-for-go/sdk/azcore/to"
 	"github.com/Azure/azure-sdk-for-go/sdk/internal/log"
-	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azeventhubs"
+	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azeventhubs/v2"
 	"github.com/Azure/azure-sdk-for-go/sdk/storage/azblob/blob"
 	"github.com/Azure/azure-sdk-for-go/sdk/storage/azblob/bloberror"
 	"github.com/Azure/azure-sdk-for-go/sdk/storage/azblob/blockblob"
@@ -324,13 +324,7 @@ func updateCheckpoint(metadata map[string]*string, destCheckpoint *azeventhubs.C
 		return errors.New("offset is missing from metadata")
 	}
 
-	offset, err := strconv.ParseInt(*offsetStr, 10, 64)
-
-	if err != nil {
-		return fmt.Errorf("offset could not be parsed as an int64: %s", err.Error())
-	}
-
-	destCheckpoint.Offset = &offset
+	destCheckpoint.Offset = offsetStr
 	destCheckpoint.SequenceNumber = &sequenceNumber
 	return nil
 }
@@ -343,7 +337,7 @@ func newCheckpointBlobMetadata(cpd azeventhubs.Checkpoint) map[string]*string {
 	}
 
 	if cpd.Offset != nil {
-		m["offset"] = to.Ptr(strconv.FormatInt(*cpd.Offset, 10))
+		m["offset"] = cpd.Offset
 	}
 
 	return m
diff --git a/sdk/messaging/azeventhubs/connection_string_properties.go b/sdk/messaging/azeventhubs/connection_string_properties.go
index d28e837e23..616b496cff 100644
--- a/sdk/messaging/azeventhubs/connection_string_properties.go
+++ b/sdk/messaging/azeventhubs/connection_string_properties.go
@@ -3,7 +3,7 @@
 
 package azeventhubs
 
-import "github.com/Azure/azure-sdk-for-go/sdk/messaging/azeventhubs/internal/exported"
+import "github.com/Azure/azure-sdk-for-go/sdk/messaging/azeventhubs/v2/internal/exported"
 
 // ConnectionStringProperties are the properties of a connection string
 // as returned by [ParseConnectionString].
diff --git a/sdk/messaging/azeventhubs/consumer_client.go b/sdk/messaging/azeventhubs/consumer_client.go
index ad4d04d97d..14399b3191 100644
--- a/sdk/messaging/azeventhubs/consumer_client.go
+++ b/sdk/messaging/azeventhubs/consumer_client.go
@@ -11,8 +11,8 @@ import (
 
 	"github.com/Azure/azure-sdk-for-go/sdk/azcore"
 	"github.com/Azure/azure-sdk-for-go/sdk/internal/uuid"
-	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azeventhubs/internal"
-	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azeventhubs/internal/amqpwrap"
+	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azeventhubs/v2/internal"
+	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azeventhubs/v2/internal/amqpwrap"
 )
 
 // ConsumerClientOptions configures optional parameters for a ConsumerClient.
diff --git a/sdk/messaging/azeventhubs/error.go b/sdk/messaging/azeventhubs/error.go
index 39a7eaa016..597bc60a02 100644
--- a/sdk/messaging/azeventhubs/error.go
+++ b/sdk/messaging/azeventhubs/error.go
@@ -3,7 +3,7 @@
 
 package azeventhubs
 
-import "github.com/Azure/azure-sdk-for-go/sdk/messaging/azeventhubs/internal/exported"
+import "github.com/Azure/azure-sdk-for-go/sdk/messaging/azeventhubs/v2/internal/exported"
 
 // Error represents an Event Hub specific error.
 // NOTE: the Code is considered part of the published API but the message that
diff --git a/sdk/messaging/azeventhubs/event_data.go b/sdk/messaging/azeventhubs/event_data.go
index 00b89a3ca0..9db19dad77 100644
--- a/sdk/messaging/azeventhubs/event_data.go
+++ b/sdk/messaging/azeventhubs/event_data.go
@@ -5,11 +5,10 @@ package azeventhubs
 
 import (
 	"errors"
-	"strconv"
 	"time"
 
 	"github.com/Azure/azure-sdk-for-go/sdk/azcore/to"
-	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azeventhubs/internal/eh"
+	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azeventhubs/v2/internal/eh"
 	"github.com/Azure/go-amqp"
 )
 
@@ -52,7 +51,7 @@ type ReceivedEventData struct {
 	PartitionKey *string
 
 	// Offset is the offset of the event.
-	Offset int64
+	Offset string
 
 	// RawAMQPMessage is the AMQP message, as received by the client. This can be useful to get access
 	// to properties that are not exposed by ReceivedEventData such as payloads encoded into the
@@ -176,10 +175,8 @@ func updateFromAMQPAnnotations(src *amqp.Message, dest *ReceivedEventData) error
 			return errors.New("enqueued time cannot be converted to a time.Time")
 		case offsetNumberAnnotation:
 			if offsetStr, ok := v.(string); ok {
-				if offset, err := strconv.ParseInt(offsetStr, 10, 64); err == nil {
-					dest.Offset = offset
-					continue
-				}
+				dest.Offset = offsetStr
+				continue
 			}
 			return errors.New("offset cannot be converted to an int64")
 		default:
diff --git a/sdk/messaging/azeventhubs/event_data_batch.go b/sdk/messaging/azeventhubs/event_data_batch.go
index edc6517b90..b3ff8e834d 100644
--- a/sdk/messaging/azeventhubs/event_data_batch.go
+++ b/sdk/messaging/azeventhubs/event_data_batch.go
@@ -9,8 +9,8 @@ import (
 	"sync"
 
 	"github.com/Azure/azure-sdk-for-go/sdk/internal/uuid"
-	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azeventhubs/internal"
-	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azeventhubs/internal/amqpwrap"
+	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azeventhubs/v2/internal"
+	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azeventhubs/v2/internal/amqpwrap"
 	"github.com/Azure/go-amqp"
 )
 
diff --git a/sdk/messaging/azeventhubs/go.mod b/sdk/messaging/azeventhubs/go.mod
index dba49fa5a9..f85a311610 100644
--- a/sdk/messaging/azeventhubs/go.mod
+++ b/sdk/messaging/azeventhubs/go.mod
@@ -1,11 +1,11 @@
-module github.com/Azure/azure-sdk-for-go/sdk/messaging/azeventhubs
+module github.com/Azure/azure-sdk-for-go/sdk/messaging/azeventhubs/v2
 
 go 1.23.0
 
 require (
 	github.com/Azure/azure-sdk-for-go/sdk/azcore v1.18.0
-	github.com/Azure/azure-sdk-for-go/sdk/azidentity v1.8.2
-	github.com/Azure/azure-sdk-for-go/sdk/internal v1.11.0
+	github.com/Azure/azure-sdk-for-go/sdk/azidentity v1.9.0
+	github.com/Azure/azure-sdk-for-go/sdk/internal v1.11.1
 	github.com/Azure/azure-sdk-for-go/sdk/resourcemanager/eventhub/armeventhub v1.3.0
 	github.com/Azure/azure-sdk-for-go/sdk/storage/azblob v1.6.0
 	github.com/Azure/go-amqp v1.4.0
@@ -24,9 +24,9 @@ require (
 	github.com/kylelemons/godebug v1.1.0 // indirect
 	github.com/pkg/browser v0.0.0-20240102092130-5ac0b6a4141c // indirect
 	github.com/pmezard/go-difflib v1.0.0 // indirect
-	golang.org/x/crypto v0.36.0 // indirect
-	golang.org/x/net v0.38.0 // indirect
-	golang.org/x/sys v0.31.0 // indirect
-	golang.org/x/text v0.23.0 // indirect
+	golang.org/x/crypto v0.37.0 // indirect
+	golang.org/x/net v0.39.0 // indirect
+	golang.org/x/sys v0.32.0 // indirect
+	golang.org/x/text v0.24.0 // indirect
 	gopkg.in/yaml.v3 v3.0.1 // indirect
 )
diff --git a/sdk/messaging/azeventhubs/go.sum b/sdk/messaging/azeventhubs/go.sum
index e94c0eefe0..168cbaceff 100644
--- a/sdk/messaging/azeventhubs/go.sum
+++ b/sdk/messaging/azeventhubs/go.sum
@@ -1,11 +1,11 @@
 github.com/Azure/azure-sdk-for-go/sdk/azcore v1.18.0 h1:Gt0j3wceWMwPmiazCa8MzMA0MfhmPIz0Qp0FJ6qcM0U=
 github.com/Azure/azure-sdk-for-go/sdk/azcore v1.18.0/go.mod h1:Ot/6aikWnKWi4l9QB7qVSwa8iMphQNqkWALMoNT3rzM=
-github.com/Azure/azure-sdk-for-go/sdk/azidentity v1.8.2 h1:F0gBpfdPLGsw+nsgk6aqqkZS1jiixa5WwFe3fk/T3Ys=
-github.com/Azure/azure-sdk-for-go/sdk/azidentity v1.8.2/go.mod h1:SqINnQ9lVVdRlyC8cd1lCI0SdX4n2paeABd2K8ggfnE=
+github.com/Azure/azure-sdk-for-go/sdk/azidentity v1.9.0 h1:OVoM452qUFBrX+URdH3VpR299ma4kfom0yB0URYky9g=
+github.com/Azure/azure-sdk-for-go/sdk/azidentity v1.9.0/go.mod h1:kUjrAo8bgEwLeZ/CmHqNl3Z/kPm7y6FKfxxK0izYUg4=
 github.com/Azure/azure-sdk-for-go/sdk/azidentity/cache v0.3.2 h1:yz1bePFlP5Vws5+8ez6T3HWXPmwOK7Yvq8QxDBD3SKY=
 github.com/Azure/azure-sdk-for-go/sdk/azidentity/cache v0.3.2/go.mod h1:Pa9ZNPuoNu/GztvBSKk9J1cDJW6vk/n0zLtV4mgd8N8=
-github.com/Azure/azure-sdk-for-go/sdk/internal v1.11.0 h1:Bg8m3nq/X1DeePkAbCfb6ml6F3F0IunEhE8TMh+lY48=
-github.com/Azure/azure-sdk-for-go/sdk/internal v1.11.0/go.mod h1:j2chePtV91HrC22tGoRX3sGY42uF13WzmmV80/OdVAA=
+github.com/Azure/azure-sdk-for-go/sdk/internal v1.11.1 h1:FPKJS1T+clwv+OLGt13a8UjqeRuh0O4SJ3lUriThc+4=
+github.com/Azure/azure-sdk-for-go/sdk/internal v1.11.1/go.mod h1:j2chePtV91HrC22tGoRX3sGY42uF13WzmmV80/OdVAA=
 github.com/Azure/azure-sdk-for-go/sdk/resourcemanager/eventhub/armeventhub v1.3.0 h1:4hGvxD72TluuFIXVr8f4XkKZfqAa7Pj61t0jmQ7+kes=
 github.com/Azure/azure-sdk-for-go/sdk/resourcemanager/eventhub/armeventhub v1.3.0/go.mod h1:TSH7DcFItwAufy0Lz+Ft2cyopExCpxbOxI5SkH4dRNo=
 github.com/Azure/azure-sdk-for-go/sdk/resourcemanager/internal/v3 v3.1.0 h1:2qsIIvxVT+uE6yrNldntJKlLRgxGbZ85kgtz5SNBhMw=
@@ -42,8 +42,8 @@ github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0=
 github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
 github.com/joho/godotenv v1.5.1 h1:7eLL/+HRGLY0ldzfGMeQkb7vMd0as4CfYvUVzLqw0N0=
 github.com/joho/godotenv v1.5.1/go.mod h1:f4LDr5Voq0i2e/R5DDNOoa2zzDfwtkZa6DnEwAbqwq4=
-github.com/keybase/go-keychain v0.0.0-20231219164618-57a3676c3af6 h1:IsMZxCuZqKuao2vNdfD82fjjgPLfyHLpR41Z88viRWs=
-github.com/keybase/go-keychain v0.0.0-20231219164618-57a3676c3af6/go.mod h1:3VeWNIJaW+O5xpRQbPp0Ybqu1vJd/pm7s2F473HRrkw=
+github.com/keybase/go-keychain v0.0.1 h1:way+bWYa6lDppZoZcgMbYsvC7GxljxrskdNInRtuthU=
+github.com/keybase/go-keychain v0.0.1/go.mod h1:PdEILRW3i9D8JcdM+FmY6RwkHGnhHxXwkPPMeUgOK1k=
 github.com/kr/pretty v0.3.1 h1:flRD4NNwYAUpkphVc1HcthR4KEIFJ65n8Mw5qdRn3LE=
 github.com/kr/pretty v0.3.1/go.mod h1:hoEshYVHaxMs3cyo3Yncou5ZscifuDolrwPKZanG3xk=
 github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY=
@@ -54,8 +54,8 @@ github.com/pkg/browser v0.0.0-20240102092130-5ac0b6a4141c h1:+mdjkGKdHQG3305AYmd
 github.com/pkg/browser v0.0.0-20240102092130-5ac0b6a4141c/go.mod h1:7rwL4CYBLnjLxUqIJNnCWiEdr3bn6IUYi15bNlnbCCU=
 github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
 github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
-github.com/redis/go-redis/v9 v9.7.0 h1:HhLSs+B6O021gwzl+locl0zEDnyNkxMtf/Z3NNBMa9E=
-github.com/redis/go-redis/v9 v9.7.0/go.mod h1:f6zhXITC7JUJIlPEiBOTXxJgPLdZcA93GewI7inzyWw=
+github.com/redis/go-redis/v9 v9.7.3 h1:YpPyAayJV+XErNsatSElgRZZVCwXX9QzkKYNvO7x0wM=
+github.com/redis/go-redis/v9 v9.7.3/go.mod h1:bGUrSggJ9X9GUmZpZNEOQKaANxSGgOEBRltRTZHSvrA=
 github.com/rogpeppe/go-internal v1.12.0 h1:exVL4IDcn6na9z1rAb56Vxr+CgyK3nn3O+epU5NdKM8=
 github.com/rogpeppe/go-internal v1.12.0/go.mod h1:E+RYuTGaKKdloAfM02xzb0FW3Paa99yedzYV+kq4uf4=
 github.com/stretchr/testify v1.10.0 h1:Xv5erBjTwe/5IxqUQTdXv5kgmIvbHo3QQyRwhJsOfJA=
@@ -63,14 +63,14 @@ github.com/stretchr/testify v1.10.0/go.mod h1:r2ic/lqez/lEtzL7wO/rwa5dbSLXVDPFyf
 github.com/yuin/goldmark v1.3.5/go.mod h1:mwnBkeHKe2W/ZEtQ+71ViKU8L12m81fl3OWwC1Zlc8k=
 golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w=
 golang.org/x/crypto v0.0.0-20191011191535-87dc89f01550/go.mod h1:yigFU9vqHzYiE8UmvKecakEJjdnWj3jj499lnFckfCI=
-golang.org/x/crypto v0.36.0 h1:AnAEvhDddvBdpY+uR+MyHmuZzzNqXSe/GvuDeob5L34=
-golang.org/x/crypto v0.36.0/go.mod h1:Y4J0ReaxCR1IMaabaSMugxJES1EpwhBHhv2bDHklZvc=
+golang.org/x/crypto v0.37.0 h1:kJNSjF/Xp7kU0iB2Z+9viTPMW4EqqsrywMXLJOOsXSE=
+golang.org/x/crypto v0.37.0/go.mod h1:vg+k43peMZ0pUMhYmVAWysMK35e6ioLh3wB8ZCAfbVc=
 golang.org/x/mod v0.4.2/go.mod h1:s0Qsj1ACt9ePp/hMypM3fl4fZqREWJwdYDEqhRiZZUA=
 golang.org/x/net v0.0.0-20190404232315-eb5bcb51f2a3/go.mod h1:t9HGtf8HONx5eT2rtn7q6eTqICYqUVnKs3thJo3Qplg=
 golang.org/x/net v0.0.0-20190620200207-3b0461eec859/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
 golang.org/x/net v0.0.0-20210405180319-a5a99cb37ef4/go.mod h1:p54w0d4576C0XHj96bSt6lcn1PtDYWL6XObtHCRCNQM=
-golang.org/x/net v0.38.0 h1:vRMAPTMaeGqVhG5QyLJHqNDwecKTomGeqbnfZyKlBI8=
-golang.org/x/net v0.38.0/go.mod h1:ivrbrMbzFq5J41QOQh0siUuly180yBYtLp+CKbEaFx8=
+golang.org/x/net v0.39.0 h1:ZCu7HMWDxpXpaiKdhzIfaltL9Lp31x/3fCP11bc6/fY=
+golang.org/x/net v0.39.0/go.mod h1:X7NRbYVEA+ewNkCNyJ513WmMdQ3BineSwVtN2zD/d+E=
 golang.org/x/sync v0.0.0-20190423024810-112230192c58/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sync v0.0.0-20210220032951-036812b2e83c/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
@@ -79,13 +79,13 @@ golang.org/x/sys v0.0.0-20201119102817-f84b799fce68/go.mod h1:h1NjWce9XRLGQEsW7w
 golang.org/x/sys v0.0.0-20210330210617-4fbd30eecc44/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20210510120138-977fb7262007/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.1.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
-golang.org/x/sys v0.31.0 h1:ioabZlmFYtWhL+TRYpcnNlLwhyxaM9kWTDEmfnprqik=
-golang.org/x/sys v0.31.0/go.mod h1:BJP2sWEmIv4KK5OTEluFJCKSidICx8ciO85XgH3Ak8k=
+golang.org/x/sys v0.32.0 h1:s77OFDvIQeibCmezSnk/q6iAfkdiQaJi4VzroCFrN20=
+golang.org/x/sys v0.32.0/go.mod h1:BJP2sWEmIv4KK5OTEluFJCKSidICx8ciO85XgH3Ak8k=
 golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo=
 golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
 golang.org/x/text v0.3.3/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ=
-golang.org/x/text v0.23.0 h1:D71I7dUrlY+VX0gQShAThNGHFxZ13dGLBHQLVl1mJlY=
-golang.org/x/text v0.23.0/go.mod h1:/BLNzu4aZCJ1+kcD0DNRotWKage4q2rGVAg4o22unh4=
+golang.org/x/text v0.24.0 h1:dd5Bzh4yt5KYA8f9CJHCP4FB4D51c2c6JvN37xJJkJ0=
+golang.org/x/text v0.24.0/go.mod h1:L8rBsPeo2pSS+xqN0d5u2ikmjtmoJbDBT1b7nHvFCdU=
 golang.org/x/tools v0.0.0-20180917221912-90fa682c2a6e/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ=
 golang.org/x/tools v0.0.0-20191119224855-298f0cb1881e/go.mod h1:b+2E5dAYhXwXZwtnZ6UAqBI28+e2cm9otk0dWdXHAEo=
 golang.org/x/tools v0.1.1/go.mod h1:o0xws9oXOQQZyjljx8fwUC0k7L1pTE6eaCbjGeHmOkk=
diff --git a/sdk/messaging/azeventhubs/internal/amqpInterfaces.go b/sdk/messaging/azeventhubs/internal/amqpInterfaces.go
index f6ea7f0cc3..2d167eacfe 100644
--- a/sdk/messaging/azeventhubs/internal/amqpInterfaces.go
+++ b/sdk/messaging/azeventhubs/internal/amqpInterfaces.go
@@ -6,7 +6,7 @@ package internal
 import (
 	"context"
 
-	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azeventhubs/internal/amqpwrap"
+	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azeventhubs/v2/internal/amqpwrap"
 )
 
 type AMQPReceiver = amqpwrap.AMQPReceiver
diff --git a/sdk/messaging/azeventhubs/internal/amqp_fakes.go b/sdk/messaging/azeventhubs/internal/amqp_fakes.go
index a08b084f76..deb1fc2323 100644
--- a/sdk/messaging/azeventhubs/internal/amqp_fakes.go
+++ b/sdk/messaging/azeventhubs/internal/amqp_fakes.go
@@ -6,7 +6,7 @@ package internal
 import (
 	"context"
 
-	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azeventhubs/internal/amqpwrap"
+	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azeventhubs/v2/internal/amqpwrap"
 	"github.com/Azure/go-amqp"
 )
 
diff --git a/sdk/messaging/azeventhubs/internal/cbs.go b/sdk/messaging/azeventhubs/internal/cbs.go
index 103f71a9a9..f88f654b0a 100644
--- a/sdk/messaging/azeventhubs/internal/cbs.go
+++ b/sdk/messaging/azeventhubs/internal/cbs.go
@@ -7,9 +7,9 @@ import (
 	"context"
 
 	azlog "github.com/Azure/azure-sdk-for-go/sdk/internal/log"
-	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azeventhubs/internal/amqpwrap"
-	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azeventhubs/internal/auth"
-	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azeventhubs/internal/exported"
+	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azeventhubs/v2/internal/amqpwrap"
+	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azeventhubs/v2/internal/auth"
+	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azeventhubs/v2/internal/exported"
 	"github.com/Azure/go-amqp"
 )
 
diff --git a/sdk/messaging/azeventhubs/internal/constants.go b/sdk/messaging/azeventhubs/internal/constants.go
index e2c7ec40a0..2ca5bb9f07 100644
--- a/sdk/messaging/azeventhubs/internal/constants.go
+++ b/sdk/messaging/azeventhubs/internal/constants.go
@@ -4,4 +4,7 @@
 package internal
 
 // Version is the semantic version number
-const Version = "v1.3.3-beta.1"
+const Version = "v2.0.0-beta.1"
+
+// CapabilityGeoDRReplication is passed as part of our desired capabilities when creating links.
+const CapabilityGeoDRReplication = "com.microsoft:georeplication"
diff --git a/sdk/messaging/azeventhubs/internal/eh/errors_eh.go b/sdk/messaging/azeventhubs/internal/eh/errors_eh.go
new file mode 100644
index 0000000000..d90e7ed1f3
--- /dev/null
+++ b/sdk/messaging/azeventhubs/internal/eh/errors_eh.go
@@ -0,0 +1,28 @@
+// Copyright (c) Microsoft Corporation. All rights reserved.
+// Licensed under the MIT License.
+
+package eh
+
+import (
+	"errors"
+
+	"github.com/Azure/go-amqp"
+)
+
+// ErrCondGeoReplicationOffset occurs when using an old integer offset against a hub that has
+// geo-replication enabled, which requires the new stroffset format.
+const ErrCondGeoReplicationOffset = amqp.ErrCond("com.microsoft:georeplication:invalid-offset")
+
+// IsGeoReplicationOffsetError checks if we've received a "bad offset" error from Event Hubs.
+// This should only happpen if:
+//
+//	a. You're working with an Event Hub namespace that has geo-replication enabled...
+//	b. ...and pass in an older style offset rather than the newer "stroffset" style (contains broker and partition information).
+func IsGeoReplicationOffsetError(err error) bool {
+	if amqpErr := (*amqp.Error)(nil); errors.As(err, &amqpErr) {
+		if amqpErr.Condition == ErrCondGeoReplicationOffset {
+			return true
+		}
+	}
+	return false
+}
diff --git a/sdk/messaging/azeventhubs/internal/eh/stress/stress.go b/sdk/messaging/azeventhubs/internal/eh/stress/stress.go
index 54ea103925..1c5028700c 100644
--- a/sdk/messaging/azeventhubs/internal/eh/stress/stress.go
+++ b/sdk/messaging/azeventhubs/internal/eh/stress/stress.go
@@ -8,7 +8,7 @@ import (
 	"os"
 	"sort"
 
-	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azeventhubs/internal/eh/stress/tests"
+	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azeventhubs/v2/internal/eh/stress/tests"
 )
 
 func main() {
diff --git a/sdk/messaging/azeventhubs/internal/errors.go b/sdk/messaging/azeventhubs/internal/errors.go
index d86c09f8bf..85ea2e06d1 100644
--- a/sdk/messaging/azeventhubs/internal/errors.go
+++ b/sdk/messaging/azeventhubs/internal/errors.go
@@ -10,8 +10,9 @@ import (
 	"net"
 	"net/http"
 
-	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azeventhubs/internal/amqpwrap"
-	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azeventhubs/internal/exported"
+	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azeventhubs/v2/internal/amqpwrap"
+	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azeventhubs/v2/internal/eh"
+	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azeventhubs/v2/internal/exported"
 	"github.com/Azure/go-amqp"
 )
 
@@ -142,6 +143,7 @@ var amqpConditionsToRecoveryKind = map[amqp.ErrCond]RecoveryKind{
 	amqp.ErrCond("com.microsoft:session-cannot-be-locked"): RecoveryKindFatal,
 	amqp.ErrCond("com.microsoft:argument-out-of-range"):    RecoveryKindFatal, // asked for a partition ID that doesn't exist
 	errorConditionLockLost:                                 RecoveryKindFatal,
+	eh.ErrCondGeoReplicationOffset:                         RecoveryKindFatal,
 }
 
 // GetRecoveryKind determines the recovery type for non-session based links.
diff --git a/sdk/messaging/azeventhubs/internal/links.go b/sdk/messaging/azeventhubs/internal/links.go
index 1eceba1eef..fb5f32b597 100644
--- a/sdk/messaging/azeventhubs/internal/links.go
+++ b/sdk/messaging/azeventhubs/internal/links.go
@@ -9,8 +9,8 @@ import (
 	"sync"
 
 	azlog "github.com/Azure/azure-sdk-for-go/sdk/internal/log"
-	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azeventhubs/internal/amqpwrap"
-	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azeventhubs/internal/exported"
+	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azeventhubs/v2/internal/amqpwrap"
+	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azeventhubs/v2/internal/exported"
 )
 
 type AMQPLink interface {
diff --git a/sdk/messaging/azeventhubs/internal/links_recover.go b/sdk/messaging/azeventhubs/internal/links_recover.go
index 534302e276..bd578e231e 100644
--- a/sdk/messaging/azeventhubs/internal/links_recover.go
+++ b/sdk/messaging/azeventhubs/internal/links_recover.go
@@ -8,9 +8,9 @@ import (
 	"errors"
 
 	azlog "github.com/Azure/azure-sdk-for-go/sdk/internal/log"
-	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azeventhubs/internal/amqpwrap"
-	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azeventhubs/internal/exported"
-	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azeventhubs/internal/utils"
+	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azeventhubs/v2/internal/amqpwrap"
+	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azeventhubs/v2/internal/exported"
+	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azeventhubs/v2/internal/utils"
 )
 
 type LinkRetrier[LinkT AMQPLink] struct {
diff --git a/sdk/messaging/azeventhubs/internal/mock/mock_amqp.go b/sdk/messaging/azeventhubs/internal/mock/mock_amqp.go
index 843545200e..2468276abd 100644
--- a/sdk/messaging/azeventhubs/internal/mock/mock_amqp.go
+++ b/sdk/messaging/azeventhubs/internal/mock/mock_amqp.go
@@ -12,7 +12,7 @@ import (
 	context "context"
 	reflect "reflect"
 
-	amqpwrap "github.com/Azure/azure-sdk-for-go/sdk/messaging/azeventhubs/internal/amqpwrap"
+	amqpwrap "github.com/Azure/azure-sdk-for-go/sdk/messaging/azeventhubs/v2/internal/amqpwrap"
 	amqp "github.com/Azure/go-amqp"
 	gomock "github.com/golang/mock/gomock"
 )
diff --git a/sdk/messaging/azeventhubs/internal/mock/mock_namespace.go b/sdk/messaging/azeventhubs/internal/mock/mock_namespace.go
index 5db363568f..e29dde7fab 100644
--- a/sdk/messaging/azeventhubs/internal/mock/mock_namespace.go
+++ b/sdk/messaging/azeventhubs/internal/mock/mock_namespace.go
@@ -12,7 +12,7 @@ import (
 	context "context"
 	reflect "reflect"
 
-	amqpwrap "github.com/Azure/azure-sdk-for-go/sdk/messaging/azeventhubs/internal/amqpwrap"
+	amqpwrap "github.com/Azure/azure-sdk-for-go/sdk/messaging/azeventhubs/v2/internal/amqpwrap"
 	gomock "github.com/golang/mock/gomock"
 )
 
diff --git a/sdk/messaging/azeventhubs/internal/mock/mock_token.go b/sdk/messaging/azeventhubs/internal/mock/mock_token.go
index d2de1871c3..f4879d12e1 100644
--- a/sdk/messaging/azeventhubs/internal/mock/mock_token.go
+++ b/sdk/messaging/azeventhubs/internal/mock/mock_token.go
@@ -9,7 +9,7 @@ package mock
 import (
 	reflect "reflect"
 
-	auth "github.com/Azure/azure-sdk-for-go/sdk/messaging/azeventhubs/internal/auth"
+	auth "github.com/Azure/azure-sdk-for-go/sdk/messaging/azeventhubs/v2/internal/auth"
 	gomock "github.com/golang/mock/gomock"
 )
 
diff --git a/sdk/messaging/azeventhubs/internal/namespace.go b/sdk/messaging/azeventhubs/internal/namespace.go
index e60a45fc30..f9efba5845 100644
--- a/sdk/messaging/azeventhubs/internal/namespace.go
+++ b/sdk/messaging/azeventhubs/internal/namespace.go
@@ -15,11 +15,11 @@ import (
 	"github.com/Azure/azure-sdk-for-go/sdk/azcore"
 	"github.com/Azure/azure-sdk-for-go/sdk/internal/log"
 	"github.com/Azure/azure-sdk-for-go/sdk/internal/telemetry"
-	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azeventhubs/internal/amqpwrap"
-	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azeventhubs/internal/auth"
-	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azeventhubs/internal/exported"
-	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azeventhubs/internal/sbauth"
-	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azeventhubs/internal/utils"
+	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azeventhubs/v2/internal/amqpwrap"
+	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azeventhubs/v2/internal/auth"
+	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azeventhubs/v2/internal/exported"
+	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azeventhubs/v2/internal/sbauth"
+	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azeventhubs/v2/internal/utils"
 	"github.com/Azure/go-amqp"
 )
 
diff --git a/sdk/messaging/azeventhubs/internal/namespace_eh.go b/sdk/messaging/azeventhubs/internal/namespace_eh.go
index 3d827c4067..2790932e3c 100644
--- a/sdk/messaging/azeventhubs/internal/namespace_eh.go
+++ b/sdk/messaging/azeventhubs/internal/namespace_eh.go
@@ -5,9 +5,9 @@ package internal
 import (
 	"context"
 
-	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azeventhubs/internal/amqpwrap"
-	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azeventhubs/internal/auth"
-	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azeventhubs/internal/exported"
+	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azeventhubs/v2/internal/amqpwrap"
+	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azeventhubs/v2/internal/auth"
+	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azeventhubs/v2/internal/exported"
 )
 
 func (l *rpcLink) LinkName() string {
@@ -25,6 +25,9 @@ func (ns *Namespace) NewRPCLink(ctx context.Context, managementPath string) (amq
 		Client:   client,
 		Address:  managementPath,
 		LogEvent: exported.EventProducer,
+		DesiredCapabilities: []string{
+			CapabilityGeoDRReplication,
+		},
 	})
 
 	if err != nil {
diff --git a/sdk/messaging/azeventhubs/internal/rpc.go b/sdk/messaging/azeventhubs/internal/rpc.go
index c731ce0d4e..34f801e648 100644
--- a/sdk/messaging/azeventhubs/internal/rpc.go
+++ b/sdk/messaging/azeventhubs/internal/rpc.go
@@ -13,7 +13,7 @@ import (
 
 	azlog "github.com/Azure/azure-sdk-for-go/sdk/internal/log"
 	"github.com/Azure/azure-sdk-for-go/sdk/internal/uuid"
-	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azeventhubs/internal/amqpwrap"
+	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azeventhubs/v2/internal/amqpwrap"
 	"github.com/Azure/go-amqp"
 )
 
@@ -79,9 +79,10 @@ func (e RPCError) RPCCode() int {
 }
 
 type RPCLinkArgs struct {
-	Client   amqpwrap.AMQPClient
-	Address  string
-	LogEvent azlog.Event
+	Client              amqpwrap.AMQPClient
+	Address             string
+	LogEvent            azlog.Event
+	DesiredCapabilities []string
 }
 
 // NewRPCLink will build a new request response link
@@ -115,7 +116,9 @@ func NewRPCLink(ctx context.Context, args RPCLinkArgs) (amqpwrap.RPCLink, error)
 		ctx,
 		args.Address,
 		"",
-		nil,
+		&amqp.SenderOptions{
+			DesiredCapabilities: args.DesiredCapabilities,
+		},
 	)
 	if err != nil {
 		_ = session.Close(ctx)
@@ -129,6 +132,8 @@ func NewRPCLink(ctx context.Context, args RPCLinkArgs) (amqpwrap.RPCLink, error)
 		// set our receiver link into the "receive and delete" mode - messages arrive pre-settled.
 		SettlementMode:            amqp.ReceiverSettleModeFirst.Ptr(),
 		RequestedSenderSettleMode: amqp.SenderSettleModeSettled.Ptr(),
+
+		DesiredCapabilities: args.DesiredCapabilities,
 	}
 
 	if link.sessionID != nil {
diff --git a/sdk/messaging/azeventhubs/internal/sas/sas.go b/sdk/messaging/azeventhubs/internal/sas/sas.go
index 0b5854ea27..7b14f3c394 100644
--- a/sdk/messaging/azeventhubs/internal/sas/sas.go
+++ b/sdk/messaging/azeventhubs/internal/sas/sas.go
@@ -16,8 +16,8 @@ import (
 	"strings"
 	"time"
 
-	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azeventhubs/internal/auth"
-	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azeventhubs/internal/exported"
+	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azeventhubs/v2/internal/auth"
+	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azeventhubs/v2/internal/exported"
 )
 
 type (
diff --git a/sdk/messaging/azeventhubs/internal/sbauth/token_provider.go b/sdk/messaging/azeventhubs/internal/sbauth/token_provider.go
index f44dc22aad..1a19617ca4 100644
--- a/sdk/messaging/azeventhubs/internal/sbauth/token_provider.go
+++ b/sdk/messaging/azeventhubs/internal/sbauth/token_provider.go
@@ -10,9 +10,9 @@ import (
 
 	"github.com/Azure/azure-sdk-for-go/sdk/azcore"
 	"github.com/Azure/azure-sdk-for-go/sdk/azcore/policy"
-	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azeventhubs/internal/auth"
-	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azeventhubs/internal/exported"
-	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azeventhubs/internal/sas"
+	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azeventhubs/v2/internal/auth"
+	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azeventhubs/v2/internal/exported"
+	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azeventhubs/v2/internal/sas"
 )
 
 // TokenProvider handles access tokens and expiration calculation for SAS
diff --git a/sdk/messaging/azeventhubs/internal/utils/retrier.go b/sdk/messaging/azeventhubs/internal/utils/retrier.go
index 99d74e5eaf..2eae064150 100644
--- a/sdk/messaging/azeventhubs/internal/utils/retrier.go
+++ b/sdk/messaging/azeventhubs/internal/utils/retrier.go
@@ -11,7 +11,7 @@ import (
 	"time"
 
 	"github.com/Azure/azure-sdk-for-go/sdk/internal/log"
-	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azeventhubs/internal/exported"
+	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azeventhubs/v2/internal/exported"
 )
 
 type RetryFnArgs struct {
diff --git a/sdk/messaging/azeventhubs/log.go b/sdk/messaging/azeventhubs/log.go
index 9e48700716..59eb02a029 100644
--- a/sdk/messaging/azeventhubs/log.go
+++ b/sdk/messaging/azeventhubs/log.go
@@ -5,7 +5,7 @@ package azeventhubs
 
 import (
 	"github.com/Azure/azure-sdk-for-go/sdk/azcore/log"
-	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azeventhubs/internal/exported"
+	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azeventhubs/v2/internal/exported"
 )
 
 const (
diff --git a/sdk/messaging/azeventhubs/mgmt.go b/sdk/messaging/azeventhubs/mgmt.go
index 3ac3dba46b..5711acaa50 100644
--- a/sdk/messaging/azeventhubs/mgmt.go
+++ b/sdk/messaging/azeventhubs/mgmt.go
@@ -7,21 +7,28 @@ import (
 	"context"
 	"errors"
 	"fmt"
-	"strconv"
 	"time"
 
 	"github.com/Azure/azure-sdk-for-go/sdk/azcore/log"
-	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azeventhubs/internal"
-	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azeventhubs/internal/amqpwrap"
-	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azeventhubs/internal/eh"
+	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azeventhubs/v2/internal"
+	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azeventhubs/v2/internal/amqpwrap"
+	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azeventhubs/v2/internal/eh"
 	"github.com/Azure/go-amqp"
 )
 
-// EventHubProperties represents properties of the Event Hub, like the number of partitions.
+// EventHubProperties represents properties of an event hub, like the number of partitions.
 type EventHubProperties struct {
-	CreatedOn    time.Time
-	Name         string
+	// CreatedOn is the time when the event hub was created.
+	CreatedOn time.Time
+
+	// Name of the event hub
+	Name string
+
+	// PartitionIDs for the event hub
 	PartitionIDs []string
+
+	// GeoReplicationEnabled is true if the event hub has geo-replication enabled.
+	GeoReplicationEnabled bool
 }
 
 // GetEventHubPropertiesOptions contains optional parameters for the GetEventHubProperties function
@@ -64,7 +71,7 @@ func getEventHubPropertiesInternal(ctx context.Context, ns internal.NamespaceFor
 		},
 	}
 
-	resp, err := rpcLink.RPC(context.Background(), amqpMsg)
+	resp, err := rpcLink.RPC(ctx, amqpMsg)
 
 	if err != nil {
 		return EventHubProperties{}, err
@@ -88,7 +95,7 @@ type PartitionProperties struct {
 	IsEmpty bool
 
 	// LastEnqueuedOffset is the offset of latest enqueued event.
-	LastEnqueuedOffset int64
+	LastEnqueuedOffset string
 
 	// LastEnqueuedOn is the date of latest enqueued event.
 	LastEnqueuedOn time.Time
@@ -164,25 +171,32 @@ func newEventHubProperties(amqpValue any) (EventHubProperties, error) {
 	partitionIDs, ok := m["partition_ids"].([]string)
 
 	if !ok {
-		return EventHubProperties{}, fmt.Errorf("invalid message format")
+		return EventHubProperties{}, fmt.Errorf("invalid value for partition_ids")
 	}
 
 	name, ok := m["name"].(string)
 
 	if !ok {
-		return EventHubProperties{}, fmt.Errorf("invalid message format")
+		return EventHubProperties{}, fmt.Errorf("invalid value for name")
 	}
 
 	createdOn, ok := m["created_at"].(time.Time)
 
 	if !ok {
-		return EventHubProperties{}, fmt.Errorf("invalid message format")
+		return EventHubProperties{}, fmt.Errorf("invalid value for created_at")
+	}
+
+	geoFactor, ok := eh.ConvertToInt64(m["georeplication_factor"])
+
+	if !ok {
+		return EventHubProperties{}, fmt.Errorf("invalid value for georeplication_factor")
 	}
 
 	return EventHubProperties{
-		Name:         name,
-		CreatedOn:    createdOn,
-		PartitionIDs: partitionIDs,
+		Name:                  name,
+		CreatedOn:             createdOn,
+		PartitionIDs:          partitionIDs,
+		GeoReplicationEnabled: geoFactor > 1,
 	}, nil
 }
 
@@ -223,12 +237,6 @@ func newPartitionProperties(amqpValue any) (PartitionProperties, error) {
 		return PartitionProperties{}, errors.New("invalid message format")
 	}
 
-	lastEnqueuedOffset, err := strconv.ParseInt(lastEnqueuedOffsetStr, 10, 64)
-
-	if err != nil {
-		return PartitionProperties{}, fmt.Errorf("invalid message format: %w", err)
-	}
-
 	lastEnqueuedTime, ok := m["last_enqueued_time_utc"].(time.Time)
 
 	if !ok {
@@ -244,7 +252,7 @@ func newPartitionProperties(amqpValue any) (PartitionProperties, error) {
 	return PartitionProperties{
 		BeginningSequenceNumber:    beginningSequenceNumber,
 		LastEnqueuedSequenceNumber: lastEnqueuedSequenceNumber,
-		LastEnqueuedOffset:         lastEnqueuedOffset,
+		LastEnqueuedOffset:         lastEnqueuedOffsetStr,
 		LastEnqueuedOn:             lastEnqueuedTime,
 		IsEmpty:                    isEmpty,
 		PartitionID:                partition,
diff --git a/sdk/messaging/azeventhubs/partition_client.go b/sdk/messaging/azeventhubs/partition_client.go
index 8eb01be00d..5697d5de5f 100644
--- a/sdk/messaging/azeventhubs/partition_client.go
+++ b/sdk/messaging/azeventhubs/partition_client.go
@@ -11,8 +11,8 @@ import (
 
 	"github.com/Azure/azure-sdk-for-go/sdk/azcore/to"
 	"github.com/Azure/azure-sdk-for-go/sdk/internal/log"
-	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azeventhubs/internal"
-	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azeventhubs/internal/amqpwrap"
+	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azeventhubs/v2/internal"
+	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azeventhubs/v2/internal/amqpwrap"
 
 	"github.com/Azure/go-amqp"
 )
@@ -29,14 +29,13 @@ const defaultMaxCreditSize = uint32(5000)
 
 // StartPosition indicates the position to start receiving events within a partition.
 // The default position is Latest.
-//
 // You can set this in the options for [ConsumerClient.NewPartitionClient].
 type StartPosition struct {
 	// Offset will start the consumer after the specified offset. Can be exclusive
 	// or inclusive, based on the Inclusive property.
 	// NOTE: offsets are not stable values, and might refer to different events over time
 	// as the Event Hub events reach their age limit and are discarded.
-	Offset *int64
+	Offset *string
 
 	// SequenceNumber will start the consumer after the specified sequence number. Can be exclusive
 	// or inclusive, based on the Inclusive property.
@@ -221,6 +220,9 @@ func (pc *PartitionClient) newEventHubConsumerLink(ctx context.Context, session
 		},
 		Properties:    props,
 		TargetAddress: pc.instanceID,
+		DesiredCapabilities: []string{
+			internal.CapabilityGeoDRReplication,
+		},
 	}
 
 	if pc.prefetch > 0 {
@@ -340,7 +342,7 @@ func getStartExpression(startPosition StartPosition) (string, error) {
 			return "", errMultipleFieldsSet
 		}
 
-		offsetExpr = fmt.Sprintf("amqp.annotation.x-opt-offset %s '%d'", gt, *startPosition.Offset)
+		offsetExpr = fmt.Sprintf("amqp.annotation.x-opt-offset %s '%s'", gt, *startPosition.Offset)
 	}
 
 	if startPosition.Latest != nil && *startPosition.Latest {
diff --git a/sdk/messaging/azeventhubs/processor.go b/sdk/messaging/azeventhubs/processor.go
index dbc3278444..44f7791d43 100644
--- a/sdk/messaging/azeventhubs/processor.go
+++ b/sdk/messaging/azeventhubs/processor.go
@@ -13,7 +13,9 @@ import (
 	"time"
 
 	"github.com/Azure/azure-sdk-for-go/sdk/azcore/to"
+	"github.com/Azure/azure-sdk-for-go/sdk/internal/log"
 	azlog "github.com/Azure/azure-sdk-for-go/sdk/internal/log"
+	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azeventhubs/v2/internal/eh"
 )
 
 // processorOwnerLevel is the owner level we assign to every ProcessorPartitionClient
@@ -53,7 +55,9 @@ type ProcessorOptions struct {
 
 	// StartPositions are the default start positions (configurable per partition, or with an overall
 	// default value) if a checkpoint is not found in the CheckpointStore.
-	// The default position is Latest.
+	//
+	// - If the Event Hubs namespace has geo-replication enabled, the default is Earliest.
+	// - If the Event Hubs namespace does NOT have geo-replication enabled, the default position is Latest
 	StartPositions StartPositions
 
 	// Prefetch represents the size of the internal prefetch buffer for each ProcessorPartitionClient
@@ -361,7 +365,7 @@ func (p *Processor) dispatch(ctx context.Context, eventHubProperties EventHubPro
 		go func(o Ownership) {
 			defer wg.Done()
 
-			err := p.addPartitionClient(ctx, o, getCheckpoints, consumers)
+			err := p.addPartitionClient(ctx, o, getCheckpoints, p.openPartitionClientImpl, consumers)
 
 			if err != nil {
 				azlog.Writef(EventConsumer, "failed to create partition client for partition '%s': %s", o.PartitionID, err.Error())
@@ -374,8 +378,34 @@ func (p *Processor) dispatch(ctx context.Context, eventHubProperties EventHubPro
 	return nil
 }
 
+// openPartitionClient creates a PartitionClient and initializes it, causing it to open AMQP links.
+// Implemented by [Processor.openPartitionClientImpl]
+type openPartitionClient func(ctx context.Context, partitionID string, startPosition StartPosition) (partitionClient *PartitionClient, err error)
+
+// openPartitionClientImpl creates a PartitionClient and initializes it, causing it to open AMQP links.
+func (p *Processor) openPartitionClientImpl(ctx context.Context, partitionID string, startPosition StartPosition) (partitionClient *PartitionClient, err error) {
+	partitionClient, err = p.consumerClient.NewPartitionClient(partitionID, &PartitionClientOptions{
+		StartPosition: startPosition,
+		OwnerLevel:    processorOwnerLevel,
+		Prefetch:      p.prefetch,
+	})
+
+	if err != nil {
+		return nil, err
+	}
+
+	// make sure we create the link _now_ - if we're stealing we want to stake a claim _now_, rather than
+	// later when the user actually calls ReceiveEvents(), since the acquisition of the link is lazy.
+	if err := partitionClient.init(ctx); err != nil {
+		_ = partitionClient.Close(ctx) // ignore close error here, we're just cleaning up after a failed init()
+		return nil, err
+	}
+
+	return partitionClient, nil
+}
+
 // addPartitionClient creates a ProcessorPartitionClient
-func (p *Processor) addPartitionClient(ctx context.Context, ownership Ownership, getCheckpoints getCheckpoints, consumers *sync.Map) error {
+func (p *Processor) addPartitionClient(ctx context.Context, ownership Ownership, getCheckpoints getCheckpoints, openPartitionClient openPartitionClient, consumers *sync.Map) error {
 	processorPartClient := &ProcessorPartitionClient{
 		consumerClientDetails: p.consumerClientDetails,
 		checkpointStore:       p.checkpointStore,
@@ -390,28 +420,25 @@ func (p *Processor) addPartitionClient(ctx context.Context, ownership Ownership,
 		return nil
 	}
 
-	startPosition, err := p.getStartPosition(getCheckpoints, ownership)
+	preferredStartPosition, err := p.getStartPosition(getCheckpoints, ownership)
 
 	if err != nil {
 		return err
 	}
 
-	partClient, err := p.consumerClient.NewPartitionClient(ownership.PartitionID, &PartitionClientOptions{
-		StartPosition: startPosition,
-		OwnerLevel:    processorOwnerLevel,
-		Prefetch:      p.prefetch,
-	})
+	partClient, err := openPartitionClient(ctx, ownership.PartitionID, preferredStartPosition)
 
-	if err != nil {
-		consumers.Delete(ownership.PartitionID)
-		return err
+	if eh.IsGeoReplicationOffsetError(err) {
+		log.Writef(EventConsumer, "Event Hub is in geo-replication mode and we only have an integer offset, will fallback to starting at earliest+inclusive: %s", err.Error())
+
+		partClient, err = openPartitionClient(ctx, ownership.PartitionID, StartPosition{
+			Earliest:  to.Ptr(true),
+			Inclusive: true,
+		})
 	}
 
-	// make sure we create the link _now_ - if we're stealing we want to stake a claim _now_, rather than
-	// later when the user actually calls ReceiveEvents(), since the acquisition of the link is lazy.
-	if err := partClient.init(ctx); err != nil {
+	if err != nil {
 		consumers.Delete(ownership.PartitionID)
-		_ = partClient.Close(ctx)
 		return err
 	}
 
@@ -426,6 +453,8 @@ func (p *Processor) addPartitionClient(ctx context.Context, ownership Ownership,
 	}
 }
 
+// getStartPosition gets the start position, preferring a stored Checkpoint, then falling back to per-partition defaults
+// and then the global default position.
 func (p *Processor) getStartPosition(getCheckpoints getCheckpoints, ownership Ownership) (StartPosition, error) {
 	checkpoints, err := getCheckpoints()
 
diff --git a/sdk/messaging/azeventhubs/producer_client.go b/sdk/messaging/azeventhubs/producer_client.go
index d92991f0ba..d8582bfbde 100644
--- a/sdk/messaging/azeventhubs/producer_client.go
+++ b/sdk/messaging/azeventhubs/producer_client.go
@@ -13,9 +13,9 @@ import (
 	"github.com/Azure/azure-sdk-for-go/sdk/azcore"
 	"github.com/Azure/azure-sdk-for-go/sdk/azcore/to"
 	azlog "github.com/Azure/azure-sdk-for-go/sdk/internal/log"
-	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azeventhubs/internal"
-	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azeventhubs/internal/amqpwrap"
-	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azeventhubs/internal/exported"
+	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azeventhubs/v2/internal"
+	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azeventhubs/v2/internal/amqpwrap"
+	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azeventhubs/v2/internal/exported"
 	"github.com/Azure/go-amqp"
 )
 
@@ -213,6 +213,9 @@ func (pc *ProducerClient) newEventHubProducerLink(ctx context.Context, session a
 	sender, err := session.NewSender(ctx, entityPath, partitionID, &amqp.SenderOptions{
 		SettlementMode:              to.Ptr(amqp.SenderSettleModeMixed),
 		RequestedReceiverSettleMode: to.Ptr(amqp.ReceiverSettleModeFirst),
+		DesiredCapabilities: []string{
+			internal.CapabilityGeoDRReplication,
+		},
 	})
 
 	if err != nil {
diff --git a/sdk/monitor/azquery/CHANGELOG.md b/sdk/monitor/azquery/CHANGELOG.md
index 1a6c1a4759..45b3648168 100644
--- a/sdk/monitor/azquery/CHANGELOG.md
+++ b/sdk/monitor/azquery/CHANGELOG.md
@@ -3,6 +3,7 @@
 ## 1.2.0-beta.2 (Unreleased)
 
 ### Features Added
+* Added API Version support for `MetricsClient`. Users can now change the default API Version by setting `MetricsClientOptions.APIVersion`.
 
 ### Breaking Changes
 
diff --git a/sdk/monitor/azquery/custom_client.go b/sdk/monitor/azquery/custom_client.go
index 0b30171f6c..189dc2ab9c 100644
--- a/sdk/monitor/azquery/custom_client.go
+++ b/sdk/monitor/azquery/custom_client.go
@@ -72,7 +72,14 @@ func NewMetricsClient(credential azcore.TokenCredential, options *MetricsClientO
 	}
 
 	authPolicy := runtime.NewBearerTokenPolicy(credential, []string{c.Audience + "/.default"}, nil)
-	azcoreClient, err := azcore.NewClient(moduleName, version, runtime.PipelineOptions{PerRetry: []policy.Policy{authPolicy}}, &options.ClientOptions)
+	pipelineOptions := runtime.PipelineOptions{
+		APIVersion: runtime.APIVersionOptions{
+			Location: runtime.APIVersionLocationQueryParam,
+			Name:     "api-version",
+		},
+		PerRetry: []policy.Policy{authPolicy},
+	}
+	azcoreClient, err := azcore.NewClient(moduleName, version, pipelineOptions, &options.ClientOptions)
 	if err != nil {
 		return nil, err
 	}


Test patch:
diff --git a/sdk/messaging/azeventhubs/checkpoints/blob_store_test.go b/sdk/messaging/azeventhubs/checkpoints/blob_store_test.go
index 4c1b1bf9f3..38f9fc4e4d 100644
--- a/sdk/messaging/azeventhubs/checkpoints/blob_store_test.go
+++ b/sdk/messaging/azeventhubs/checkpoints/blob_store_test.go
@@ -13,9 +13,9 @@ import (
 	"github.com/Azure/azure-sdk-for-go/sdk/azcore"
 	"github.com/Azure/azure-sdk-for-go/sdk/azcore/to"
 	"github.com/Azure/azure-sdk-for-go/sdk/internal/test/credential"
-	"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/messaging/azeventhubs/internal/test"
+	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azeventhubs/v2"
+	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azeventhubs/v2/checkpoints"
+	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azeventhubs/v2/internal/test"
 	"github.com/Azure/azure-sdk-for-go/sdk/storage/azblob/container"
 	"github.com/joho/godotenv"
 	"github.com/stretchr/testify/require"
@@ -33,7 +33,7 @@ func TestBlobStore_Checkpoints(t *testing.T) {
 		EventHubName:            "event-hub-name",
 		FullyQualifiedNamespace: "ns.servicebus.windows.net",
 		PartitionID:             "partition-id",
-		Offset:                  to.Ptr[int64](101),
+		Offset:                  to.Ptr("101"),
 		SequenceNumber:          to.Ptr[int64](202),
 	}, nil)
 	require.NoError(t, err)
@@ -46,7 +46,7 @@ func TestBlobStore_Checkpoints(t *testing.T) {
 		EventHubName:            "event-hub-name",
 		FullyQualifiedNamespace: "ns.servicebus.windows.net",
 		PartitionID:             "partition-id",
-		Offset:                  to.Ptr[int64](101),
+		Offset:                  to.Ptr("101"),
 		SequenceNumber:          to.Ptr[int64](202),
 	}, checkpoints[0])
 
@@ -57,7 +57,7 @@ func TestBlobStore_Checkpoints(t *testing.T) {
 		EventHubName:            "event-hub-name",
 		FullyQualifiedNamespace: "ns.servicebus.windows.net",
 		PartitionID:             "partition-id",
-		Offset:                  to.Ptr[int64](102),
+		Offset:                  to.Ptr("102"),
 		SequenceNumber:          to.Ptr[int64](203),
 	}, nil)
 	require.NoError(t, err)
@@ -70,7 +70,7 @@ func TestBlobStore_Checkpoints(t *testing.T) {
 		EventHubName:            "event-hub-name",
 		FullyQualifiedNamespace: "ns.servicebus.windows.net",
 		PartitionID:             "partition-id",
-		Offset:                  to.Ptr[int64](102),
+		Offset:                  to.Ptr("102"),
 		SequenceNumber:          to.Ptr[int64](203),
 	}, checkpoints[0])
 }
diff --git a/sdk/messaging/azeventhubs/checkpoints/blob_store_unit_test.go b/sdk/messaging/azeventhubs/checkpoints/blob_store_unit_test.go
index 606813f4f8..79bc861112 100644
--- a/sdk/messaging/azeventhubs/checkpoints/blob_store_unit_test.go
+++ b/sdk/messaging/azeventhubs/checkpoints/blob_store_unit_test.go
@@ -8,7 +8,7 @@ import (
 
 	"github.com/Azure/azure-sdk-for-go/sdk/azcore"
 	"github.com/Azure/azure-sdk-for-go/sdk/azcore/to"
-	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azeventhubs"
+	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azeventhubs/v2"
 	"github.com/Azure/azure-sdk-for-go/sdk/storage/azblob/container"
 	"github.com/stretchr/testify/require"
 )
diff --git a/sdk/messaging/azeventhubs/consumer_client_internal_test.go b/sdk/messaging/azeventhubs/consumer_client_internal_test.go
index adcc5c6123..c263582d3b 100644
--- a/sdk/messaging/azeventhubs/consumer_client_internal_test.go
+++ b/sdk/messaging/azeventhubs/consumer_client_internal_test.go
@@ -13,9 +13,9 @@ import (
 	"time"
 
 	"github.com/Azure/azure-sdk-for-go/sdk/internal/test/credential"
-	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azeventhubs/internal"
-	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azeventhubs/internal/amqpwrap"
-	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azeventhubs/internal/test"
+	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azeventhubs/v2/internal"
+	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azeventhubs/v2/internal/amqpwrap"
+	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azeventhubs/v2/internal/test"
 	"github.com/stretchr/testify/require"
 )
 
@@ -44,7 +44,7 @@ func TestConsumerClient_Recovery(t *testing.T) {
 
 	type sendResult struct {
 		PartitionID  string
-		OffsetBefore int64
+		OffsetBefore string
 	}
 
 	sendResults := make([]sendResult, len(ehProps.PartitionIDs))
@@ -126,7 +126,7 @@ func TestConsumerClient_Recovery(t *testing.T) {
 			require.NoError(t, err)
 			require.EqualValues(t, 1, len(events))
 
-			t.Logf("[%s] Received seq:%d, offset:%d", sr.PartitionID, events[0].SequenceNumber, events[0].Offset)
+			t.Logf("[%s] Received seq:%d, offset:%s", sr.PartitionID, events[0].SequenceNumber, events[0].Offset)
 
 			require.Equal(t, fmt.Sprintf("event 1 for partition %s", sr.PartitionID), string(events[0].Body))
 		}(i, sr)
@@ -193,7 +193,7 @@ func TestConsumerClient_RecoveryLink(t *testing.T) {
 
 	type sendResult struct {
 		PartitionID  string
-		OffsetBefore int64
+		OffsetBefore string
 	}
 
 	sendResults := make([]sendResult, len(ehProps.PartitionIDs))
diff --git a/sdk/messaging/azeventhubs/consumer_client_test.go b/sdk/messaging/azeventhubs/consumer_client_test.go
index 8222d70e81..95cd9e2d5c 100644
--- a/sdk/messaging/azeventhubs/consumer_client_test.go
+++ b/sdk/messaging/azeventhubs/consumer_client_test.go
@@ -18,8 +18,8 @@ import (
 	"github.com/Azure/azure-sdk-for-go/sdk/azcore/to"
 	"github.com/Azure/azure-sdk-for-go/sdk/internal/test/credential"
 	"github.com/Azure/azure-sdk-for-go/sdk/internal/uuid"
-	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azeventhubs"
-	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azeventhubs/internal/test"
+	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azeventhubs/v2"
+	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azeventhubs/v2/internal/test"
 	"github.com/Azure/azure-sdk-for-go/sdk/resourcemanager/eventhub/armeventhub"
 	"github.com/coder/websocket"
 	"github.com/stretchr/testify/require"
diff --git a/sdk/messaging/azeventhubs/consumer_client_unit_test.go b/sdk/messaging/azeventhubs/consumer_client_unit_test.go
index 1d91ff1642..57c89aa5fe 100644
--- a/sdk/messaging/azeventhubs/consumer_client_unit_test.go
+++ b/sdk/messaging/azeventhubs/consumer_client_unit_test.go
@@ -65,11 +65,11 @@ func TestUnit_getOffsetExpression(t *testing.T) {
 		require.NoError(t, err)
 		require.Equal(t, "amqp.annotation.x-opt-offset >= '@latest'", expr)
 
-		expr, err = getStartExpression(StartPosition{Offset: to.Ptr(int64(101))})
+		expr, err = getStartExpression(StartPosition{Offset: to.Ptr("101")})
 		require.NoError(t, err)
 		require.Equal(t, "amqp.annotation.x-opt-offset > '101'", expr)
 
-		expr, err = getStartExpression(StartPosition{Offset: to.Ptr(int64(101)), Inclusive: true})
+		expr, err = getStartExpression(StartPosition{Offset: to.Ptr("101"), Inclusive: true})
 		require.NoError(t, err)
 		require.Equal(t, "amqp.annotation.x-opt-offset >= '101'", expr)
 
@@ -99,13 +99,13 @@ func TestUnit_getOffsetExpression(t *testing.T) {
 
 		expr, err := getStartExpression(StartPosition{
 			EnqueuedTime: &enqueueTime,
-			Offset:       to.Ptr[int64](101),
+			Offset:       to.Ptr("101"),
 		})
 		require.EqualError(t, err, "only a single start point can be set: Earliest, EnqueuedTime, Latest, Offset, or SequenceNumber")
 		require.Empty(t, expr)
 
 		expr, err = getStartExpression(StartPosition{
-			Offset: to.Ptr[int64](202),
+			Offset: to.Ptr("202"),
 			Latest: to.Ptr(true),
 		})
 		require.EqualError(t, err, "only a single start point can be set: Earliest, EnqueuedTime, Latest, Offset, or SequenceNumber")
diff --git a/sdk/messaging/azeventhubs/event_data_batch_unit_test.go b/sdk/messaging/azeventhubs/event_data_batch_unit_test.go
index 4144f09a36..a26714665f 100644
--- a/sdk/messaging/azeventhubs/event_data_batch_unit_test.go
+++ b/sdk/messaging/azeventhubs/event_data_batch_unit_test.go
@@ -9,8 +9,8 @@ import (
 	"testing"
 
 	"github.com/Azure/azure-sdk-for-go/sdk/azcore/to"
-	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azeventhubs/internal/amqpwrap"
-	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azeventhubs/internal/mock"
+	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azeventhubs/v2/internal/amqpwrap"
+	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azeventhubs/v2/internal/mock"
 	"github.com/Azure/go-amqp"
 	"github.com/golang/mock/gomock"
 	"github.com/stretchr/testify/require"
diff --git a/sdk/messaging/azeventhubs/event_data_test.go b/sdk/messaging/azeventhubs/event_data_test.go
index 0c58c9bde4..dd1876079e 100644
--- a/sdk/messaging/azeventhubs/event_data_test.go
+++ b/sdk/messaging/azeventhubs/event_data_test.go
@@ -99,7 +99,7 @@ func TestEventData_newReceivedEventData(t *testing.T) {
 		SystemProperties: map[string]any{
 			"hello": "world",
 		},
-		Offset:       int64(102),
+		Offset:       "102",
 		PartitionKey: to.Ptr("partition key"),
 		RawAMQPMessage: &AMQPAnnotatedMessage{
 			Properties: &AMQPAnnotatedMessageProperties{
diff --git a/sdk/messaging/azeventhubs/example_checkpoint_migration_test.go b/sdk/messaging/azeventhubs/example_checkpoint_migration_test.go
index 937c364a8a..1361f211d2 100644
--- a/sdk/messaging/azeventhubs/example_checkpoint_migration_test.go
+++ b/sdk/messaging/azeventhubs/example_checkpoint_migration_test.go
@@ -8,10 +8,9 @@ import (
 	"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/messaging/azeventhubs/v2"
+	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azeventhubs/v2/checkpoints"
 	"github.com/Azure/azure-sdk-for-go/sdk/storage/azblob"
 	"github.com/Azure/azure-sdk-for-go/sdk/storage/azblob/container"
 )
@@ -28,7 +27,7 @@ type LegacyCheckpoint struct {
 }
 
 // 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`
+// the format used by this package, `github.com/Azure/azure-sdk-for-go/sdk/messaging/azeventhubs/v2/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.
@@ -93,13 +92,7 @@ func Example_migrateCheckpoints() {
 			PartitionID:             oldCheckpoint.PartitionID,
 		}
 
-		offset, err := strconv.ParseInt(oldCheckpoint.Checkpoint.Offset, 10, 64)
-
-		if err != nil {
-			panic(err)
-		}
-
-		newCheckpoint.Offset = &offset
+		newCheckpoint.Offset = &oldCheckpoint.Checkpoint.Offset
 		newCheckpoint.SequenceNumber = &oldCheckpoint.Checkpoint.SequenceNumber
 
 		if err := newCheckpointStore.SetCheckpoint(context.Background(), newCheckpoint, nil); err != nil {
diff --git a/sdk/messaging/azeventhubs/example_consumerclient_test.go b/sdk/messaging/azeventhubs/example_consumerclient_test.go
index 7853a589ca..bcb5a0e0f8 100644
--- a/sdk/messaging/azeventhubs/example_consumerclient_test.go
+++ b/sdk/messaging/azeventhubs/example_consumerclient_test.go
@@ -8,7 +8,7 @@ import (
 	"time"
 
 	"github.com/Azure/azure-sdk-for-go/sdk/azidentity"
-	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azeventhubs"
+	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azeventhubs/v2"
 )
 
 var consumerClient *azeventhubs.ConsumerClient
diff --git a/sdk/messaging/azeventhubs/example_consuming_events_test.go b/sdk/messaging/azeventhubs/example_consuming_events_test.go
index b4e5de3ecf..1de12e083e 100644
--- a/sdk/messaging/azeventhubs/example_consuming_events_test.go
+++ b/sdk/messaging/azeventhubs/example_consuming_events_test.go
@@ -12,7 +12,7 @@ import (
 
 	"github.com/Azure/azure-sdk-for-go/sdk/azcore/to"
 	"github.com/Azure/azure-sdk-for-go/sdk/azidentity"
-	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azeventhubs"
+	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azeventhubs/v2"
 )
 
 // Shows how to start consuming events in partitions in an Event Hub using the [ConsumerClient].
diff --git a/sdk/messaging/azeventhubs/example_consuming_with_checkpoints_test.go b/sdk/messaging/azeventhubs/example_consuming_with_checkpoints_test.go
index e64c1a7c75..f04491f97d 100644
--- a/sdk/messaging/azeventhubs/example_consuming_with_checkpoints_test.go
+++ b/sdk/messaging/azeventhubs/example_consuming_with_checkpoints_test.go
@@ -12,8 +12,8 @@ import (
 	"time"
 
 	"github.com/Azure/azure-sdk-for-go/sdk/azidentity"
-	"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/messaging/azeventhubs/v2"
+	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azeventhubs/v2/checkpoints"
 	"github.com/Azure/azure-sdk-for-go/sdk/storage/azblob"
 )
 
diff --git a/sdk/messaging/azeventhubs/example_enabling_logging_test.go b/sdk/messaging/azeventhubs/example_enabling_logging_test.go
index 4efdc99dd3..5a80e51b74 100644
--- a/sdk/messaging/azeventhubs/example_enabling_logging_test.go
+++ b/sdk/messaging/azeventhubs/example_enabling_logging_test.go
@@ -7,7 +7,7 @@ import (
 	"fmt"
 
 	azlog "github.com/Azure/azure-sdk-for-go/sdk/azcore/log"
-	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azeventhubs"
+	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azeventhubs/v2"
 )
 
 func Example_enableLogging() {
diff --git a/sdk/messaging/azeventhubs/example_producerclient_test.go b/sdk/messaging/azeventhubs/example_producerclient_test.go
index 9fa004f98f..61146ed4a0 100644
--- a/sdk/messaging/azeventhubs/example_producerclient_test.go
+++ b/sdk/messaging/azeventhubs/example_producerclient_test.go
@@ -8,7 +8,7 @@ import (
 	"fmt"
 
 	"github.com/Azure/azure-sdk-for-go/sdk/azidentity"
-	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azeventhubs"
+	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azeventhubs/v2"
 )
 
 var producerClient *azeventhubs.ProducerClient
diff --git a/sdk/messaging/azeventhubs/example_producing_events_test.go b/sdk/messaging/azeventhubs/example_producing_events_test.go
index 18bfae4357..7196e582e4 100644
--- a/sdk/messaging/azeventhubs/example_producing_events_test.go
+++ b/sdk/messaging/azeventhubs/example_producing_events_test.go
@@ -9,7 +9,7 @@ import (
 	"os"
 
 	"github.com/Azure/azure-sdk-for-go/sdk/azidentity"
-	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azeventhubs"
+	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azeventhubs/v2"
 )
 
 // Shows how to send events to an Event Hub partition using the [ProducerClient]
diff --git a/sdk/messaging/azeventhubs/example_websockets_and_proxies_test.go b/sdk/messaging/azeventhubs/example_websockets_and_proxies_test.go
index 059abc9dc7..6fcf9f3d16 100644
--- a/sdk/messaging/azeventhubs/example_websockets_and_proxies_test.go
+++ b/sdk/messaging/azeventhubs/example_websockets_and_proxies_test.go
@@ -11,7 +11,7 @@ import (
 	"os"
 
 	"github.com/Azure/azure-sdk-for-go/sdk/azidentity"
-	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azeventhubs"
+	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azeventhubs/v2"
 	"github.com/coder/websocket"
 )
 
diff --git a/sdk/messaging/azeventhubs/geodr_test.go b/sdk/messaging/azeventhubs/geodr_test.go
new file mode 100644
index 0000000000..97b9ffd7ac
--- /dev/null
+++ b/sdk/messaging/azeventhubs/geodr_test.go
@@ -0,0 +1,225 @@
+// Copyright (c) Microsoft Corporation. All rights reserved.
+// Licensed under the MIT License.
+
+package azeventhubs_test
+
+import (
+	"context"
+	"strconv"
+	"testing"
+
+	"github.com/Azure/azure-sdk-for-go/sdk/azcore/to"
+	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azeventhubs/v2"
+	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azeventhubs/v2/internal/test"
+	"github.com/stretchr/testify/require"
+)
+
+func TestConsumerClient_GeoReplication(t *testing.T) {
+	// this test just needs a single partition to test
+	var partitionID = "0"
+
+	testParams := test.GetConnectionParamsForTest(t)
+
+	if testParams.GeoDRNamespace == "" || testParams.GeoDRHubName == "" || testParams.GeoDRStorageEndpoint == "" {
+		t.Skipf("Skipping GeoDR test, EVENTHUBS_GEODR_NAMESPACE or EVENTHUBS_GEODR_HUBNAME or EVENTHUBS_GEODR_CHECKPOINTSTORE_STORAGE_ENDPOINT was not set")
+	}
+
+	propsBeforeTest := func() azeventhubs.PartitionProperties {
+		producer, err := azeventhubs.NewProducerClient(testParams.GeoDRNamespace, testParams.GeoDRHubName, testParams.Cred, nil)
+		require.NoError(t, err)
+
+		defer test.RequireClose(t, producer)
+
+		props, err := producer.GetEventHubProperties(context.Background(), nil)
+		require.NoError(t, err)
+		require.True(t, props.GeoReplicationEnabled)
+
+		propsBeforeTest, err := producer.GetPartitionProperties(context.Background(), "0", nil)
+		require.NoError(t, err)
+
+		// This is what the partition properties look like, with geo-replication enabled
+		// (note, this example event hub was empty):
+		// {
+		//     BeginningSequenceNumber:-1,
+		// 	   EventHubName:"ehrp2",
+		// 	   IsEmpty:true,
+		// 	   LastEnqueuedOffset:"2:-1:-1",
+		// 	   LastEnqueuedOn:time.Date(1, time.January, 1, 0, 0, 0, 0, time.Local),
+		//     LastEnqueuedSequenceNumber:-1,
+		//     PartitionID:"0"
+		// }
+
+		t.Logf("LastEnqueuedOffset: %#v, LastEnqueuedSequenceNumber: %#v", propsBeforeTest.LastEnqueuedOffset, propsBeforeTest.LastEnqueuedSequenceNumber)
+
+		// we send a couple of events so the processor tests, that can't be started inclusive, will still have something
+		// predictable to retrieve.
+		batch, err := producer.NewEventDataBatch(context.Background(), &azeventhubs.EventDataBatchOptions{
+			PartitionID: &partitionID,
+		})
+		require.NoError(t, err)
+
+		// the Event Hub is re-used, so this is the first in our sent messages, but not necessarily
+		// the actual first message in the partition.
+		err = batch.AddEventData(&azeventhubs.EventData{Body: []byte("1")}, nil)
+		require.NoError(t, err)
+
+		err = batch.AddEventData(&azeventhubs.EventData{Body: []byte("2")}, nil)
+		require.NoError(t, err)
+
+		// even if the event hub is re-used, this is still the last message for sure.
+		err = batch.AddEventData(&azeventhubs.EventData{Body: []byte("3")}, nil)
+		require.NoError(t, err)
+
+		err = producer.SendEventDataBatch(context.Background(), batch, nil)
+		require.NoError(t, err)
+
+		return propsBeforeTest
+	}()
+
+	earliestEvent, ourFirstEvent := func() (*azeventhubs.ReceivedEventData, *azeventhubs.ReceivedEventData) {
+		consumer, err := azeventhubs.NewConsumerClient(testParams.GeoDRNamespace, testParams.GeoDRHubName, azeventhubs.DefaultConsumerGroup, testParams.Cred, nil)
+		require.NoError(t, err)
+
+		pc, err := consumer.NewPartitionClient(partitionID, &azeventhubs.PartitionClientOptions{
+			StartPosition: azeventhubs.StartPosition{
+				Earliest:  to.Ptr(true),
+				Inclusive: true,
+			},
+		})
+		require.NoError(t, err)
+		defer test.RequireClose(t, pc)
+
+		earliestEvents, err := pc.ReceiveEvents(context.Background(), 1, nil)
+		require.NoError(t, err)
+		require.NotEmpty(t, earliestEvents)
+
+		// Now let's start at the point just before we started sending events. We want those offsets as well.
+		pc, err = consumer.NewPartitionClient(partitionID, &azeventhubs.PartitionClientOptions{
+			StartPosition: azeventhubs.StartPosition{
+				Offset:    &propsBeforeTest.LastEnqueuedOffset,
+				Inclusive: false,
+			},
+		})
+		require.NoError(t, err)
+		defer test.RequireClose(t, pc)
+
+		ourEvents, err := pc.ReceiveEvents(context.Background(), 3, nil)
+		require.NoError(t, err)
+		require.NotEmpty(t, ourEvents)
+
+		return earliestEvents[0], ourEvents[0]
+	}()
+
+	runTest := func(t *testing.T, proc *azeventhubs.Processor) *azeventhubs.ReceivedEventData {
+		ctx, cancel := context.WithCancel(context.Background())
+		defer cancel()
+
+		ch := make(chan struct{})
+
+		go func() {
+			defer close(ch)
+			err := proc.Run(ctx)
+			require.NoError(t, err)
+		}()
+
+		var event *azeventhubs.ReceivedEventData
+
+		for {
+			pc := proc.NextPartitionClient(context.Background())
+
+			if pc.PartitionID() == "0" {
+				events, err := pc.ReceiveEvents(context.Background(), 1, nil)
+				require.NoError(t, err)
+				require.NotEmpty(t, events)
+
+				event = events[0]
+
+				cancel()
+				_ = pc.Close(context.Background())
+				break
+			} else {
+				_ = pc.Close(context.Background())
+			}
+		}
+
+		require.NotNil(t, event)
+		<-ch
+		return event
+	}
+
+	t.Run("ProcessorWithLegacyOffset", func(t *testing.T) {
+		setup := func(t *testing.T) *processorTestData {
+			td := setupProcessorTest(t, true)
+
+			err = td.CheckpointStore.SetCheckpoint(context.Background(), azeventhubs.Checkpoint{
+				ConsumerGroup:           azeventhubs.DefaultConsumerGroup,
+				FullyQualifiedNamespace: testParams.GeoDRNamespace,
+				EventHubName:            testParams.GeoDRHubName,
+				PartitionID:             "0",
+				// this is invalid - you can't use old offsets with a new GeoDR-enabled Event Hub once it's
+				// been promoted.
+				Offset:         to.Ptr("0"),
+				SequenceNumber: to.Ptr(int64(1)),
+			}, nil)
+			require.NoError(t, err)
+
+			return td
+		}
+
+		td := setup(t)
+		proc := td.Create(nil)
+		event := runTest(t, proc)
+
+		// Here's what happens here:
+		// 1. Processor loads up checkpoint, which contains a "legacy" offset (ie, just an integer)
+		// 2. It attempts to create a consumer using that offset, which Event Hubs rejects, with a GeoDR related error
+		// 3. We then fallback to opening up the start of the partition instead (ie: earliest)
+		require.Equal(t, earliestEvent, event)
+	})
+
+	t.Run("Processor", func(t *testing.T) {
+		setup := func(t *testing.T) *processorTestData {
+			td := setupProcessorTest(t, true)
+
+			err = td.CheckpointStore.SetCheckpoint(context.Background(), azeventhubs.Checkpoint{
+				ConsumerGroup:           azeventhubs.DefaultConsumerGroup,
+				FullyQualifiedNamespace: testParams.GeoDRNamespace,
+				EventHubName:            testParams.GeoDRHubName,
+				PartitionID:             "0",
+				// Checkpoints always point to the last event received, so we will receive the event just after ourFirstEvent
+				Offset:         &ourFirstEvent.Offset,
+				SequenceNumber: &ourFirstEvent.SequenceNumber,
+			}, nil)
+			require.NoError(t, err)
+
+			return td
+		}
+
+		td := setup(t)
+		proc := td.Create(nil)
+		event := runTest(t, proc)
+
+		require.Equal(t, "2", string(event.Body))
+	})
+
+	t.Run("StartWithOffsetFromGetPartitionProperties", func(t *testing.T) {
+		cc, err := azeventhubs.NewConsumerClient(testParams.GeoDRNamespace, testParams.GeoDRHubName, azeventhubs.DefaultConsumerGroup, testParams.Cred, nil)
+		require.NoError(t, err)
+		defer test.RequireClose(t, cc)
+
+		_, err = strconv.ParseInt(propsBeforeTest.LastEnqueuedOffset, 10, 64)
+		require.Error(t, err, "offsets are no longer just integers")
+
+		pc, err := cc.NewPartitionClient("0", &azeventhubs.PartitionClientOptions{
+			StartPosition: azeventhubs.StartPosition{
+				Offset: &propsBeforeTest.LastEnqueuedOffset,
+			},
+		})
+		require.NoError(t, err)
+		defer test.RequireClose(t, pc)
+
+		events, err := pc.ReceiveEvents(context.Background(), 1, nil)
+		require.NoError(t, err)
+		require.NotEmpty(t, events)
+	})
+}
diff --git a/sdk/messaging/azeventhubs/inmemory_checkpoint_store_test.go b/sdk/messaging/azeventhubs/inmemory_checkpoint_store_test.go
index 55c3650d48..7b3be9cb83 100644
--- a/sdk/messaging/azeventhubs/inmemory_checkpoint_store_test.go
+++ b/sdk/messaging/azeventhubs/inmemory_checkpoint_store_test.go
@@ -5,6 +5,7 @@ package azeventhubs
 import (
 	"context"
 	"sort"
+	"strconv"
 	"strings"
 	"sync"
 	"testing"
@@ -29,7 +30,7 @@ func Test_InMemoryCheckpointStore_Checkpoints(t *testing.T) {
 			EventHubName:            "eh",
 			ConsumerGroup:           "cg",
 			PartitionID:             "100",
-			Offset:                  to.Ptr(i),
+			Offset:                  to.Ptr(strconv.Itoa(int(i))),
 			SequenceNumber:          to.Ptr(i + 1),
 		}, nil)
 		require.NoError(t, err)
@@ -43,7 +44,7 @@ func Test_InMemoryCheckpointStore_Checkpoints(t *testing.T) {
 				EventHubName:            "eh",
 				ConsumerGroup:           "cg",
 				PartitionID:             "100",
-				Offset:                  to.Ptr(i),
+				Offset:                  to.Ptr(strconv.Itoa(int(i))),
 				SequenceNumber:          to.Ptr(i + 1),
 			},
 		}, checkpoints)
diff --git a/sdk/messaging/azeventhubs/internal/amqpwrap/amqpwrap_test.go b/sdk/messaging/azeventhubs/internal/amqpwrap/amqpwrap_test.go
index d7a16b73c3..20d8f3bd59 100644
--- a/sdk/messaging/azeventhubs/internal/amqpwrap/amqpwrap_test.go
+++ b/sdk/messaging/azeventhubs/internal/amqpwrap/amqpwrap_test.go
@@ -10,7 +10,7 @@ import (
 	"errors"
 	"testing"
 
-	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azeventhubs/internal/test"
+	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azeventhubs/v2/internal/test"
 	"github.com/Azure/go-amqp"
 	gomock "github.com/golang/mock/gomock"
 	"github.com/stretchr/testify/require"
diff --git a/sdk/messaging/azeventhubs/internal/cbs_test.go b/sdk/messaging/azeventhubs/internal/cbs_test.go
index a76964a730..d2348cb500 100644
--- a/sdk/messaging/azeventhubs/internal/cbs_test.go
+++ b/sdk/messaging/azeventhubs/internal/cbs_test.go
@@ -8,10 +8,10 @@ import (
 	"fmt"
 	"testing"
 
-	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azeventhubs/internal/amqpwrap"
-	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azeventhubs/internal/auth"
-	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azeventhubs/internal/mock"
-	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azeventhubs/internal/test"
+	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azeventhubs/v2/internal/amqpwrap"
+	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azeventhubs/v2/internal/auth"
+	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azeventhubs/v2/internal/mock"
+	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azeventhubs/v2/internal/test"
 	"github.com/Azure/go-amqp"
 	"github.com/golang/mock/gomock"
 	"github.com/stretchr/testify/require"
diff --git a/sdk/messaging/azeventhubs/internal/eh/stress/tests/balance_tester.go b/sdk/messaging/azeventhubs/internal/eh/stress/tests/balance_tester.go
index 52a9339de0..14426cdc48 100644
--- a/sdk/messaging/azeventhubs/internal/eh/stress/tests/balance_tester.go
+++ b/sdk/messaging/azeventhubs/internal/eh/stress/tests/balance_tester.go
@@ -16,8 +16,8 @@ import (
 	"time"
 
 	"github.com/Azure/azure-sdk-for-go/sdk/internal/log"
-	"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/messaging/azeventhubs/v2"
+	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azeventhubs/v2/checkpoints"
 	"github.com/Azure/azure-sdk-for-go/sdk/storage/azblob"
 	"github.com/Azure/azure-sdk-for-go/sdk/storage/azblob/bloberror"
 )
diff --git a/sdk/messaging/azeventhubs/internal/eh/stress/tests/batch_stress_tester.go b/sdk/messaging/azeventhubs/internal/eh/stress/tests/batch_stress_tester.go
index bd95b4566f..110ca54286 100644
--- a/sdk/messaging/azeventhubs/internal/eh/stress/tests/batch_stress_tester.go
+++ b/sdk/messaging/azeventhubs/internal/eh/stress/tests/batch_stress_tester.go
@@ -13,7 +13,7 @@ import (
 	"os"
 	"time"
 
-	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azeventhubs"
+	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azeventhubs/v2"
 )
 
 /*
diff --git a/sdk/messaging/azeventhubs/internal/eh/stress/tests/multi_balance_tester.go b/sdk/messaging/azeventhubs/internal/eh/stress/tests/multi_balance_tester.go
index d2fd99f5c3..02b5b00cb1 100644
--- a/sdk/messaging/azeventhubs/internal/eh/stress/tests/multi_balance_tester.go
+++ b/sdk/messaging/azeventhubs/internal/eh/stress/tests/multi_balance_tester.go
@@ -12,7 +12,7 @@ import (
 	"strings"
 
 	"github.com/Azure/azure-sdk-for-go/sdk/internal/log"
-	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azeventhubs"
+	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azeventhubs/v2"
 )
 
 // MultiBalanceTester runs the BalanceTest multiple times against different
diff --git a/sdk/messaging/azeventhubs/internal/eh/stress/tests/processor_stress_tester.go b/sdk/messaging/azeventhubs/internal/eh/stress/tests/processor_stress_tester.go
index 7d3cd5576b..def81e2221 100644
--- a/sdk/messaging/azeventhubs/internal/eh/stress/tests/processor_stress_tester.go
+++ b/sdk/messaging/azeventhubs/internal/eh/stress/tests/processor_stress_tester.go
@@ -15,9 +15,9 @@ import (
 	"sync"
 	"time"
 
-	"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/messaging/azeventhubs/internal/test"
+	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azeventhubs/v2"
+	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azeventhubs/v2/checkpoints"
+	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azeventhubs/v2/internal/test"
 	"github.com/Azure/azure-sdk-for-go/sdk/storage/azblob/container"
 )
 
diff --git a/sdk/messaging/azeventhubs/internal/eh/stress/tests/shared.go b/sdk/messaging/azeventhubs/internal/eh/stress/tests/shared.go
index 1aee3a3495..6ffc6db319 100644
--- a/sdk/messaging/azeventhubs/internal/eh/stress/tests/shared.go
+++ b/sdk/messaging/azeventhubs/internal/eh/stress/tests/shared.go
@@ -18,10 +18,10 @@ import (
 	"github.com/Azure/azure-sdk-for-go/sdk/azcore/to"
 	"github.com/Azure/azure-sdk-for-go/sdk/azidentity"
 	azlog "github.com/Azure/azure-sdk-for-go/sdk/internal/log"
-	"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/messaging/azeventhubs/internal/eh/stress/shared"
-	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azeventhubs/internal/test"
+	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azeventhubs/v2"
+	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azeventhubs/v2/checkpoints"
+	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azeventhubs/v2/internal/eh/stress/shared"
+	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azeventhubs/v2/internal/test"
 	"github.com/Azure/azure-sdk-for-go/sdk/storage/azblob/container"
 	"github.com/joho/godotenv"
 )
@@ -294,7 +294,7 @@ func initCheckpointStore(ctx context.Context, containerName string, testData *st
 		}
 
 		if partProps.IsEmpty {
-			newCheckpoint.Offset = to.Ptr[int64](-1)
+			newCheckpoint.Offset = to.Ptr("-1")
 			newCheckpoint.SequenceNumber = to.Ptr[int64](0)
 		} else {
 			newCheckpoint.Offset = &partProps.LastEnqueuedOffset
diff --git a/sdk/messaging/azeventhubs/internal/errors_test.go b/sdk/messaging/azeventhubs/internal/errors_test.go
index ed728ca323..737a9f7b3c 100644
--- a/sdk/messaging/azeventhubs/internal/errors_test.go
+++ b/sdk/messaging/azeventhubs/internal/errors_test.go
@@ -11,8 +11,8 @@ import (
 	"testing"
 
 	"github.com/Azure/azure-sdk-for-go/sdk/azidentity"
-	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azeventhubs/internal/amqpwrap"
-	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azeventhubs/internal/exported"
+	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azeventhubs/v2/internal/amqpwrap"
+	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azeventhubs/v2/internal/exported"
 	"github.com/Azure/go-amqp"
 	"github.com/stretchr/testify/require"
 )
diff --git a/sdk/messaging/azeventhubs/internal/exported/connection_string_properties_test.go b/sdk/messaging/azeventhubs/internal/exported/connection_string_properties_test.go
index f9d0353f7a..f14337afe4 100644
--- a/sdk/messaging/azeventhubs/internal/exported/connection_string_properties_test.go
+++ b/sdk/messaging/azeventhubs/internal/exported/connection_string_properties_test.go
@@ -9,7 +9,7 @@ import (
 	"github.com/stretchr/testify/require"
 
 	"github.com/Azure/azure-sdk-for-go/sdk/azcore/to"
-	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azeventhubs/internal/exported"
+	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azeventhubs/v2/internal/exported"
 )
 
 var (
diff --git a/sdk/messaging/azeventhubs/internal/links_test.go b/sdk/messaging/azeventhubs/internal/links_test.go
index 2a9eea668d..2c3310f093 100644
--- a/sdk/messaging/azeventhubs/internal/links_test.go
+++ b/sdk/messaging/azeventhubs/internal/links_test.go
@@ -12,9 +12,9 @@ import (
 	"github.com/Azure/azure-sdk-for-go/sdk/azcore/to"
 	"github.com/Azure/azure-sdk-for-go/sdk/internal/log"
 	"github.com/Azure/azure-sdk-for-go/sdk/internal/test/credential"
-	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azeventhubs/internal/amqpwrap"
-	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azeventhubs/internal/exported"
-	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azeventhubs/internal/test"
+	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azeventhubs/v2/internal/amqpwrap"
+	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azeventhubs/v2/internal/exported"
+	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azeventhubs/v2/internal/test"
 	"github.com/Azure/go-amqp"
 	"github.com/stretchr/testify/require"
 )
diff --git a/sdk/messaging/azeventhubs/internal/links_unit_test.go b/sdk/messaging/azeventhubs/internal/links_unit_test.go
index a5bb31972e..dd0b31386e 100644
--- a/sdk/messaging/azeventhubs/internal/links_unit_test.go
+++ b/sdk/messaging/azeventhubs/internal/links_unit_test.go
@@ -8,9 +8,9 @@ import (
 	"fmt"
 	"testing"
 
-	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azeventhubs/internal/amqpwrap"
-	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azeventhubs/internal/mock"
-	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azeventhubs/internal/test"
+	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azeventhubs/v2/internal/amqpwrap"
+	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azeventhubs/v2/internal/mock"
+	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azeventhubs/v2/internal/test"
 	"github.com/Azure/go-amqp"
 	"github.com/golang/mock/gomock"
 	"github.com/stretchr/testify/require"
diff --git a/sdk/messaging/azeventhubs/internal/namespace_test.go b/sdk/messaging/azeventhubs/internal/namespace_test.go
index 3d29498959..f63abcc6ec 100644
--- a/sdk/messaging/azeventhubs/internal/namespace_test.go
+++ b/sdk/messaging/azeventhubs/internal/namespace_test.go
@@ -13,11 +13,11 @@ import (
 	"github.com/Azure/azure-sdk-for-go/sdk/azcore"
 	"github.com/Azure/azure-sdk-for-go/sdk/azcore/policy"
 	"github.com/Azure/azure-sdk-for-go/sdk/internal/telemetry"
-	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azeventhubs/internal/amqpwrap"
-	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azeventhubs/internal/auth"
-	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azeventhubs/internal/exported"
-	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azeventhubs/internal/sbauth"
-	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azeventhubs/internal/test"
+	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azeventhubs/v2/internal/amqpwrap"
+	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azeventhubs/v2/internal/auth"
+	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azeventhubs/v2/internal/exported"
+	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azeventhubs/v2/internal/sbauth"
+	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azeventhubs/v2/internal/test"
 	"github.com/Azure/go-amqp"
 	"github.com/stretchr/testify/require"
 )
diff --git a/sdk/messaging/azeventhubs/internal/rpc_test.go b/sdk/messaging/azeventhubs/internal/rpc_test.go
index a9687cc1a4..1e4d6e5971 100644
--- a/sdk/messaging/azeventhubs/internal/rpc_test.go
+++ b/sdk/messaging/azeventhubs/internal/rpc_test.go
@@ -12,9 +12,9 @@ import (
 	"time"
 
 	"github.com/Azure/azure-sdk-for-go/sdk/internal/log"
-	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azeventhubs/internal/amqpwrap"
-	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azeventhubs/internal/mock"
-	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azeventhubs/internal/test"
+	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azeventhubs/v2/internal/amqpwrap"
+	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azeventhubs/v2/internal/mock"
+	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azeventhubs/v2/internal/test"
 	"github.com/Azure/go-amqp"
 	"github.com/golang/mock/gomock"
 	"github.com/stretchr/testify/require"
diff --git a/sdk/messaging/azeventhubs/internal/sas/sas_test.go b/sdk/messaging/azeventhubs/internal/sas/sas_test.go
index 51d3e57619..458c6e50b6 100644
--- a/sdk/messaging/azeventhubs/internal/sas/sas_test.go
+++ b/sdk/messaging/azeventhubs/internal/sas/sas_test.go
@@ -10,7 +10,7 @@ import (
 	"testing"
 	"time"
 
-	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azeventhubs/internal/auth"
+	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azeventhubs/v2/internal/auth"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 )
diff --git a/sdk/messaging/azeventhubs/internal/test/test_helpers.go b/sdk/messaging/azeventhubs/internal/test/test_helpers.go
index cea38865db..861b465284 100644
--- a/sdk/messaging/azeventhubs/internal/test/test_helpers.go
+++ b/sdk/messaging/azeventhubs/internal/test/test_helpers.go
@@ -18,7 +18,7 @@ import (
 	"github.com/Azure/azure-sdk-for-go/sdk/azcore"
 	azlog "github.com/Azure/azure-sdk-for-go/sdk/internal/log"
 	"github.com/Azure/azure-sdk-for-go/sdk/internal/test/credential"
-	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azeventhubs/internal/exported"
+	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azeventhubs/v2/internal/exported"
 	"github.com/joho/godotenv"
 	"github.com/stretchr/testify/require"
 )
@@ -99,6 +99,9 @@ type ConnectionParamsForTest struct {
 	EventHubName          string
 	EventHubLinksOnlyName string
 	EventHubNamespace     string
+	GeoDRNamespace        string // optional: resource requires special setup to create, so it's manual
+	GeoDRHubName          string // optional: resource requires special setup to create, so it's manual
+	GeoDRStorageEndpoint  string // optional: resource requires special setup to create, so it's manual
 	StorageEndpoint       string
 	ResourceGroup         string
 	SubscriptionID        string
@@ -150,6 +153,9 @@ func GetConnectionParamsForTest(t *testing.T) ConnectionParamsForTest {
 		EventHubName:          envVars["EVENTHUB_NAME"],
 		EventHubLinksOnlyName: envVars["EVENTHUB_LINKSONLY_NAME"],
 		EventHubNamespace:     envVars["EVENTHUB_NAMESPACE"],
+		GeoDRNamespace:        os.Getenv("EVENTHUBS_GEODR_NAMESPACE"),
+		GeoDRHubName:          os.Getenv("EVENTHUBS_GEODR_HUBNAME"),
+		GeoDRStorageEndpoint:  os.Getenv("EVENTHUBS_GEODR_CHECKPOINTSTORE_STORAGE_ENDPOINT"),
 		ResourceGroup:         envVars["RESOURCE_GROUP"],
 		StorageEndpoint:       envVars["CHECKPOINTSTORE_STORAGE_ENDPOINT"],
 		SubscriptionID:        envVars["AZURE_SUBSCRIPTION_ID"],
diff --git a/sdk/messaging/azeventhubs/internal/utils/retrier_test.go b/sdk/messaging/azeventhubs/internal/utils/retrier_test.go
index 201423a8eb..ef4d52200d 100644
--- a/sdk/messaging/azeventhubs/internal/utils/retrier_test.go
+++ b/sdk/messaging/azeventhubs/internal/utils/retrier_test.go
@@ -13,7 +13,7 @@ import (
 	"time"
 
 	azlog "github.com/Azure/azure-sdk-for-go/sdk/internal/log"
-	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azeventhubs/internal/exported"
+	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azeventhubs/v2/internal/exported"
 	"github.com/Azure/go-amqp"
 	"github.com/stretchr/testify/require"
 )
diff --git a/sdk/messaging/azeventhubs/main_test.go b/sdk/messaging/azeventhubs/main_test.go
deleted file mode 100644
index e15a7a0948..0000000000
--- a/sdk/messaging/azeventhubs/main_test.go
+++ /dev/null
@@ -1,64 +0,0 @@
-// Copyright (c) Microsoft Corporation. All rights reserved.
-// Licensed under the MIT License.
-
-package azeventhubs_test
-
-import (
-	"context"
-	"fmt"
-	"log"
-	"os"
-	"testing"
-	"time"
-
-	"github.com/Azure/azure-sdk-for-go/sdk/internal/recording"
-	"github.com/Azure/azure-sdk-for-go/sdk/internal/test/credential"
-	"github.com/Azure/azure-sdk-for-go/sdk/storage/azblob"
-	"github.com/joho/godotenv"
-)
-
-func TestMain(m *testing.M) {
-	ec := testMain(m)
-	os.Exit(ec)
-}
-
-func testMain(m *testing.M) int {
-	if recording.GetRecordMode() == recording.LiveMode {
-		// if one of our vars isn't defined just assume that they're supposed to come from an .env file
-		if os.Getenv("CHECKPOINTSTORE_STORAGE_ENDPOINT") == "" {
-			if err := godotenv.Load(); err != nil {
-				log.Printf("Failed to load .env file when running live tests: %s", err.Error())
-				return 1
-			}
-		}
-
-		// create a test storage container so our examples can run.
-		tokenCred, err := credential.New(nil)
-
-		if err != nil {
-			log.Printf("Failed to create DAC: %s", err)
-			return 1
-		}
-
-		blobClient, err := azblob.NewClient(os.Getenv("CHECKPOINTSTORE_STORAGE_ENDPOINT"), tokenCred, nil)
-
-		if err != nil {
-			log.Printf("Failed to create blob client: %s", err)
-			return 1
-		}
-
-		containerName := fmt.Sprintf("container-%x", time.Now().UnixNano())
-
-		defer func() { _, _ = blobClient.ServiceClient().DeleteContainer(context.Background(), containerName, nil) }()
-
-		if _, err := blobClient.ServiceClient().CreateContainer(context.Background(), containerName, nil); err != nil {
-			log.Printf("Failed to create blob container for examples: %s", err)
-			return 1
-		}
-
-		log.Printf("Example checkpoint container created: %s", containerName)
-		os.Setenv("CHECKPOINTSTORE_STORAGE_CONTAINER_NAME", containerName)
-	}
-
-	return m.Run()
-}
diff --git a/sdk/messaging/azeventhubs/partition_client_unit_test.go b/sdk/messaging/azeventhubs/partition_client_unit_test.go
index 7560dbf7e2..28d6aac896 100644
--- a/sdk/messaging/azeventhubs/partition_client_unit_test.go
+++ b/sdk/messaging/azeventhubs/partition_client_unit_test.go
@@ -9,8 +9,8 @@ import (
 	"testing"
 	"time"
 
-	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azeventhubs/internal"
-	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azeventhubs/internal/test"
+	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azeventhubs/v2/internal"
+	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azeventhubs/v2/internal/test"
 	"github.com/Azure/go-amqp"
 	"github.com/stretchr/testify/require"
 )
diff --git a/sdk/messaging/azeventhubs/processor_test.go b/sdk/messaging/azeventhubs/processor_test.go
index 4b307362dd..f5a9caaa83 100644
--- a/sdk/messaging/azeventhubs/processor_test.go
+++ b/sdk/messaging/azeventhubs/processor_test.go
@@ -14,10 +14,10 @@ import (
 	"testing"
 	"time"
 
-	"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/messaging/azeventhubs/internal/exported"
-	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azeventhubs/internal/test"
+	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azeventhubs/v2"
+	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azeventhubs/v2/checkpoints"
+	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azeventhubs/v2/internal/exported"
+	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azeventhubs/v2/internal/test"
 	"github.com/Azure/azure-sdk-for-go/sdk/storage/azblob/container"
 	"github.com/stretchr/testify/require"
 )
@@ -223,7 +223,7 @@ func TestProcessor_Contention(t *testing.T) {
 }
 
 func TestProcessor_RunNotConcurrent(t *testing.T) {
-	newProc := setupProcessorTest(t)
+	newProc := setupProcessorTest(t, false).Create
 	proc := newProc(nil)
 
 	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
@@ -258,7 +258,7 @@ func TestProcessor_RunNotConcurrent(t *testing.T) {
 
 func TestProcessor_Run(t *testing.T) {
 	t.Run("VariationsWithoutCallingNextPartitionClient", func(t *testing.T) {
-		newProc := setupProcessorTest(t)
+		newProc := setupProcessorTest(t, false).Create
 
 		processor := newProc(&azeventhubs.ProcessorOptions{
 			UpdateInterval:        time.Millisecond,
@@ -285,7 +285,7 @@ func TestProcessor_Run(t *testing.T) {
 	})
 
 	t.Run("NextPartitionClient quits properly when Run is stopped.", func(t *testing.T) {
-		newProc := setupProcessorTest(t)
+		newProc := setupProcessorTest(t, false).Create
 
 		processor := newProc(&azeventhubs.ProcessorOptions{
 			UpdateInterval:        time.Millisecond,
@@ -328,7 +328,7 @@ func TestProcessor_Run(t *testing.T) {
 	})
 
 	t.Run("Run stopped quickly", func(t *testing.T) {
-		newProc := setupProcessorTest(t)
+		newProc := setupProcessorTest(t, false).Create
 
 		processor := newProc(&azeventhubs.ProcessorOptions{
 			UpdateInterval:        time.Millisecond,
@@ -399,11 +399,24 @@ func newNextPartitionCounter(t *testing.T, processor *azeventhubs.Processor) str
 	}
 }
 
-func setupProcessorTest(t *testing.T) func(*azeventhubs.ProcessorOptions) *azeventhubs.Processor {
+type processorTestData struct {
+	Create          func(*azeventhubs.ProcessorOptions) *azeventhubs.Processor
+	CheckpointStore azeventhubs.CheckpointStore
+}
+
+func setupProcessorTest(t *testing.T, geoDR bool) *processorTestData {
 	testParams := test.GetConnectionParamsForTest(t)
 
 	containerName := test.RandomString("proctest", 10)
-	cc, err := container.NewClient(test.URLJoinPaths(testParams.StorageEndpoint, containerName), testParams.Cred, nil)
+
+	storageEndpoint := testParams.StorageEndpoint
+
+	if geoDR {
+		storageEndpoint = testParams.GeoDRStorageEndpoint
+	}
+
+	cc, err := container.NewClient(test.URLJoinPaths(storageEndpoint, containerName), testParams.Cred, nil)
+
 	require.NoError(t, err)
 
 	t.Logf("Creating storage container %s", containerName)
@@ -422,8 +435,14 @@ func setupProcessorTest(t *testing.T) func(*azeventhubs.ProcessorOptions) *azeve
 	checkpointStore, err := checkpoints.NewBlobStore(cc, nil)
 	require.NoError(t, err)
 
+	namespace, eventHubName := testParams.EventHubNamespace, testParams.EventHubName
+
+	if geoDR {
+		namespace, eventHubName = testParams.GeoDRNamespace, testParams.GeoDRHubName
+	}
+
 	t.Logf("Consumer client created")
-	consumerClient, err := azeventhubs.NewConsumerClient(testParams.EventHubNamespace, testParams.EventHubName, azeventhubs.DefaultConsumerGroup, testParams.Cred, nil)
+	consumerClient, err := azeventhubs.NewConsumerClient(namespace, eventHubName, azeventhubs.DefaultConsumerGroup, testParams.Cred, nil)
 	require.NoError(t, err)
 
 	t.Cleanup(func() {
@@ -431,13 +450,18 @@ func setupProcessorTest(t *testing.T) func(*azeventhubs.ProcessorOptions) *azeve
 		require.NoError(t, err)
 	})
 
-	return func(options *azeventhubs.ProcessorOptions) *azeventhubs.Processor {
+	mustCreateProcessor := func(options *azeventhubs.ProcessorOptions) *azeventhubs.Processor {
 		t.Logf("Processor created")
 		processor, err := azeventhubs.NewProcessor(consumerClient, checkpointStore, options)
 		require.NoError(t, err)
 
 		return processor
 	}
+
+	return &processorTestData{
+		Create:          mustCreateProcessor,
+		CheckpointStore: checkpointStore,
+	}
 }
 
 func testPartitionAcquisition(t *testing.T, loadBalancerStrategy azeventhubs.ProcessorStrategy) {
diff --git a/sdk/messaging/azeventhubs/processor_unit_test.go b/sdk/messaging/azeventhubs/processor_unit_test.go
index bfce08fde9..eecd586754 100644
--- a/sdk/messaging/azeventhubs/processor_unit_test.go
+++ b/sdk/messaging/azeventhubs/processor_unit_test.go
@@ -4,6 +4,7 @@ package azeventhubs
 
 import (
 	"context"
+	"errors"
 	"fmt"
 	"sync"
 	"sync/atomic"
@@ -12,15 +13,16 @@ import (
 
 	"github.com/Azure/azure-sdk-for-go/sdk/azcore/log"
 	"github.com/Azure/azure-sdk-for-go/sdk/azcore/to"
-	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azeventhubs/internal"
-	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azeventhubs/internal/amqpwrap"
-	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azeventhubs/internal/exported"
+	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azeventhubs/v2/internal"
+	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azeventhubs/v2/internal/amqpwrap"
+	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azeventhubs/v2/internal/exported"
+	"github.com/Azure/go-amqp"
 	"github.com/stretchr/testify/require"
 )
 
 func TestUnit_Processor_loadBalancing(t *testing.T) {
 	cps := newCheckpointStoreForTest()
-	firstProcessor := newProcessorForTest(t, "first-processor", cps)
+	firstProcessor := newProcessorForTest(t, "first-processor", cps, nil)
 	newTestOwnership := func(base Ownership) Ownership {
 		base.ConsumerGroup = "consumer-group"
 		base.EventHubName = "event-hub"
@@ -80,7 +82,7 @@ func TestUnit_Processor_loadBalancing(t *testing.T) {
 	// 1 of those partitions is owned by our client ("first-processor")
 	// 2 are still available.
 
-	secondProcessor := newProcessorForTest(t, "second-processor", cps)
+	secondProcessor := newProcessorForTest(t, "second-processor", cps, nil)
 
 	// when we ask for available partitions we take into account the owners that are
 	// present in the checkpoint store and ourselves, since we're about to try to claim
@@ -279,7 +281,7 @@ func TestUnit_Processor_Run_startPosition(t *testing.T) {
 
 func TestUnit_Processor_RunCancelledQuickly(t *testing.T) {
 	cps := newCheckpointStoreForTest()
-	processor := newProcessorForTest(t, "processor", cps)
+	processor := newProcessorForTest(t, "processor", cps, nil)
 
 	precancelledContext, cancel := context.WithCancel(context.Background())
 	cancel()
@@ -323,6 +325,282 @@ func TestUnit_Processor_Run_cancellation(t *testing.T) {
 	require.NoError(t, err)
 }
 
+func TestUnit_Processor_getStartPosition_checkpointStore(t *testing.T) {
+	ignoredSP := StartPosition{Offset: to.Ptr("IGNORED: checkpoint wins")}
+
+	checkpointTests := []struct {
+		Name           string
+		Checkpoint     Checkpoint
+		StartPositions StartPositions
+		Actual         StartPosition
+	}{
+		{
+			Name:           "Offset wins over Sequence, when present",
+			Checkpoint:     Checkpoint{SequenceNumber: to.Ptr[int64](123), Offset: to.Ptr("offset from checkpoint")},
+			StartPositions: StartPositions{Default: ignoredSP},
+			Actual:         StartPosition{Offset: to.Ptr("offset from checkpoint")},
+		},
+		{
+			Name:           "Use SequenceNumber, if no offset",
+			Checkpoint:     Checkpoint{SequenceNumber: to.Ptr[int64](123)},
+			StartPositions: StartPositions{Default: ignoredSP},
+			Actual:         StartPosition{SequenceNumber: to.Ptr[int64](123)},
+		},
+		{
+			Name:           "Overrides Default",
+			Checkpoint:     Checkpoint{Offset: to.Ptr("offset from checkpoint")},
+			StartPositions: StartPositions{Default: ignoredSP},
+			Actual:         StartPosition{Offset: to.Ptr("offset from checkpoint")},
+		},
+		{
+			Name:       "Overrides Partition Default",
+			Checkpoint: Checkpoint{Offset: to.Ptr("offset from checkpoint")},
+			StartPositions: StartPositions{
+				PerPartition: map[string]StartPosition{"a": {Offset: to.Ptr("IGNORED: checkpoint wins")}},
+			},
+			Actual: StartPosition{Offset: to.Ptr("offset from checkpoint")},
+		},
+	}
+
+	for _, td := range checkpointTests {
+		cps := newCheckpointStoreForTest()
+		td.Checkpoint.ConsumerGroup = "cg"
+		td.Checkpoint.EventHubName = "eh"
+		td.Checkpoint.FullyQualifiedNamespace = "ns"
+		td.Checkpoint.PartitionID = "a"
+
+		err := cps.SetCheckpoint(context.Background(), td.Checkpoint, nil)
+		require.NoError(t, err)
+
+		processor := newProcessorForTest(t, "client-id", cps, &ProcessorOptions{
+			StartPositions: td.StartPositions,
+		})
+
+		called := false
+
+		actualStartPosition, err := processor.getStartPosition(func() (map[string]Checkpoint, error) {
+			called = true
+			return processor.getCheckpointsMap(context.Background())
+		}, Ownership{PartitionID: "a"})
+		require.NoError(t, err)
+
+		require.True(t, called)
+		require.Equal(t, td.Actual, actualStartPosition)
+	}
+
+	t.Run("InvalidCheckpoint", func(t *testing.T) {
+		cps := newCheckpointStoreForTest()
+
+		err := cps.SetCheckpoint(context.Background(), Checkpoint{
+			ConsumerGroup: "cg", EventHubName: "eh", FullyQualifiedNamespace: "ns", PartitionID: "a",
+			// no offset or sequence number set
+		}, nil)
+		require.NoError(t, err)
+
+		processor := newProcessorForTest(t, "client-id", cps, &ProcessorOptions{
+			StartPositions: StartPositions{Default: ignoredSP},
+		})
+
+		called := false
+
+		actualStartPosition, err := processor.getStartPosition(func() (map[string]Checkpoint, error) {
+			called = true
+			return processor.getCheckpointsMap(context.Background())
+		}, Ownership{PartitionID: "a"})
+		require.Empty(t, actualStartPosition)
+		require.Error(t, err)
+		require.True(t, called)
+	})
+}
+
+func TestUnit_Processor_getStartPosition_noCheckpoint(t *testing.T) {
+	t.Run("PartitionPreferredOverDefault", func(t *testing.T) {
+		cps := newCheckpointStoreForTest()
+		called := false
+
+		processor := newProcessorForTest(t, "client-id", cps, &ProcessorOptions{
+			StartPositions: StartPositions{
+				Default: StartPosition{
+					Offset: to.Ptr("default start position"),
+				},
+				PerPartition: map[string]StartPosition{
+					"a": {
+						Offset: to.Ptr("a partition start position"),
+					},
+				},
+			},
+		})
+
+		actualStartPosition, err := processor.getStartPosition(func() (map[string]Checkpoint, error) {
+			called = true
+			return processor.getCheckpointsMap(context.Background())
+		}, Ownership{PartitionID: "a"})
+		require.NoError(t, err)
+		require.True(t, called)
+		require.Equal(t, StartPosition{Offset: to.Ptr("a partition start position")}, actualStartPosition)
+	})
+
+	t.Run("DefaultUsedOtherwise_HasPartitionMap", func(t *testing.T) {
+		cps := newCheckpointStoreForTest()
+		called := false
+
+		processor := newProcessorForTest(t, "client-id", cps, &ProcessorOptions{
+			StartPositions: StartPositions{
+				Default: StartPosition{
+					Offset: to.Ptr("default start position"),
+				},
+				PerPartition: map[string]StartPosition{
+					"not-our-partition": {
+						Offset: to.Ptr("ignored, doesn't match partition"),
+					},
+				},
+			},
+		})
+
+		actualStartPosition, err := processor.getStartPosition(func() (map[string]Checkpoint, error) {
+			called = true
+			return processor.getCheckpointsMap(context.Background())
+		}, Ownership{PartitionID: "a"})
+		require.NoError(t, err)
+		require.True(t, called)
+		require.Equal(t, StartPosition{Offset: to.Ptr("default start position")}, actualStartPosition)
+	})
+
+	t.Run("DefaultUsedOtherwise_NoPartitionMap", func(t *testing.T) {
+		cps := newCheckpointStoreForTest()
+		called := false
+
+		processor := newProcessorForTest(t, "client-id", cps, &ProcessorOptions{
+			StartPositions: StartPositions{
+				Default: StartPosition{
+					Offset: to.Ptr("default start position"),
+				},
+			},
+		})
+
+		actualStartPosition, err := processor.getStartPosition(func() (map[string]Checkpoint, error) {
+			called = true
+			return processor.getCheckpointsMap(context.Background())
+		}, Ownership{PartitionID: "a"})
+		require.NoError(t, err)
+		require.True(t, called)
+		require.Equal(t, StartPosition{Offset: to.Ptr("default start position")}, actualStartPosition)
+	})
+}
+
+func TestUnit_Processor_addPartitionClient(t *testing.T) {
+	cps := newCheckpointStoreForTest()
+
+	err := cps.SetCheckpoint(context.Background(), Checkpoint{
+		ConsumerGroup: "cg", EventHubName: "eh", FullyQualifiedNamespace: "fqdn", PartitionID: "0",
+		Offset: to.Ptr("100"), // ie, a legacy offset
+	}, nil)
+	require.NoError(t, err)
+
+	processor := newProcessorForTest(t, "client-id", cps, &ProcessorOptions{})
+
+	getCheckpoints := func() (map[string]Checkpoint, error) {
+		return processor.getCheckpointsMap(context.Background())
+	}
+
+	geodrErr := &amqp.Error{
+		Condition:   amqp.ErrCond("com.microsoft:georeplication:invalid-offset"),
+		Description: "The supplied offset '1' is invalid for geo replication enabled namespace (current epoch:'1') (lots of other error text)",
+	}
+
+	t.Run("GeoDRFallback", func(t *testing.T) {
+		processor.nextClients = make(chan *ProcessorPartitionClient, 1)
+
+		gotGeodrFailure := false
+		called := 0
+		consumers := &sync.Map{}
+
+		openPartitionClient := func(ctx context.Context, partitionID string, startPosition StartPosition) (partitionClient *PartitionClient, err error) {
+			called++
+
+			if startPosition.Offset != nil {
+				gotGeodrFailure = true
+				return nil, geodrErr
+			}
+
+			return &PartitionClient{}, nil
+		}
+
+		err = processor.addPartitionClient(context.Background(), Ownership{PartitionID: "0"}, getCheckpoints, openPartitionClient, consumers)
+		require.NoError(t, err)
+		require.True(t, gotGeodrFailure)
+		require.Equal(t, 2, called)
+
+		pc, _ := consumers.Load("0")
+		require.NotNil(t, pc)
+	})
+
+	t.Run("Normal", func(t *testing.T) {
+		processor.nextClients = make(chan *ProcessorPartitionClient, 1)
+
+		called := 0
+		consumers := &sync.Map{}
+
+		openPartitionClient := func(ctx context.Context, partitionID string, startPosition StartPosition) (partitionClient *PartitionClient, err error) {
+			called++
+			return &PartitionClient{}, nil
+		}
+
+		err = processor.addPartitionClient(context.Background(), Ownership{PartitionID: "0"}, getCheckpoints, openPartitionClient, consumers)
+		require.NoError(t, err)
+		require.Equal(t, 1, called)
+
+		pc, _ := consumers.Load("0")
+		require.NotNil(t, pc)
+	})
+
+	t.Run("Failure", func(t *testing.T) {
+		processor.nextClients = make(chan *ProcessorPartitionClient, 1)
+
+		called := 0
+		consumers := &sync.Map{}
+
+		openPartitionClient := func(ctx context.Context, partitionID string, startPosition StartPosition) (partitionClient *PartitionClient, err error) {
+			called++
+			return nil, errors.New("any other error aborts immediately")
+		}
+
+		err = processor.addPartitionClient(context.Background(), Ownership{PartitionID: "0"}, getCheckpoints, openPartitionClient, consumers)
+		require.EqualError(t, err, "any other error aborts immediately")
+		require.Equal(t, 1, called, "We don't fall back if the error is not the specific GeoDR failure")
+
+		pc, _ := consumers.Load("0")
+		require.Nil(t, pc)
+	})
+
+	t.Run("GeoDR failure", func(t *testing.T) {
+		processor.nextClients = make(chan *ProcessorPartitionClient, 1)
+
+		called := 0
+		consumers := &sync.Map{}
+		gotGeodrFailure := false
+
+		openPartitionClient := func(ctx context.Context, partitionID string, startPosition StartPosition) (partitionClient *PartitionClient, err error) {
+			called++
+
+			if startPosition.Offset != nil {
+				gotGeodrFailure = true
+				return nil, geodrErr
+			}
+
+			return nil, errors.New("error trying to open geodr partition client")
+		}
+
+		err = processor.addPartitionClient(context.Background(), Ownership{PartitionID: "0"}, getCheckpoints, openPartitionClient, consumers)
+		require.EqualError(t, err, "error trying to open geodr partition client")
+		require.Equal(t, 2, called)
+		require.True(t, gotGeodrFailure)
+
+		pc, _ := consumers.Load("0")
+		require.Nil(t, pc)
+	})
+}
+
 // updateDynamicData updates the passed in `expected` Ownership with any fields that are
 // dynamically or randomly chosen. It returns the updated value.
 func updateDynamicData(t *testing.T, src Ownership, expected Ownership, allPartitionIDs []string) Ownership {
@@ -340,7 +618,15 @@ func updateDynamicData(t *testing.T, src Ownership, expected Ownership, allParti
 	return expected
 }
 
-func newProcessorForTest(t *testing.T, clientID string, cps CheckpointStore) *Processor {
+func newProcessorForTest(t *testing.T, clientID string, cps CheckpointStore, userOptions *ProcessorOptions) *Processor {
+	var options ProcessorOptions
+
+	if userOptions != nil {
+		options = *userOptions
+	}
+
+	options.PartitionExpirationDuration = time.Hour
+
 	processor, err := newProcessorImpl(&fakeConsumerClient{
 		details: consumerClientDetails{
 			ConsumerGroup:           "consumer-group",
@@ -348,9 +634,7 @@ func newProcessorForTest(t *testing.T, clientID string, cps CheckpointStore) *Pr
 			FullyQualifiedNamespace: "fqdn",
 			ClientID:                clientID,
 		},
-	}, cps, &ProcessorOptions{
-		PartitionExpirationDuration: time.Hour,
-	})
+	}, cps, &options)
 	require.NoError(t, err)
 	return processor
 }
diff --git a/sdk/messaging/azeventhubs/producer_client_test.go b/sdk/messaging/azeventhubs/producer_client_test.go
index ed57629a7b..314ab55ea0 100644
--- a/sdk/messaging/azeventhubs/producer_client_test.go
+++ b/sdk/messaging/azeventhubs/producer_client_test.go
@@ -16,9 +16,9 @@ import (
 	"github.com/Azure/azure-sdk-for-go/sdk/azcore/policy"
 	"github.com/Azure/azure-sdk-for-go/sdk/azcore/to"
 	"github.com/Azure/azure-sdk-for-go/sdk/azidentity"
-	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azeventhubs"
-	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azeventhubs/internal/sas"
-	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azeventhubs/internal/test"
+	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azeventhubs/v2"
+	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azeventhubs/v2/internal/sas"
+	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azeventhubs/v2/internal/test"
 	"github.com/stretchr/testify/require"
 )
 
diff --git a/sdk/monitor/azquery/metrics_client_test.go b/sdk/monitor/azquery/metrics_client_test.go
index 10b058ff0c..818fbf030d 100644
--- a/sdk/monitor/azquery/metrics_client_test.go
+++ b/sdk/monitor/azquery/metrics_client_test.go
@@ -8,11 +8,14 @@ package azquery_test
 
 import (
 	"context"
+	"net/http"
 	"testing"
 
 	"github.com/Azure/azure-sdk-for-go/sdk/azcore"
 	"github.com/Azure/azure-sdk-for-go/sdk/azcore/cloud"
 	"github.com/Azure/azure-sdk-for-go/sdk/azcore/to"
+	"github.com/Azure/azure-sdk-for-go/sdk/internal/mock"
+	azcred "github.com/Azure/azure-sdk-for-go/sdk/internal/test/credential"
 	"github.com/Azure/azure-sdk-for-go/sdk/monitor/azquery"
 	"github.com/stretchr/testify/require"
 )
@@ -156,3 +159,31 @@ func TestNewListNamespacesPager_Failure(t *testing.T) {
 	}
 
 }
+
+func TestMetricsAPIVersion(t *testing.T) {
+	apiVersion := "2023-10-01"
+	var requireVersion = func(t *testing.T) func(req *http.Request) bool {
+		return func(r *http.Request) bool {
+			version := r.URL.Query().Get("api-version")
+			require.Equal(t, version, apiVersion)
+			return true
+		}
+	}
+	srv, close := mock.NewServer(mock.WithTransformAllRequestsToTestServerUrl())
+	defer close()
+	srv.AppendResponse(
+		mock.WithStatusCode(200),
+		mock.WithPredicate(requireVersion(t)),
+	)
+	srv.AppendResponse(mock.WithStatusCode(http.StatusInternalServerError))
+	opts := &azquery.MetricsClientOptions{
+		ClientOptions: azcore.ClientOptions{
+			Transport:  srv,
+			APIVersion: apiVersion,
+		},
+	}
+	client, err := azquery.NewMetricsClient(&azcred.Fake{}, opts)
+	require.NoError(t, err)
+	_, err = client.QueryResource(context.Background(), resourceURI, nil)
+	require.NoError(t, err)
+}