Repository: Azure/azure-sdk-for-go
Pull request #17939 (2022-05-13 14:13:55)

tool: upgrade codegen and testgen version
<!--
Thank you for contributing to the Azure SDK for Go.

Please verify the following before submitting your PR, thank you!
-->

- [ ] The purpose of this PR is explained in this or a referenced issue.
- [ ] The PR does not update generated files.
   - These files are managed by the codegen framework at [Azure/autorest.go][].
- [ ] Tests are included and/or updated for code changes.
- [ ] Updates to [CHANGELOG.md][] are included.
- [ ] MIT license headers are included in each file.

[Azure/autorest.go]: https://github.com/Azure/autorest.go
[CHANGELOG.md]: https://github.com/Azure/azure-sdk-for-go/blob/main/CHANGELOG.md



Patch:
diff --git a/eng/scripts/build.ps1 b/eng/scripts/build.ps1
index cfc980b986..f9c4bc6e76 100644
--- a/eng/scripts/build.ps1
+++ b/eng/scripts/build.ps1
@@ -10,7 +10,7 @@ param(
     [switch]$tidy,
     [string]$config = "autorest.md",
     [string]$autorestVersion = "3.8.2",
-    [string]$goExtension = "@autorest/go@4.0.0-preview.39",
+    [string]$goExtension = "@autorest/go@4.0.0-preview.41",
     [string]$outputFolder
 )
 
diff --git a/sdk/data/azcosmos/cosmos_client.go b/sdk/data/azcosmos/cosmos_client.go
index 200097565b..e5b7e496b8 100644
--- a/sdk/data/azcosmos/cosmos_client.go
+++ b/sdk/data/azcosmos/cosmos_client.go
@@ -34,7 +34,7 @@ func (c *Client) Endpoint() string {
 // cred - The credential used to authenticate with the cosmos service.
 // options - Optional Cosmos client options.  Pass nil to accept default values.
 func NewClientWithKey(endpoint string, cred KeyCredential, o *ClientOptions) (*Client, error) {
-	return &Client{endpoint: endpoint, pipeline: newPipeline([]policy.Policy{newSharedKeyCredPolicy(cred)}, o)}, nil
+	return &Client{endpoint: endpoint, pipeline: newPipeline(newSharedKeyCredPolicy(cred), o)}, nil
 }
 
 // NewClient creates a new instance of Cosmos client with Azure AD access token authentication. It uses the default pipeline configuration.
@@ -46,10 +46,10 @@ func NewClient(endpoint string, cred azcore.TokenCredential, o *ClientOptions) (
 	if err != nil {
 		return nil, err
 	}
-	return &Client{endpoint: endpoint, pipeline: newPipeline([]policy.Policy{azruntime.NewBearerTokenPolicy(cred, scope, nil), &cosmosBearerTokenPolicy{}}, o)}, nil
+	return &Client{endpoint: endpoint, pipeline: newPipeline(newCosmosBearerTokenPolicy(cred, scope, nil), o)}, nil
 }
 
-func newPipeline(authPolicy []policy.Policy, options *ClientOptions) azruntime.Pipeline {
+func newPipeline(authPolicy policy.Policy, options *ClientOptions) azruntime.Pipeline {
 	if options == nil {
 		options = &ClientOptions{}
 	}
@@ -61,7 +61,9 @@ func newPipeline(authPolicy []policy.Policy, options *ClientOptions) azruntime.P
 					enableContentResponseOnWrite: options.EnableContentResponseOnWrite,
 				},
 			},
-			PerRetry: authPolicy,
+			PerRetry: []policy.Policy{
+				authPolicy,
+			},
 		},
 		&options.ClientOptions)
 }
diff --git a/sdk/data/azcosmos/cosmos_policy_bearer_token.go b/sdk/data/azcosmos/cosmos_policy_bearer_token.go
index dfba0e9b3c..da71fa52a2 100644
--- a/sdk/data/azcosmos/cosmos_policy_bearer_token.go
+++ b/sdk/data/azcosmos/cosmos_policy_bearer_token.go
@@ -4,25 +4,62 @@
 package azcosmos
 
 import (
-	"errors"
 	"fmt"
 	"net/http"
+	"time"
 
+	"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/temporal"
 )
 
-const lenBearerTokenPrefix = len("Bearer ")
-
+// cosmosBearerTokenPolicy authorizes requests with bearer tokens acquired from a TokenCredential.
+// Copy of sdk/azcore/runtime/policy_bearer_token.go using Cosmos header format
 type cosmosBearerTokenPolicy struct {
+	// mainResource is the resource to be retrieved using the tenant specified in the credential
+	mainResource *temporal.Resource[azcore.AccessToken, acquiringResourceState]
+	// the following fields are read-only
+	cred   azcore.TokenCredential
+	scopes []string
 }
 
-func (b *cosmosBearerTokenPolicy) Do(req *policy.Request) (*http.Response, error) {
-	currentAuthorization := req.Raw().Header.Get(headerAuthorization)
-	if currentAuthorization == "" {
-		return nil, errors.New("authorization header is missing")
+type acquiringResourceState struct {
+	req *policy.Request
+	p   *cosmosBearerTokenPolicy
+}
+
+// acquire acquires or updates the resource; only one
+// thread/goroutine at a time ever calls this function
+func acquire(state acquiringResourceState) (newResource azcore.AccessToken, newExpiration time.Time, err error) {
+	tk, err := state.p.cred.GetToken(state.req.Raw().Context(), policy.TokenRequestOptions{Scopes: state.p.scopes})
+	if err != nil {
+		return azcore.AccessToken{}, time.Time{}, err
 	}
+	return tk, tk.ExpiresOn, nil
+}
 
-	token := currentAuthorization[lenBearerTokenPrefix:]
-	req.Raw().Header.Set(headerAuthorization, fmt.Sprintf("type=aad&ver=1.0&sig=%v", token))
+// NewBearerTokenPolicy creates a policy object that authorizes requests with bearer tokens.
+// cred: an azcore.TokenCredential implementation such as a credential object from azidentity
+// scopes: the list of permission scopes required for the token.
+// opts: optional settings. Pass nil to accept default values; this is the same as passing a zero-value options.
+func newCosmosBearerTokenPolicy(cred azcore.TokenCredential, scopes []string, opts *policy.BearerTokenOptions) *cosmosBearerTokenPolicy {
+	return &cosmosBearerTokenPolicy{
+		cred:         cred,
+		scopes:       scopes,
+		mainResource: temporal.NewResource(acquire),
+	}
+}
+
+// Do authorizes a request with a bearer token
+func (b *cosmosBearerTokenPolicy) Do(req *policy.Request) (*http.Response, error) {
+	as := acquiringResourceState{
+		p:   b,
+		req: req,
+	}
+	tk, err := b.mainResource.Get(as)
+	if err != nil {
+		return nil, err
+	}
+	req.Raw().Header.Set(headerAuthorization, fmt.Sprintf("type=aad&ver=1.0&sig=%v", tk.Token))
 	return req.Next()
 }
diff --git a/sdk/messaging/azservicebus/admin/admin_client_queue.go b/sdk/messaging/azservicebus/admin/admin_client_queue.go
index 40ac9ca7e4..1bef8639ff 100644
--- a/sdk/messaging/azservicebus/admin/admin_client_queue.go
+++ b/sdk/messaging/azservicebus/admin/admin_client_queue.go
@@ -10,7 +10,7 @@ import (
 
 	"github.com/Azure/azure-sdk-for-go/sdk/azcore/runtime"
 	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azservicebus/internal/atom"
-	"github.com/Azure/azure-sdk-for-go/sdk/messaging/internal/auth"
+	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azservicebus/internal/auth"
 )
 
 // QueueProperties represents the static properties of the queue.
diff --git a/sdk/messaging/azservicebus/admin/admin_client_subscription.go b/sdk/messaging/azservicebus/admin/admin_client_subscription.go
index 931f599ab7..4d4a1c4bfe 100644
--- a/sdk/messaging/azservicebus/admin/admin_client_subscription.go
+++ b/sdk/messaging/azservicebus/admin/admin_client_subscription.go
@@ -11,7 +11,7 @@ import (
 
 	"github.com/Azure/azure-sdk-for-go/sdk/azcore/runtime"
 	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azservicebus/internal/atom"
-	"github.com/Azure/azure-sdk-for-go/sdk/messaging/internal/auth"
+	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azservicebus/internal/auth"
 )
 
 // SubscriptionProperties represents the static properties of the subscription.
diff --git a/sdk/messaging/azservicebus/admin/admin_client_topic.go b/sdk/messaging/azservicebus/admin/admin_client_topic.go
index 04f6c6a5df..31c1961843 100644
--- a/sdk/messaging/azservicebus/admin/admin_client_topic.go
+++ b/sdk/messaging/azservicebus/admin/admin_client_topic.go
@@ -10,7 +10,7 @@ import (
 
 	"github.com/Azure/azure-sdk-for-go/sdk/azcore/runtime"
 	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azservicebus/internal/atom"
-	"github.com/Azure/azure-sdk-for-go/sdk/messaging/internal/auth"
+	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azservicebus/internal/auth"
 )
 
 // TopicProperties represents the static properties of the topic.
diff --git a/sdk/messaging/azservicebus/go.mod b/sdk/messaging/azservicebus/go.mod
index 77911e7604..0f8a3155d8 100644
--- a/sdk/messaging/azservicebus/go.mod
+++ b/sdk/messaging/azservicebus/go.mod
@@ -6,7 +6,6 @@ require (
 	github.com/Azure/azure-sdk-for-go/sdk/azcore v1.0.0
 	github.com/Azure/azure-sdk-for-go/sdk/azidentity v1.0.0
 	github.com/Azure/azure-sdk-for-go/sdk/internal v1.0.0
-	github.com/Azure/azure-sdk-for-go/sdk/messaging/internal v0.1.1
 	github.com/Azure/go-amqp v0.17.4
 	golang.org/x/xerrors v0.0.0-20200804184101-5ec99f83aff1 // indirect
 )
diff --git a/sdk/messaging/azservicebus/go.sum b/sdk/messaging/azservicebus/go.sum
index d4e7834eb7..d84bf9d70f 100644
--- a/sdk/messaging/azservicebus/go.sum
+++ b/sdk/messaging/azservicebus/go.sum
@@ -6,28 +6,14 @@ github.com/Azure/azure-sdk-for-go/sdk/azidentity v1.0.0 h1:Yoicul8bnVdQrhDMTHxdE
 github.com/Azure/azure-sdk-for-go/sdk/azidentity v1.0.0/go.mod h1:+6sju8gk8FRmSajX3Oz4G5Gm7P+mbqE9FVaXXFYTkCM=
 github.com/Azure/azure-sdk-for-go/sdk/internal v1.0.0 h1:jp0dGvZ7ZK0mgqnTSClMxa5xuRL7NZgHameVYF6BurY=
 github.com/Azure/azure-sdk-for-go/sdk/internal v1.0.0/go.mod h1:eWRD7oawr1Mu1sLCawqVc0CUiF43ia3qQMxLscsKQ9w=
-github.com/Azure/azure-sdk-for-go/sdk/messaging/internal v0.1.1 h1:QAW1Y+jk6hk4Xst1n9cGRq9xBrBi/U3rYUcbvmox6lw=
-github.com/Azure/azure-sdk-for-go/sdk/messaging/internal v0.1.1/go.mod h1:L8126Syo70Ua6lotLqdLEPj5tbCfwGsyt6QQT2oPYFc=
-github.com/Azure/go-amqp v0.17.0/go.mod h1:9YJ3RhxRT1gquYnzpZO1vcYMMpAdJT+QEg6fwmw9Zlg=
 github.com/Azure/go-amqp v0.17.4 h1:6t9wEiwA4uXMRoUj3Cd3K2gmH8cW8ylizmBnSeF0bzM=
 github.com/Azure/go-amqp v0.17.4/go.mod h1:9YJ3RhxRT1gquYnzpZO1vcYMMpAdJT+QEg6fwmw9Zlg=
-github.com/Azure/go-autorest v14.2.0+incompatible/go.mod h1:r+4oMnoxhatjLLJ6zxSWATqVooLgysK6ZNox3g/xq24=
-github.com/Azure/go-autorest/autorest v0.11.22/go.mod h1:BAWYUWGPEtKPzjVkp0Q6an0MJcJDsoh5Z1BFAEFs4Xs=
-github.com/Azure/go-autorest/autorest/adal v0.9.14/go.mod h1:W/MM4U6nLxnIskrw4UwWzlHfGjwUS50aOsc/I3yuU8M=
-github.com/Azure/go-autorest/autorest/adal v0.9.17/go.mod h1:XVVeme+LZwABT8K5Lc3hA4nAe8LDBVle26gTrguhhPQ=
-github.com/Azure/go-autorest/autorest/date v0.3.0/go.mod h1:BI0uouVdmngYNUzGWeSYnokU+TrmwEsOqdt8Y6sso74=
-github.com/Azure/go-autorest/autorest/mocks v0.4.1/go.mod h1:LTp+uSrOhSkaKrUy935gNZuuIPPVsHlr9DSOxSayd+k=
-github.com/Azure/go-autorest/logger v0.2.1/go.mod h1:T9E3cAhj2VqvPOtCYAvby9aBXkZmbF5NWuPV8+WeEW8=
-github.com/Azure/go-autorest/tracing v0.6.0/go.mod h1:+vhtPC754Xsa23ID7GlGsrdKBpUA79WCAKPPZVC2DeU=
 github.com/AzureAD/microsoft-authentication-library-for-go v0.4.0 h1:WVsrXCnHlDDX8ls+tootqRE87/hL9S/g4ewig9RsD/c=
 github.com/AzureAD/microsoft-authentication-library-for-go v0.4.0/go.mod h1:Vt9sXTKwMyGcOxSmLDMnGPgqsUg7m8pe215qMLrDXw4=
 github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
 github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
 github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
-github.com/devigned/tab v0.1.1/go.mod h1:XG9mPq0dFghrYvoBF3xdRrJzSTX1b7IQrvaL9mzjeJY=
 github.com/dnaeon/go-vcr v1.1.0 h1:ReYa/UBrRyQdant9B4fNHGoCNKw6qh6P0fsdGmZpR7c=
-github.com/dnaeon/go-vcr v1.1.0/go.mod h1:M7tiix8f0r6mKKJ3Yq/kqU1OYf3MnfmBWVbPx/yU9ko=
-github.com/form3tech-oss/jwt-go v3.2.2+incompatible/go.mod h1:pbq4aXjuKjdthFRnoDwaVPLA+WlJuPGy+QneDUgJi2k=
 github.com/fortytw2/leaktest v1.3.0 h1:u8491cBMTQ8ft8aeV+adlcytMZylmA5nnwwkRZjI8vw=
 github.com/fortytw2/leaktest v1.3.0/go.mod h1:jDsjWgpAGjm2CA7WthBh/CdZYEPF31XHquHwclZch5g=
 github.com/fsnotify/fsnotify v1.4.7/go.mod h1:jwhsz4b93w/PPRr/qN1Yymfu8t87LnFCMoQvtojpjFo=
@@ -54,7 +40,6 @@ github.com/gofrs/uuid v3.3.0+incompatible h1:8K4tyRfvU1CYPgJsveYFQMhpFd/wXNM7iK6
 github.com/gofrs/uuid v3.3.0+incompatible/go.mod h1:b2aQJv3Z4Fp6yNu3cdSllBxTCLRxnplIgP/c0N/04lM=
 github.com/golang-jwt/jwt v3.2.1+incompatible h1:73Z+4BJcrTC+KczS6WvTPvRGOp1WmfEP4Q1lOd9Z/+c=
 github.com/golang-jwt/jwt v3.2.1+incompatible/go.mod h1:8pz2t5EyA70fFQQSrl6XZXzqecmYZeUEB8OUGHkxJ+I=
-github.com/golang-jwt/jwt/v4 v4.0.0/go.mod h1:/xlHOz8bRuivTWchD4jCa+NbatV+wEUSzwAxVc6locg=
 github.com/golang-jwt/jwt/v4 v4.2.0 h1:besgBTC8w8HjP6NzQdxwKH9Z5oQMZ24ThTrHp3cZ8eU=
 github.com/golang/protobuf v1.2.0/go.mod h1:6lQm79b+lXiMfvg/cZm0SGofjICqVBUtrP5yJMmIC1U=
 github.com/golang/protobuf v1.3.3/go.mod h1:vzj43D7+SQXF/4pzW/hwtAqwc6iTitCiVSaWz5lYuqw=
@@ -92,7 +77,6 @@ github.com/modern-go/concurrent v0.0.0-20180228061459-e0a39a4cb421 h1:ZqeYNhU3OH
 github.com/modern-go/concurrent v0.0.0-20180228061459-e0a39a4cb421/go.mod h1:6dJC0mAP4ikYIbvyc7fijjWJddQyLn8Ig3JB5CqoB9Q=
 github.com/modern-go/reflect2 v0.0.0-20180701023420-4b7aa43c6742 h1:Esafd1046DLDQ0W1YjYsBW+p8U2u7vzgW2SQVmlNazg=
 github.com/modern-go/reflect2 v0.0.0-20180701023420-4b7aa43c6742/go.mod h1:bx2lNnkwVCuqBIxFjflWJWanXIb3RllmbCylyMrvgv0=
-github.com/modocache/gover v0.0.0-20171022184752-b58185e213c5/go.mod h1:caMODM3PzxT8aQXRPkAt8xlV/e7d7w8GM5g0fa5F0D8=
 github.com/montanaflynn/stats v0.6.6/go.mod h1:etXPPgVO6n31NxCd9KQUMvCM+ve0ruNzt6R8Bnaayow=
 github.com/onsi/ginkgo v1.6.0/go.mod h1:lLunBs/Ym6LB5Z9jYTR76FiuTmxDTDusOGeTQH+WWjE=
 github.com/onsi/ginkgo v1.8.0/go.mod h1:lLunBs/Ym6LB5Z9jYTR76FiuTmxDTDusOGeTQH+WWjE=
@@ -113,14 +97,10 @@ github.com/ugorji/go/codec v1.1.7 h1:2SvQaVZ1ouYrrKKwoSk2pzd4A9evlKJb9oTL+OaLUSs
 github.com/ugorji/go/codec v1.1.7/go.mod h1:Ax+UKWsSmolVDwsd+7N3ZtXu+yMGCf907BLYF3GoBXY=
 golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w=
 golang.org/x/crypto v0.0.0-20200622213623-75b288015ac9/go.mod h1:LzIPMQfyMNhhGPhUkYOs5KpL4U8rLKemX1yGLhDgUto=
-golang.org/x/crypto v0.0.0-20201002170205-7f63de1d35b0/go.mod h1:LzIPMQfyMNhhGPhUkYOs5KpL4U8rLKemX1yGLhDgUto=
-golang.org/x/crypto v0.0.0-20210921155107-089bfa567519/go.mod h1:GvvjBRRGRdwPK5ydBHafDWAxML/pGHZbMvKqRZ5+Abc=
 golang.org/x/crypto v0.0.0-20220511200225-c6db032c6c88 h1:Tgea0cVUD0ivh5ADBX4WwuI12DUd2to3nCYe2eayMIw=
 golang.org/x/crypto v0.0.0-20220511200225-c6db032c6c88/go.mod h1:IxCIyHEi3zRg3s0A5j5BB6A9Jmi73HwBIUl50j+osU4=
 golang.org/x/net v0.0.0-20180906233101-161cd47e91fd/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4=
 golang.org/x/net v0.0.0-20190404232315-eb5bcb51f2a3/go.mod h1:t9HGtf8HONx5eT2rtn7q6eTqICYqUVnKs3thJo3Qplg=
-golang.org/x/net v0.0.0-20201010224723-4f7140c49acb/go.mod h1:sp8m0HH+o8qH0wwXwYZr8TS3Oi6o0r6Gce1SSxlDquU=
-golang.org/x/net v0.0.0-20210226172049-e18ecbb05110/go.mod h1:m0MpNAwzfU5UDzcl9v0D8zg8gWTRqZa9RBIspLL5mdg=
 golang.org/x/net v0.0.0-20220425223048-2871e0cb64e4 h1:HVyaeDAYux4pnY+D/SiwmLOR36ewZ4iGQIIrtnuCjFA=
 golang.org/x/net v0.0.0-20220425223048-2871e0cb64e4/go.mod h1:CfG3xpIq0wQ8r1q4Su4UZFWDARRcnwPjda9FqA0JpMk=
 golang.org/x/sync v0.0.0-20180314180146-1d60e4601c6f/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
@@ -128,16 +108,11 @@ golang.org/x/sys v0.0.0-20180909124046-d0be0721c37e/go.mod h1:STP8DvDyc/dI5b8T5h
 golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
 golang.org/x/sys v0.0.0-20190412213103-97732733099d/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20200116001909-b77594299b42/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
-golang.org/x/sys v0.0.0-20200930185726-fdedc70b468f/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
-golang.org/x/sys v0.0.0-20201119102817-f84b799fce68/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20210124154548-22da62e12c0c/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
-golang.org/x/sys v0.0.0-20210615035016-665e8c7367d1/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.0.0-20211216021012-1d35b9e2eb4e h1:fLOSk5Q00efkSvAm+4xcoXD+RRmLmmulPn5I3Y9F2EM=
 golang.org/x/sys v0.0.0-20211216021012-1d35b9e2eb4e/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
-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.2/go.mod h1:bEr9sfX3Q8Zfm5fL9x+3itogRgK3+ptLWKqgva+5dAk=
-golang.org/x/text v0.3.3/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ=
 golang.org/x/text v0.3.7 h1:olpwvP2KacW1ZWvsR7uQhoyTYvKAupfQrRGBFM352Gk=
 golang.org/x/text v0.3.7/go.mod h1:u+2+/6zg+i71rQMx5EYifcz6MCKuco9NR6JIITiCfzQ=
 golang.org/x/time v0.0.0-20191024005414-555d28b269f0/go.mod h1:tRJNPiyCQ0inRvYxbN9jk5I+vvW/OXSQhTDSoE431IQ=
@@ -154,7 +129,6 @@ gopkg.in/yaml.v2 v2.2.1/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=
 gopkg.in/yaml.v2 v2.2.2/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=
 gopkg.in/yaml.v2 v2.2.8/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=
 gopkg.in/yaml.v2 v2.4.0 h1:D8xgwECY7CYvx+Y2n4sBz93Jn9JRvxdiyyo8CTfuKaY=
-gopkg.in/yaml.v2 v2.4.0/go.mod h1:RDklbk79AGWmwhnvt/jBztapEOGDOx6ZbXqjP6csGnQ=
 gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
 gopkg.in/yaml.v3 v3.0.0-20210107192922-496545a6307b h1:h8qDotaEPuJATrMmW04NCwg7v22aHH28wwpauUhK9Oo=
 gopkg.in/yaml.v3 v3.0.0-20210107192922-496545a6307b/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
diff --git a/sdk/messaging/azservicebus/internal/atom/converters.go b/sdk/messaging/azservicebus/internal/atom/converters.go
index cb89b13bb0..702fff5185 100644
--- a/sdk/messaging/azservicebus/internal/atom/converters.go
+++ b/sdk/messaging/azservicebus/internal/atom/converters.go
@@ -5,7 +5,7 @@ package atom
 
 import (
 	"github.com/Azure/azure-sdk-for-go/sdk/azcore/to"
-	"github.com/Azure/azure-sdk-for-go/sdk/messaging/internal/auth"
+	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azservicebus/internal/auth"
 )
 
 func WrapWithQueueEnvelope(qd *QueueDescription, tokenProvider auth.TokenProvider) *QueueEnvelope {
diff --git a/sdk/messaging/azservicebus/internal/atom/entity_manager.go b/sdk/messaging/azservicebus/internal/atom/entity_manager.go
index dbbc8ba678..9216111b3d 100644
--- a/sdk/messaging/azservicebus/internal/atom/entity_manager.go
+++ b/sdk/messaging/azservicebus/internal/atom/entity_manager.go
@@ -17,9 +17,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/azcore/runtime"
+	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azservicebus/internal/auth"
+	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azservicebus/internal/conn"
 	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azservicebus/internal/sbauth"
-	"github.com/Azure/azure-sdk-for-go/sdk/messaging/internal/auth"
-	"github.com/Azure/azure-sdk-for-go/sdk/messaging/internal/conn"
 )
 
 const (
diff --git a/sdk/messaging/azservicebus/internal/atom/per_retry_auth_policy.go b/sdk/messaging/azservicebus/internal/atom/per_retry_auth_policy.go
index 2c39cf669c..16deb247c9 100644
--- a/sdk/messaging/azservicebus/internal/atom/per_retry_auth_policy.go
+++ b/sdk/messaging/azservicebus/internal/atom/per_retry_auth_policy.go
@@ -7,7 +7,7 @@ import (
 	"net/http"
 
 	"github.com/Azure/azure-sdk-for-go/sdk/azcore/policy"
-	"github.com/Azure/azure-sdk-for-go/sdk/messaging/internal/auth"
+	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azservicebus/internal/auth"
 )
 
 type perRetryAuthPolicy struct {
diff --git a/sdk/messaging/internal/auth/token.go b/sdk/messaging/azservicebus/internal/auth/token.go
similarity index 100%
rename from sdk/messaging/internal/auth/token.go
rename to sdk/messaging/azservicebus/internal/auth/token.go
diff --git a/sdk/messaging/azservicebus/internal/cbs.go b/sdk/messaging/azservicebus/internal/cbs.go
index 33a31e86ed..790bcf21d8 100644
--- a/sdk/messaging/azservicebus/internal/cbs.go
+++ b/sdk/messaging/azservicebus/internal/cbs.go
@@ -8,8 +8,8 @@ import (
 
 	azlog "github.com/Azure/azure-sdk-for-go/sdk/internal/log"
 	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azservicebus/internal/amqpwrap"
+	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azservicebus/internal/auth"
 	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azservicebus/internal/exported"
-	"github.com/Azure/azure-sdk-for-go/sdk/messaging/internal/auth"
 	"github.com/Azure/go-amqp"
 )
 
diff --git a/sdk/messaging/internal/conn/conn.go b/sdk/messaging/azservicebus/internal/conn/conn.go
similarity index 100%
rename from sdk/messaging/internal/conn/conn.go
rename to sdk/messaging/azservicebus/internal/conn/conn.go
diff --git a/sdk/messaging/internal/conn/conn_test.go b/sdk/messaging/azservicebus/internal/conn/conn_test.go
similarity index 100%
rename from sdk/messaging/internal/conn/conn_test.go
rename to sdk/messaging/azservicebus/internal/conn/conn_test.go
diff --git a/sdk/messaging/azservicebus/internal/namespace.go b/sdk/messaging/azservicebus/internal/namespace.go
index ba5a144c3b..3af63478f7 100644
--- a/sdk/messaging/azservicebus/internal/namespace.go
+++ b/sdk/messaging/azservicebus/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/messaging/azservicebus/internal/amqpwrap"
+	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azservicebus/internal/auth"
+	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azservicebus/internal/conn"
 	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azservicebus/internal/exported"
 	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azservicebus/internal/sbauth"
 	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azservicebus/internal/utils"
-	"github.com/Azure/azure-sdk-for-go/sdk/messaging/internal/auth"
-	"github.com/Azure/azure-sdk-for-go/sdk/messaging/internal/conn"
 	"github.com/Azure/go-amqp"
 )
 
diff --git a/sdk/messaging/internal/sas/sas.go b/sdk/messaging/azservicebus/internal/sas/sas.go
similarity index 96%
rename from sdk/messaging/internal/sas/sas.go
rename to sdk/messaging/azservicebus/internal/sas/sas.go
index e50e3aabeb..11fcdbec5c 100644
--- a/sdk/messaging/internal/sas/sas.go
+++ b/sdk/messaging/azservicebus/internal/sas/sas.go
@@ -15,8 +15,8 @@ import (
 	"strings"
 	"time"
 
-	"github.com/Azure/azure-sdk-for-go/sdk/messaging/internal/auth"
-	"github.com/Azure/azure-sdk-for-go/sdk/messaging/internal/conn"
+	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azservicebus/internal/auth"
+	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azservicebus/internal/conn"
 )
 
 type (
diff --git a/sdk/messaging/azservicebus/internal/sbauth/token_provider.go b/sdk/messaging/azservicebus/internal/sbauth/token_provider.go
index ba36841f07..8fddc63e5d 100644
--- a/sdk/messaging/azservicebus/internal/sbauth/token_provider.go
+++ b/sdk/messaging/azservicebus/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/internal/auth"
-	"github.com/Azure/azure-sdk-for-go/sdk/messaging/internal/conn"
-	"github.com/Azure/azure-sdk-for-go/sdk/messaging/internal/sas"
+	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azservicebus/internal/auth"
+	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azservicebus/internal/conn"
+	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azservicebus/internal/sas"
 )
 
 // TokenProvider handles access tokens and expiration calculation for SAS
diff --git a/sdk/messaging/azservicebus/internal/stress/.dockerignore b/sdk/messaging/azservicebus/internal/stress/.dockerignore
new file mode 100644
index 0000000000..4c49bd78f1
--- /dev/null
+++ b/sdk/messaging/azservicebus/internal/stress/.dockerignore
@@ -0,0 +1,1 @@
+.env
diff --git a/sdk/messaging/azservicebus/internal/stress/.gitignore b/sdk/messaging/azservicebus/internal/stress/.gitignore
index 2baebb1fc8..f4b2e23818 100644
--- a/sdk/messaging/azservicebus/internal/stress/.gitignore
+++ b/sdk/messaging/azservicebus/internal/stress/.gitignore
@@ -1,2 +1,4 @@
 stress
-stress.exe
\ No newline at end of file
+stress.exe
+logs
+
diff --git a/sdk/messaging/azservicebus/internal/stress/Chart.lock b/sdk/messaging/azservicebus/internal/stress/Chart.lock
index 96e9785515..7c69d4b7b9 100644
--- a/sdk/messaging/azservicebus/internal/stress/Chart.lock
+++ b/sdk/messaging/azservicebus/internal/stress/Chart.lock
@@ -1,6 +1,6 @@
 dependencies:
 - name: stress-test-addons
   repository: https://stresstestcharts.blob.core.windows.net/helm/
-  version: 0.1.12
-digest: sha256:a2afbc89375d1518cd44194ca861e8a7a1ca1c16cfaea3409cdb48bfc910d878
-generated: "2021-11-16T19:17:10.5545801-05:00"
+  version: 0.1.16
+digest: sha256:75061792fdbebe57c664322579af6dc240ce8f464280ac5e192de72a29d4f63f
+generated: "2022-04-29T00:43:35.81502307Z"
diff --git a/sdk/messaging/azservicebus/internal/stress/Chart.yaml b/sdk/messaging/azservicebus/internal/stress/Chart.yaml
index 044770e86e..337eb54a57 100644
--- a/sdk/messaging/azservicebus/internal/stress/Chart.yaml
+++ b/sdk/messaging/azservicebus/internal/stress/Chart.yaml
@@ -1,5 +1,5 @@
 apiVersion: v2
-name: go-sb-stress
+name: gosb
 description: Stress tests for Go
 version: 0.1.1
 appVersion: v0.1
@@ -10,5 +10,6 @@ annotations:
   dockerfile: './Dockerfile'
 dependencies:
 - name: stress-test-addons
-  version: 0.1.12
-  repository: https://stresstestcharts.blob.core.windows.net/helm/
\ No newline at end of file
+  version: 0.1.16
+  repository: https://stresstestcharts.blob.core.windows.net/helm/
+  
\ No newline at end of file
diff --git a/sdk/messaging/azservicebus/internal/stress/Dockerfile b/sdk/messaging/azservicebus/internal/stress/Dockerfile
index 3159517f34..4ffb508f27 100644
--- a/sdk/messaging/azservicebus/internal/stress/Dockerfile
+++ b/sdk/messaging/azservicebus/internal/stress/Dockerfile
@@ -1,4 +1,4 @@
-FROM golang as build
+FROM mcr.microsoft.com/oss/go/microsoft/golang:1.18 as build
 # you'll need to run this build from the root of the repo
 ENV GOOS=linux 
 ENV GOARCH=amd64 
@@ -7,8 +7,7 @@ ADD . /src
 WORKDIR /src/internal/stress
 RUN go build -o stress .
 
-FROM alpine
-RUN mkdir -p /app
-COPY --from=build /src/internal/stress/stress /app/stress
+FROM mcr.microsoft.com/cbl-mariner/base/core:1.0
 WORKDIR /app
-ENTRYPOINT ["./stress"]
\ No newline at end of file
+COPY --from=build /src/internal/stress/stress /app/stress
+ENTRYPOINT ["./stress"]
diff --git a/sdk/messaging/azservicebus/internal/stress/sb-network-loss.yaml b/sdk/messaging/azservicebus/internal/stress/sb-network-loss.yaml
new file mode 100644
index 0000000000..a48bbd5f43
--- /dev/null
+++ b/sdk/messaging/azservicebus/internal/stress/sb-network-loss.yaml
@@ -0,0 +1,26 @@
+{{- include "stress-test-addons.chaos-wrapper.tpl" (list . "stress.azservicebus-network") -}}
+{{- define "stress.azservicebus-network" -}}
+apiVersion: chaos-mesh.org/v1alpha1
+kind: NetworkChaos
+spec:
+  action: loss
+  direction: to
+  externalTargets:
+    # Maps to the service bus resource cname, provided the resource group name, provided
+    # the service bus namespace uses the resource group name as its name in the bicep template
+    - "{{ .Stress.ResourceGroupName }}.servicebus.windows.net"
+  mode: one
+  selector:
+    labelSelectors:
+      # Maps to the test pod, provided it also sets a testInstance label of {{ .Stress.ResourceGroupName }}
+      testInstance: infiniteSendAndReceive
+      chaos: "true"
+    namespaces:
+      - {{ .Release.Namespace }}
+  duration: 10s
+  scheduler:
+    cron: '*/3 * * * *'
+  loss:
+    loss: "100"
+    correlation: "100"
+{{- end -}}
diff --git a/sdk/messaging/azservicebus/internal/stress/shared/streaming_batch.go b/sdk/messaging/azservicebus/internal/stress/shared/streaming_batch.go
index 9035cf7c4c..19b1a27ceb 100644
--- a/sdk/messaging/azservicebus/internal/stress/shared/streaming_batch.go
+++ b/sdk/messaging/azservicebus/internal/stress/shared/streaming_batch.go
@@ -59,7 +59,6 @@ func (sb *StreamingMessageBatch) Add(ctx context.Context, msg *azservicebus.Mess
 	err := sb.currentBatch.AddMessage(msg, options)
 
 	if err == nil {
-		// sent, we're done
 		return nil
 	}
 
@@ -68,12 +67,14 @@ func (sb *StreamingMessageBatch) Add(ctx context.Context, msg *azservicebus.Mess
 		return err
 	}
 
-	log.Printf("Sending message batch")
+	log.Printf("Sending message batch (%d messages)", sb.currentBatch.NumMessages())
 	if err := sb.sender.SendMessageBatch(ctx, sb.currentBatch); err != nil {
 		return err
 	}
 
-	sb.stats.AddSent(sb.currentBatch.NumMessages())
+	if sb.stats != nil {
+		sb.stats.AddSent(sb.currentBatch.NumMessages())
+	}
 
 	// throttle a teeny bit.
 	time.Sleep(time.Second)
@@ -105,6 +106,8 @@ func (sb *StreamingMessageBatch) Close(ctx context.Context) error {
 		return err
 	}
 
-	sb.stats.AddSent(sb.currentBatch.NumMessages())
+	if sb.stats != nil {
+		sb.stats.AddSent(sb.currentBatch.NumMessages())
+	}
 	return nil
 }
diff --git a/sdk/messaging/azservicebus/internal/stress/shared/stress_context.go b/sdk/messaging/azservicebus/internal/stress/shared/stress_context.go
index 45cacf0570..24e461b648 100644
--- a/sdk/messaging/azservicebus/internal/stress/shared/stress_context.go
+++ b/sdk/messaging/azservicebus/internal/stress/shared/stress_context.go
@@ -65,7 +65,7 @@ func MustCreateStressContext(testName string) *StressContext {
 
 	ctx, cancel := NewCtrlCContext()
 
-	azlog.SetEvents(azservicebus.EventAdmin, azservicebus.EventAuth, azservicebus.EventConn, azservicebus.EventReceiver, azservicebus.EventSender)
+	azlog.SetEvents(azservicebus.EventSender, azservicebus.EventReceiver, azservicebus.EventConn, azservicebus.EventAuth)
 
 	logMessages := make(chan string, 10000)
 
diff --git a/sdk/messaging/azservicebus/internal/stress/templates/deploy-job.yaml b/sdk/messaging/azservicebus/internal/stress/templates/deploy-job.yaml
index 36748f54aa..17be8fc7af 100644
--- a/sdk/messaging/azservicebus/internal/stress/templates/deploy-job.yaml
+++ b/sdk/messaging/azservicebus/internal/stress/templates/deploy-job.yaml
@@ -16,6 +16,7 @@ spec:
       args: 
       - "tests"
       # (this is injected automatically. The full list of scenarios is in `../values.yaml`)
-      - {{ .Scenario }}
+      - {{ .Stress.Scenario }}
       {{- include "stress-test-addons.container-env" . | nindent 6 }}
 {{- end -}}
+
diff --git a/sdk/messaging/azservicebus/internal/stress/tools/generate_sas.go b/sdk/messaging/azservicebus/internal/stress/tools/generate_sas.go
index 197a828369..beb1960a75 100644
--- a/sdk/messaging/azservicebus/internal/stress/tools/generate_sas.go
+++ b/sdk/messaging/azservicebus/internal/stress/tools/generate_sas.go
@@ -9,8 +9,8 @@ import (
 	"os"
 	"time"
 
+	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azservicebus/internal/sas"
 	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azservicebus/internal/stress/shared"
-	"github.com/Azure/azure-sdk-for-go/sdk/messaging/internal/sas"
 )
 
 func GenerateSas(remainingArgs []string) int {
diff --git a/sdk/messaging/internal/CHANGELOG.md b/sdk/messaging/internal/CHANGELOG.md
deleted file mode 100644
index 1962cfe613..0000000000
--- a/sdk/messaging/internal/CHANGELOG.md
+++ /dev/null
@@ -1,21 +0,0 @@
-# Release History
-
-## 0.1.2 (Unreleased)
-
-### Features Added
-
-### Breaking Changes
-
-### Bugs Fixed
-
-### Other Changes
-
-## 0.1.1 (2022-05-12)
-
-### Other Changes
-
-- Update to v1 of azcore.
-
-## 0.1.0 (2022-03-29)
-
-- Internal package, only for support of azservicebus.
diff --git a/sdk/messaging/internal/LICENSE.txt b/sdk/messaging/internal/LICENSE.txt
deleted file mode 100644
index 48ea6616b5..0000000000
--- a/sdk/messaging/internal/LICENSE.txt
+++ /dev/null
@@ -1,21 +0,0 @@
-MIT License
-
-Copyright (c) Microsoft Corporation.
-
-Permission is hereby granted, free of charge, to any person obtaining a copy
-of this software and associated documentation files (the "Software"), to deal
-in the Software without restriction, including without limitation the rights
-to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
-copies of the Software, and to permit persons to whom the Software is
-furnished to do so, subject to the following conditions:
-
-The above copyright notice and this permission notice shall be included in all
-copies or substantial portions of the Software.
-
-THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
-IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
-FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
-AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
-LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
-OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
-SOFTWARE
diff --git a/sdk/messaging/internal/aad/jwt.go b/sdk/messaging/internal/aad/jwt.go
deleted file mode 100644
index 76d1a54e17..0000000000
--- a/sdk/messaging/internal/aad/jwt.go
+++ /dev/null
@@ -1,255 +0,0 @@
-// Package aad provides an implementation of an Azure Active Directory JWT provider which implements TokenProvider
-// from package auth for use with Azure Event Hubs and Service Bus.
-package aad
-
-//	MIT License
-//
-//	Copyright (c) Microsoft Corporation. All rights reserved.
-//
-//	Permission is hereby granted, free of charge, to any person obtaining a copy
-//	of this software and associated documentation files (the "Software"), to deal
-//	in the Software without restriction, including without limitation the rights
-//	to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
-//	copies of the Software, and to permit persons to whom the Software is
-//	furnished to do so, subject to the following conditions:
-//
-//	The above copyright notice and this permission notice shall be included in all
-//	copies or substantial portions of the Software.
-//
-//	THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
-//	IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
-//	FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
-//	AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
-//	LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
-//	OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
-//	SOFTWARE
-
-import (
-	"crypto/rsa"
-	"crypto/x509"
-	"fmt"
-	"io/ioutil"
-	"os"
-	"strconv"
-	"time"
-
-	"github.com/Azure/go-autorest/autorest/adal"
-	"github.com/Azure/go-autorest/autorest/azure"
-	"golang.org/x/crypto/pkcs12"
-
-	"github.com/Azure/azure-sdk-for-go/sdk/messaging/internal/auth"
-)
-
-const (
-	eventhubResourceURI = "https://eventhubs.azure.net/"
-)
-
-type (
-	// TokenProviderConfiguration provides configuration parameters for building JWT AAD providers
-	TokenProviderConfiguration struct {
-		TenantID            string
-		ClientID            string
-		ClientSecret        string
-		CertificatePath     string
-		CertificatePassword string
-		ResourceURI         string
-		aadToken            *adal.ServicePrincipalToken
-		Env                 *azure.Environment
-	}
-
-	// TokenProvider provides cbs.TokenProvider functionality for Azure Active Directory JWTs
-	TokenProvider struct {
-		tokenProvider *adal.ServicePrincipalToken
-	}
-
-	// JWTProviderOption provides configuration options for constructing AAD Token Providers
-	JWTProviderOption func(provider *TokenProviderConfiguration) error
-)
-
-// JWTProviderWithAzureEnvironment configures the token provider to use a specific Azure Environment
-func JWTProviderWithAzureEnvironment(env *azure.Environment) JWTProviderOption {
-	return func(config *TokenProviderConfiguration) error {
-		config.Env = env
-		return nil
-	}
-}
-
-// JWTProviderWithEnvironmentVars configures the TokenProvider using the environment variables available
-//
-// 1. Client Credentials: attempt to authenticate with a Service Principal via "AZURE_TENANT_ID", "AZURE_CLIENT_ID" and
-//    "AZURE_CLIENT_SECRET"
-//
-// 2. Client Certificate: attempt to authenticate with a Service Principal via "AZURE_TENANT_ID", "AZURE_CLIENT_ID",
-//    "AZURE_CERTIFICATE_PATH" and "AZURE_CERTIFICATE_PASSWORD"
-//
-// 3. Managed Service Identity (MSI): attempt to authenticate via MSI
-//
-//
-// The Azure Environment used can be specified using the name of the Azure Environment set in "AZURE_ENVIRONMENT" var.
-func JWTProviderWithEnvironmentVars() JWTProviderOption {
-	return func(config *TokenProviderConfiguration) error {
-		config.TenantID = os.Getenv("AZURE_TENANT_ID")
-		config.ClientID = os.Getenv("AZURE_CLIENT_ID")
-		config.ClientSecret = os.Getenv("AZURE_CLIENT_SECRET")
-		config.CertificatePath = os.Getenv("AZURE_CERTIFICATE_PATH")
-		config.CertificatePassword = os.Getenv("AZURE_CERTIFICATE_PASSWORD")
-
-		if config.Env == nil {
-			env, err := azureEnvFromEnvironment()
-			if err != nil {
-				return err
-			}
-			config.Env = env
-		}
-		return nil
-	}
-}
-
-// JWTProviderWithResourceURI configures the token provider to use a specific eventhubResourceURI URI
-func JWTProviderWithResourceURI(resourceURI string) JWTProviderOption {
-	return func(config *TokenProviderConfiguration) error {
-		config.ResourceURI = resourceURI
-		return nil
-	}
-}
-
-// JWTProviderWithAADToken configures the token provider to use a specific Azure Active Directory Service Principal token
-func JWTProviderWithAADToken(aadToken *adal.ServicePrincipalToken) JWTProviderOption {
-	return func(config *TokenProviderConfiguration) error {
-		config.aadToken = aadToken
-		return nil
-	}
-}
-
-// NewJWTProvider builds an Azure Active Directory claims-based security token provider
-func NewJWTProvider(opts ...JWTProviderOption) (*TokenProvider, error) {
-	config := &TokenProviderConfiguration{
-		ResourceURI: eventhubResourceURI,
-	}
-
-	for _, opt := range opts {
-		err := opt(config)
-		if err != nil {
-			return nil, err
-		}
-	}
-
-	if config.aadToken == nil {
-		spToken, err := config.NewServicePrincipalToken()
-		if err != nil {
-			return nil, err
-		}
-		config.aadToken = spToken
-	}
-	return &TokenProvider{tokenProvider: config.aadToken}, nil
-}
-
-// NewServicePrincipalToken creates a new Azure Active Directory Service Principal token provider
-func (c *TokenProviderConfiguration) NewServicePrincipalToken() (*adal.ServicePrincipalToken, error) {
-	oauthConfig, err := adal.NewOAuthConfig(c.Env.ActiveDirectoryEndpoint, c.TenantID)
-	if err != nil {
-		return nil, err
-	}
-
-	// 1.Client Credentials
-	if c.ClientSecret != "" {
-		spToken, err := adal.NewServicePrincipalToken(*oauthConfig, c.ClientID, c.ClientSecret, c.ResourceURI)
-		if err != nil {
-			return nil, fmt.Errorf("failed to get oauth token from client credentials: %v", err)
-		}
-		if err := spToken.Refresh(); err != nil {
-			return nil, fmt.Errorf("failed to refersh token: %v", spToken)
-		}
-		return spToken, nil
-	}
-
-	// 2. Client Certificate
-	if c.CertificatePath != "" {
-		certData, err := ioutil.ReadFile(c.CertificatePath)
-		if err != nil {
-			return nil, fmt.Errorf("failed to read the certificate file (%s): %v", c.CertificatePath, err)
-		}
-		certificate, rsaPrivateKey, err := decodePkcs12(certData, c.CertificatePassword)
-		if err != nil {
-			return nil, fmt.Errorf("failed to decode pkcs12 certificate while creating spt: %v", err)
-		}
-		spToken, err := adal.NewServicePrincipalTokenFromCertificate(*oauthConfig, c.ClientID, certificate, rsaPrivateKey, c.ResourceURI)
-		if err != nil {
-			return nil, fmt.Errorf("failed to get oauth token from certificate auth: %v", err)
-		}
-		if err := spToken.Refresh(); err != nil {
-			return nil, fmt.Errorf("failed to refresh token: %v", err)
-		}
-		return spToken, nil
-	}
-
-	// 3. By default return MSI
-	msiEndpoint, err := adal.GetMSIVMEndpoint()
-	if err != nil {
-		return nil, err
-	}
-	spToken, err := c.getMSIToken(msiEndpoint)
-	if err != nil {
-		return nil, fmt.Errorf("failed to get oauth token from MSI: %v", err)
-	}
-	if err := spToken.Refresh(); err != nil {
-		return nil, fmt.Errorf("failed to refersh token: %v", spToken)
-	}
-	return spToken, nil
-}
-
-func (c *TokenProviderConfiguration) getMSIToken(msiEndpoint string) (*adal.ServicePrincipalToken, error) {
-	if c.ClientID == "" {
-		return adal.NewServicePrincipalTokenFromMSI(msiEndpoint, c.ResourceURI)
-	}
-	return adal.NewServicePrincipalTokenFromMSIWithUserAssignedID(msiEndpoint, c.ResourceURI, c.ClientID)
-}
-
-// GetToken gets a CBS JWT
-func (t *TokenProvider) GetToken(audience string) (*auth.Token, error) {
-	token := t.tokenProvider.Token()
-	expireTicks, err := strconv.ParseInt(string(token.ExpiresOn), 10, 64)
-	if err != nil {
-		return nil, err
-	}
-	expires := time.Unix(expireTicks, 0)
-
-	if expires.Before(time.Now()) {
-		if err := t.tokenProvider.Refresh(); err != nil {
-			return nil, err
-		}
-		token = t.tokenProvider.Token()
-	}
-
-	return auth.NewToken(auth.CBSTokenTypeJWT, token.AccessToken, string(token.ExpiresOn)), nil
-}
-
-func decodePkcs12(pkcs []byte, password string) (*x509.Certificate, *rsa.PrivateKey, error) {
-	privateKey, certificate, err := pkcs12.Decode(pkcs, password)
-	if err != nil {
-		return nil, nil, err
-	}
-
-	rsaPrivateKey, isRsaKey := privateKey.(*rsa.PrivateKey)
-	if !isRsaKey {
-		return nil, nil, fmt.Errorf("PKCS#12 certificate must contain an RSA private key")
-	}
-
-	return certificate, rsaPrivateKey, nil
-}
-
-func azureEnvFromEnvironment() (*azure.Environment, error) {
-	envName := os.Getenv("AZURE_ENVIRONMENT")
-
-	var env azure.Environment
-	if envName == "" {
-		env = azure.PublicCloud
-	} else {
-		var err error
-		env, err = azure.EnvironmentFromName(envName)
-		if err != nil {
-			return nil, err
-		}
-	}
-	return &env, nil
-}
diff --git a/sdk/messaging/internal/ci.messaging.yml b/sdk/messaging/internal/ci.messaging.yml
deleted file mode 100644
index 1b5b106687..0000000000
--- a/sdk/messaging/internal/ci.messaging.yml
+++ /dev/null
@@ -1,28 +0,0 @@
-# NOTE: Please refer to https://aka.ms/azsdk/engsys/ci-yaml before editing this file.
-trigger:
-  branches:
-    include:
-      - main
-      - feature/*
-      - hotfix/*
-      - release/*
-  paths:
-    include:
-    - sdk/messaging/internal/
-
-pr:
-  branches:
-    include:
-      - main
-      - feature/*
-      - hotfix/*
-      - release/*
-  paths:
-    include:
-    - sdk/messaging/internal/
-
-stages:
-- template: /eng/pipelines/templates/jobs/archetype-sdk-client.yml
-  parameters:
-    ServiceDirectory: 'messaging/internal'
-    LicenseCheck: false
diff --git a/sdk/messaging/internal/doc.go b/sdk/messaging/internal/doc.go
deleted file mode 100644
index adc8088ca6..0000000000
--- a/sdk/messaging/internal/doc.go
+++ /dev/null
@@ -1,10 +0,0 @@
-//go:build go1.16
-// +build go1.16
-
-// Copyright (c) Microsoft Corporation. All rights reserved.
-// Licensed under the MIT License.
-
-/*
-internal contains shared code for messaging packages.
-*/
-package internal
diff --git a/sdk/messaging/internal/go.mod b/sdk/messaging/internal/go.mod
deleted file mode 100644
index c631281fb6..0000000000
--- a/sdk/messaging/internal/go.mod
+++ /dev/null
@@ -1,14 +0,0 @@
-module github.com/Azure/azure-sdk-for-go/sdk/messaging/internal
-
-go 1.16
-
-require (
-	github.com/Azure/azure-sdk-for-go/sdk/internal v1.0.0
-	github.com/Azure/go-amqp v0.17.0
-	github.com/Azure/go-autorest/autorest v0.11.22
-	github.com/Azure/go-autorest/autorest/adal v0.9.17
-	github.com/davecgh/go-spew v1.1.1 // indirect
-	github.com/devigned/tab v0.1.1
-	github.com/stretchr/testify v1.7.0
-	golang.org/x/crypto v0.0.0-20210921155107-089bfa567519
-)
diff --git a/sdk/messaging/internal/go.sum b/sdk/messaging/internal/go.sum
deleted file mode 100644
index c7babcaf84..0000000000
--- a/sdk/messaging/internal/go.sum
+++ /dev/null
@@ -1,64 +0,0 @@
-github.com/Azure/azure-sdk-for-go/sdk/internal v1.0.0 h1:jp0dGvZ7ZK0mgqnTSClMxa5xuRL7NZgHameVYF6BurY=
-github.com/Azure/azure-sdk-for-go/sdk/internal v1.0.0/go.mod h1:eWRD7oawr1Mu1sLCawqVc0CUiF43ia3qQMxLscsKQ9w=
-github.com/Azure/go-amqp v0.17.0 h1:HHXa3149nKrI0IZwyM7DRcRy5810t9ZICDutn4BYzj4=
-github.com/Azure/go-amqp v0.17.0/go.mod h1:9YJ3RhxRT1gquYnzpZO1vcYMMpAdJT+QEg6fwmw9Zlg=
-github.com/Azure/go-autorest v14.2.0+incompatible h1:V5VMDjClD3GiElqLWO7mz2MxNAK/vTfRHdAubSIPRgs=
-github.com/Azure/go-autorest v14.2.0+incompatible/go.mod h1:r+4oMnoxhatjLLJ6zxSWATqVooLgysK6ZNox3g/xq24=
-github.com/Azure/go-autorest/autorest v0.11.22 h1:bXiQwDjrRmBQOE67bwlvUKAC1EU1yZTPQ38c+bstZws=
-github.com/Azure/go-autorest/autorest v0.11.22/go.mod h1:BAWYUWGPEtKPzjVkp0Q6an0MJcJDsoh5Z1BFAEFs4Xs=
-github.com/Azure/go-autorest/autorest/adal v0.9.14/go.mod h1:W/MM4U6nLxnIskrw4UwWzlHfGjwUS50aOsc/I3yuU8M=
-github.com/Azure/go-autorest/autorest/adal v0.9.17 h1:esOPl2dhcz9P3jqBSJ8tPGEj2EqzPPT6zfyuloiogKY=
-github.com/Azure/go-autorest/autorest/adal v0.9.17/go.mod h1:XVVeme+LZwABT8K5Lc3hA4nAe8LDBVle26gTrguhhPQ=
-github.com/Azure/go-autorest/autorest/date v0.3.0 h1:7gUk1U5M/CQbp9WoqinNzJar+8KY+LPI6wiWrP/myHw=
-github.com/Azure/go-autorest/autorest/date v0.3.0/go.mod h1:BI0uouVdmngYNUzGWeSYnokU+TrmwEsOqdt8Y6sso74=
-github.com/Azure/go-autorest/autorest/mocks v0.4.1 h1:K0laFcLE6VLTOwNgSxaGbUcLPuGXlNkbVvq4cW4nIHk=
-github.com/Azure/go-autorest/autorest/mocks v0.4.1/go.mod h1:LTp+uSrOhSkaKrUy935gNZuuIPPVsHlr9DSOxSayd+k=
-github.com/Azure/go-autorest/logger v0.2.1 h1:IG7i4p/mDa2Ce4TRyAO8IHnVhAVF3RFU+ZtXWSmf4Tg=
-github.com/Azure/go-autorest/logger v0.2.1/go.mod h1:T9E3cAhj2VqvPOtCYAvby9aBXkZmbF5NWuPV8+WeEW8=
-github.com/Azure/go-autorest/tracing v0.6.0 h1:TYi4+3m5t6K48TGI9AUdb+IzbnSxvnvUMfuitfgcfuo=
-github.com/Azure/go-autorest/tracing v0.6.0/go.mod h1:+vhtPC754Xsa23ID7GlGsrdKBpUA79WCAKPPZVC2DeU=
-github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
-github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
-github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
-github.com/devigned/tab v0.1.1 h1:3mD6Kb1mUOYeLpJvTVSDwSg5ZsfSxfvxGRTxRsJsITA=
-github.com/devigned/tab v0.1.1/go.mod h1:XG9mPq0dFghrYvoBF3xdRrJzSTX1b7IQrvaL9mzjeJY=
-github.com/dnaeon/go-vcr v1.1.0/go.mod h1:M7tiix8f0r6mKKJ3Yq/kqU1OYf3MnfmBWVbPx/yU9ko=
-github.com/form3tech-oss/jwt-go v3.2.2+incompatible/go.mod h1:pbq4aXjuKjdthFRnoDwaVPLA+WlJuPGy+QneDUgJi2k=
-github.com/fortytw2/leaktest v1.3.0 h1:u8491cBMTQ8ft8aeV+adlcytMZylmA5nnwwkRZjI8vw=
-github.com/fortytw2/leaktest v1.3.0/go.mod h1:jDsjWgpAGjm2CA7WthBh/CdZYEPF31XHquHwclZch5g=
-github.com/golang-jwt/jwt/v4 v4.0.0 h1:RAqyYixv1p7uEnocuy8P1nru5wprCh/MH2BIlW5z5/o=
-github.com/golang-jwt/jwt/v4 v4.0.0/go.mod h1:/xlHOz8bRuivTWchD4jCa+NbatV+wEUSzwAxVc6locg=
-github.com/google/go-cmp v0.5.1 h1:JFrFEBb2xKufg6XkJsJr+WbKb4FQlURi5RUcBveYu9k=
-github.com/google/go-cmp v0.5.1/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE=
-github.com/modocache/gover v0.0.0-20171022184752-b58185e213c5/go.mod h1:caMODM3PzxT8aQXRPkAt8xlV/e7d7w8GM5g0fa5F0D8=
-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/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
-github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UVUgZn+9EI=
-github.com/stretchr/testify v1.7.0 h1:nwc3DEeHmmLAfoZucVR881uASk0Mfjw8xYJ99tb5CcY=
-github.com/stretchr/testify v1.7.0/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
-golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w=
-golang.org/x/crypto v0.0.0-20200622213623-75b288015ac9/go.mod h1:LzIPMQfyMNhhGPhUkYOs5KpL4U8rLKemX1yGLhDgUto=
-golang.org/x/crypto v0.0.0-20201002170205-7f63de1d35b0/go.mod h1:LzIPMQfyMNhhGPhUkYOs5KpL4U8rLKemX1yGLhDgUto=
-golang.org/x/crypto v0.0.0-20210921155107-089bfa567519 h1:7I4JAnoQBe7ZtJcBaYHi5UtiO8tQHbUSXxL+pnGRANg=
-golang.org/x/crypto v0.0.0-20210921155107-089bfa567519/go.mod h1:GvvjBRRGRdwPK5ydBHafDWAxML/pGHZbMvKqRZ5+Abc=
-golang.org/x/net v0.0.0-20190404232315-eb5bcb51f2a3/go.mod h1:t9HGtf8HONx5eT2rtn7q6eTqICYqUVnKs3thJo3Qplg=
-golang.org/x/net v0.0.0-20201010224723-4f7140c49acb/go.mod h1:sp8m0HH+o8qH0wwXwYZr8TS3Oi6o0r6Gce1SSxlDquU=
-golang.org/x/net v0.0.0-20210226172049-e18ecbb05110/go.mod h1:m0MpNAwzfU5UDzcl9v0D8zg8gWTRqZa9RBIspLL5mdg=
-golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
-golang.org/x/sys v0.0.0-20190412213103-97732733099d/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
-golang.org/x/sys v0.0.0-20200930185726-fdedc70b468f/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
-golang.org/x/sys v0.0.0-20201119102817-f84b799fce68/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
-golang.org/x/sys v0.0.0-20210615035016-665e8c7367d1/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
-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/tools v0.0.0-20180917221912-90fa682c2a6e/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ=
-golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543 h1:E7g+9GITq07hpfrRu66IVDexMakfv52eLZ2CXBWiKr4=
-golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
-gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405 h1:yhCVgyC4o1eVCa2tZl7eS0r+SDo693bJlVdllGtEeKM=
-gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
-gopkg.in/yaml.v2 v2.2.1/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=
-gopkg.in/yaml.v2 v2.4.0/go.mod h1:RDklbk79AGWmwhnvt/jBztapEOGDOx6ZbXqjP6csGnQ=
-gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c h1:dUUwHk2QECo/6vqA44rthZ8ie2QXMNeKRTHCNY2nXvo=
-gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
diff --git a/sdk/messaging/internal/ptrs.go b/sdk/messaging/internal/ptrs.go
deleted file mode 100644
index 7dd4509632..0000000000
--- a/sdk/messaging/internal/ptrs.go
+++ /dev/null
@@ -1,24 +0,0 @@
-// Copyright (c) Microsoft Corporation. All rights reserved.
-// Licensed under the MIT License.
-package internal
-
-// PtrBool takes a boolean and returns a pointer to that bool. For use in literal pointers, ptrBool(true) -> *bool
-func PtrBool(toPtr bool) *bool {
-	return &toPtr
-}
-
-// PtrString takes a string and returns a pointer to that string. For use in literal pointers,
-// PtrString(fmt.Sprintf("..", foo)) -> *string
-func PtrString(toPtr string) *string {
-	return &toPtr
-}
-
-// PtrInt32 takes a int32 and returns a pointer to that int32. For use in literal pointers, ptrInt32(1) -> *int32
-func PtrInt32(number int32) *int32 {
-	return &number
-}
-
-// PtrInt64 takes a int64 and returns a pointer to that int64. For use in literal pointers, ptrInt64(1) -> *int64
-func PtrInt64(number int64) *int64 {
-	return &number
-}
diff --git a/sdk/messaging/internal/retry.go b/sdk/messaging/internal/retry.go
deleted file mode 100644
index b925c817f0..0000000000
--- a/sdk/messaging/internal/retry.go
+++ /dev/null
@@ -1,34 +0,0 @@
-// Copyright (c) Microsoft Corporation. All rights reserved.
-// Licensed under the MIT License.
-package internal
-
-import (
-	"time"
-)
-
-// Retryable represents an error which should be able to be retried
-type Retryable string
-
-// Error implementation for Retryable
-func (r Retryable) Error() string {
-	return string(r)
-}
-
-// Retry will attempt to retry an action a number of times if the action returns a retryable error
-func Retry(times int, delay time.Duration, action func() (interface{}, error)) (interface{}, error) {
-	var lastErr error
-	for i := 0; i < times; i++ {
-		item, err := action()
-		if err != nil {
-			if retryable, ok := err.(Retryable); ok {
-				lastErr = retryable
-				time.Sleep(delay)
-				continue
-			} else {
-				return nil, err
-			}
-		}
-		return item, nil
-	}
-	return nil, lastErr
-}
diff --git a/sdk/messaging/internal/rpc/rpc.go b/sdk/messaging/internal/rpc/rpc.go
deleted file mode 100644
index cce32b3714..0000000000
--- a/sdk/messaging/internal/rpc/rpc.go
+++ /dev/null
@@ -1,505 +0,0 @@
-// Package rpc provides functionality for request / reply messaging. It is used by package mgmt and cbs.
-package rpc
-
-//	MIT License
-//
-//	Copyright (c) Microsoft Corporation. All rights reserved.
-//
-//	Permission is hereby granted, free of charge, to any person obtaining a copy
-//	of this software and associated documentation files (the "Software"), to deal
-//	in the Software without restriction, including without limitation the rights
-//	to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
-//	copies of the Software, and to permit persons to whom the Software is
-//	furnished to do so, subject to the following conditions:
-//
-//	The above copyright notice and this permission notice shall be included in all
-//	copies or substantial portions of the Software.
-//
-//	THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
-//	IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
-//	FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
-//	AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
-//	LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
-//	OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
-//	SOFTWARE
-
-import (
-	"context"
-	"errors"
-	"fmt"
-	"strings"
-	"sync"
-	"time"
-
-	"github.com/devigned/tab"
-
-	"github.com/Azure/azure-sdk-for-go/sdk/internal/uuid"
-	"github.com/Azure/azure-sdk-for-go/sdk/messaging/internal"
-	"github.com/Azure/azure-sdk-for-go/sdk/messaging/internal/tracing"
-	"github.com/Azure/go-amqp"
-)
-
-const (
-	replyPostfix           = "-reply-to-"
-	statusCodeKey          = "status-code"
-	descriptionKey         = "status-description"
-	defaultReceiverCredits = 1000
-)
-
-type (
-	// Link is the bidirectional communication structure used for CBS negotiation
-	Link struct {
-		session *amqp.Session
-
-		receiver amqpReceiver // *amqp.Receiver
-		sender   amqpSender   // *amqp.Sender
-
-		clientAddress string
-		sessionID     *string
-		useSessionID  bool
-		id            string
-
-		responseMu              sync.Mutex
-		startResponseRouterOnce *sync.Once
-		responseMap             map[string]chan rpcResponse
-
-		// for unit tests
-		uuidNewV4     func() (uuid.UUID, error)
-		messageAccept func(ctx context.Context, message *amqp.Message) error
-	}
-
-	// Response is the simplified response structure from an RPC like call
-	Response struct {
-		Code        int
-		Description string
-		Message     *amqp.Message
-	}
-
-	// LinkOption provides a way to customize the construction of a Link
-	LinkOption func(link *Link) error
-
-	rpcResponse struct {
-		message *amqp.Message
-		err     error
-	}
-
-	// Actually: *amqp.Receiver
-	amqpReceiver interface {
-		Receive(ctx context.Context) (*amqp.Message, error)
-		Close(ctx context.Context) error
-	}
-
-	amqpSender interface {
-		Send(ctx context.Context, msg *amqp.Message) error
-		Close(ctx context.Context) error
-	}
-)
-
-// LinkWithSessionFilter configures a Link to use a session filter
-func LinkWithSessionFilter(sessionID *string) LinkOption {
-	return func(l *Link) error {
-		l.sessionID = sessionID
-		l.useSessionID = true
-		return nil
-	}
-}
-
-// NewLink will build a new request response link
-func NewLink(conn *amqp.Client, address string, opts ...LinkOption) (*Link, error) {
-	authSession, err := conn.NewSession()
-	if err != nil {
-		return nil, err
-	}
-
-	return NewLinkWithSession(authSession, address, opts...)
-}
-
-// NewLinkWithSession will build a new request response link, but will reuse an existing AMQP session
-func NewLinkWithSession(session *amqp.Session, address string, opts ...LinkOption) (*Link, error) {
-	linkID, err := uuid.New()
-	if err != nil {
-		return nil, err
-	}
-
-	id := linkID.String()
-	link := &Link{
-		session:       session,
-		clientAddress: strings.Replace("$", "", address, -1) + replyPostfix + id,
-		id:            id,
-
-		uuidNewV4:               uuid.New,
-		responseMap:             map[string]chan rpcResponse{},
-		startResponseRouterOnce: &sync.Once{},
-	}
-
-	for _, opt := range opts {
-		if err := opt(link); err != nil {
-			return nil, err
-		}
-	}
-
-	sender, err := session.NewSender(
-		amqp.LinkTargetAddress(address),
-	)
-	if err != nil {
-		return nil, err
-	}
-
-	receiverOpts := []amqp.LinkOption{
-		amqp.LinkSourceAddress(address),
-		amqp.LinkTargetAddress(link.clientAddress),
-		amqp.LinkCredit(defaultReceiverCredits),
-	}
-
-	if link.sessionID != nil {
-		const name = "com.microsoft:session-filter"
-		const code = uint64(0x00000137000000C)
-		if link.sessionID == nil {
-			receiverOpts = append(receiverOpts, amqp.LinkSourceFilter(name, code, nil))
-		} else {
-			receiverOpts = append(receiverOpts, amqp.LinkSourceFilter(name, code, link.sessionID))
-		}
-	}
-
-	receiver, err := session.NewReceiver(receiverOpts...)
-	if err != nil {
-		// make sure we close the sender
-		clsCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
-		defer cancel()
-
-		_ = sender.Close(clsCtx)
-		return nil, err
-	}
-
-	link.sender = sender
-	link.receiver = receiver
-	link.messageAccept = receiver.AcceptMessage
-
-	return link, nil
-}
-
-// RetryableRPC attempts to retry a request a number of times with delay
-func (l *Link) RetryableRPC(ctx context.Context, times int, delay time.Duration, msg *amqp.Message) (*Response, error) {
-	ctx, span := tracing.StartSpanFromContext(ctx, "az-amqp-common.rpc.RetryableRPC")
-	defer span.End()
-
-	res, err := internal.Retry(times, delay, func() (interface{}, error) {
-		ctx, span := tracing.StartSpanFromContext(ctx, "az-amqp-rpc.RetryableRPC.retry")
-		defer span.End()
-
-		res, err := l.RPC(ctx, msg)
-
-		if err != nil {
-			tab.For(ctx).Error(fmt.Errorf("error in RPC via link %s: %v", l.id, err))
-			return nil, err
-		}
-
-		switch {
-		case res.Code >= 200 && res.Code < 300:
-			tab.For(ctx).Debug(fmt.Sprintf("successful rpc on link %s: status code %d and description: %s", l.id, res.Code, res.Description))
-			return res, nil
-		case res.Code >= 500:
-			errMessage := fmt.Sprintf("server error link %s: status code %d and description: %s", l.id, res.Code, res.Description)
-			tab.For(ctx).Error(errors.New(errMessage))
-			return nil, internal.Retryable(errMessage)
-		default:
-			errMessage := fmt.Sprintf("unhandled error link %s: status code %d and description: %s", l.id, res.Code, res.Description)
-			tab.For(ctx).Error(errors.New(errMessage))
-			return nil, internal.Retryable(errMessage)
-		}
-	})
-	if err != nil {
-		tab.For(ctx).Error(err)
-		return nil, err
-	}
-	return res.(*Response), nil
-}
-
-// startResponseRouter is responsible for taking any messages received on the 'response'
-// link and forwarding it to the proper channel. The channel is being select'd by the
-// original `RPC` call.
-func (l *Link) startResponseRouter() {
-	for {
-		res, err := l.receiver.Receive(context.Background())
-
-		// You'll see this when the link is shutting down (either
-		// service-initiated via 'detach' or a user-initiated shutdown)
-		if isClosedError(err) {
-			l.broadcastError(err)
-			break
-		}
-
-		// I don't believe this should happen. The JS version of this same code
-		// ignores errors as well since responses should always be correlated
-		// to actual send requests. So this is just here for completeness.
-		if res == nil {
-			continue
-		}
-
-		autogenMessageId, ok := res.Properties.CorrelationID.(string)
-
-		if !ok {
-			// TODO: it'd be good to track these in some way. We don't have a good way to
-			// forward this on at this point.
-			continue
-		}
-
-		ch := l.deleteChannelFromMap(autogenMessageId)
-
-		if ch != nil {
-			ch <- rpcResponse{message: res, err: err}
-		}
-	}
-}
-
-// RPC sends a request and waits on a response for that request
-func (l *Link) RPC(ctx context.Context, msg *amqp.Message) (*Response, error) {
-	l.startResponseRouterOnce.Do(func() {
-		go l.startResponseRouter()
-	})
-
-	copiedMessage, messageID, err := addMessageID(msg, l.uuidNewV4)
-
-	if err != nil {
-		return nil, err
-	}
-
-	// use the copiedMessage from this point
-	msg = copiedMessage
-
-	const altStatusCodeKey, altDescriptionKey = "statusCode", "statusDescription"
-
-	ctx, span := tracing.StartSpanFromContext(ctx, "az-amqp-common.rpc.RPC")
-	defer span.End()
-
-	msg.Properties.ReplyTo = &l.clientAddress
-
-	if msg.ApplicationProperties == nil {
-		msg.ApplicationProperties = make(map[string]interface{})
-	}
-
-	if _, ok := msg.ApplicationProperties["server-timeout"]; !ok {
-		if deadline, ok := ctx.Deadline(); ok {
-			msg.ApplicationProperties["server-timeout"] = uint(time.Until(deadline) / time.Millisecond)
-		}
-	}
-
-	responseCh := l.addChannelToMap(messageID)
-
-	if responseCh == nil {
-		return nil, amqp.ErrLinkClosed
-	}
-
-	err = l.sender.Send(ctx, msg)
-
-	if err != nil {
-		l.deleteChannelFromMap(messageID)
-		tab.For(ctx).Error(err)
-		return nil, err
-	}
-
-	var res *amqp.Message
-
-	select {
-	case <-ctx.Done():
-		l.deleteChannelFromMap(messageID)
-		res, err = nil, ctx.Err()
-	case resp := <-responseCh:
-		// this will get triggered by the loop in 'startReceiverRouter' when it receives
-		// a message with our autoGenMessageID set in the correlation_id property.
-		res, err = resp.message, resp.err
-	}
-
-	if err != nil {
-		tab.For(ctx).Error(err)
-		return nil, err
-	}
-
-	var statusCode int
-	statusCodeCandidates := []string{statusCodeKey, altStatusCodeKey}
-	for i := range statusCodeCandidates {
-		if rawStatusCode, ok := res.ApplicationProperties[statusCodeCandidates[i]]; ok {
-			if cast, ok := rawStatusCode.(int32); ok {
-				statusCode = int(cast)
-				break
-			} else {
-				err := errors.New("status code was not of expected type int32")
-				tab.For(ctx).Error(err)
-				return nil, err
-			}
-		}
-	}
-	if statusCode == 0 {
-		err := errors.New("status codes was not found on rpc message")
-		tab.For(ctx).Error(err)
-		return nil, err
-	}
-
-	var description string
-	descriptionCandidates := []string{descriptionKey, altDescriptionKey}
-	for i := range descriptionCandidates {
-		if rawDescription, ok := res.ApplicationProperties[descriptionCandidates[i]]; ok {
-			if description, ok = rawDescription.(string); ok || rawDescription == nil {
-				break
-			} else {
-				return nil, errors.New("status description was not of expected type string")
-			}
-		}
-	}
-
-	span.AddAttributes(tab.StringAttribute("http.status_code", fmt.Sprintf("%d", statusCode)))
-
-	response := &Response{
-		Code:        int(statusCode),
-		Description: description,
-		Message:     res,
-	}
-
-	if err := l.messageAccept(ctx, res); err != nil {
-		tab.For(ctx).Error(err)
-		return response, err
-	}
-
-	return response, err
-}
-
-// Close the link receiver, sender and session
-func (l *Link) Close(ctx context.Context) error {
-	ctx, span := tracing.StartSpanFromContext(ctx, "az-amqp-common.rpc.Close")
-	defer span.End()
-
-	if err := l.closeReceiver(ctx); err != nil {
-		_ = l.closeSender(ctx)
-		_ = l.closeSession(ctx)
-		return err
-	}
-
-	if err := l.closeSender(ctx); err != nil {
-		_ = l.closeSession(ctx)
-		return err
-	}
-
-	return l.closeSession(ctx)
-}
-
-func (l *Link) closeReceiver(ctx context.Context) error {
-	ctx, span := tracing.StartSpanFromContext(ctx, "az-amqp-common.rpc.closeReceiver")
-	defer span.End()
-
-	if l.receiver != nil {
-		return l.receiver.Close(ctx)
-	}
-	return nil
-}
-
-func (l *Link) closeSender(ctx context.Context) error {
-	ctx, span := tracing.StartSpanFromContext(ctx, "az-amqp-common.rpc.closeSender")
-	defer span.End()
-
-	if l.sender != nil {
-		return l.sender.Close(ctx)
-	}
-	return nil
-}
-
-func (l *Link) closeSession(ctx context.Context) error {
-	ctx, span := tracing.StartSpanFromContext(ctx, "az-amqp-common.rpc.closeSession")
-	defer span.End()
-
-	if l.session != nil {
-		return l.session.Close(ctx)
-	}
-	return nil
-}
-
-// addChannelToMap adds a channel which will be used by the response router to
-// notify when there is a response to the request.
-// If l.responseMap is nil (for instance, via broadcastError) this function will
-// return nil.
-func (l *Link) addChannelToMap(messageID string) chan rpcResponse {
-	l.responseMu.Lock()
-	defer l.responseMu.Unlock()
-
-	if l.responseMap == nil {
-		return nil
-	}
-
-	responseCh := make(chan rpcResponse, 1)
-	l.responseMap[messageID] = responseCh
-
-	return responseCh
-}
-
-// deleteChannelFromMap removes the message from our internal map and returns
-// a channel that the corresponding RPC() call is waiting on.
-// If l.responseMap is nil (for instance, via broadcastError) this function will
-// return nil.
-func (l *Link) deleteChannelFromMap(messageID string) chan rpcResponse {
-	l.responseMu.Lock()
-	defer l.responseMu.Unlock()
-
-	if l.responseMap == nil {
-		return nil
-	}
-
-	ch := l.responseMap[messageID]
-	delete(l.responseMap, messageID)
-
-	return ch
-}
-
-// broadcastError notifies the anyone waiting for a response that the link/session/connection
-// has closed.
-func (l *Link) broadcastError(err error) {
-	l.responseMu.Lock()
-	defer l.responseMu.Unlock()
-
-	for _, ch := range l.responseMap {
-		ch <- rpcResponse{err: err}
-	}
-
-	l.responseMap = nil
-}
-
-// addMessageID generates a unique UUID for the message. When the service
-// responds it will fill out the correlation ID property of the response
-// with this ID, allowing us to link the request and response together.
-//
-// NOTE: this function copies 'message', adding in a 'Properties' object
-// if it does not already exist.
-func addMessageID(message *amqp.Message, uuidNewV4 func() (uuid.UUID, error)) (*amqp.Message, string, error) {
-	uuid, err := uuidNewV4()
-
-	if err != nil {
-		return nil, "", err
-	}
-
-	autoGenMessageID := uuid.String()
-
-	// we need to modify the message so we'll make a copy
-	copiedMessage := *message
-
-	if message.Properties == nil {
-		copiedMessage.Properties = &amqp.MessageProperties{
-			MessageID: autoGenMessageID,
-		}
-	} else {
-		// properties already exist, make a copy and then update
-		// the message ID
-		copiedProperties := *message.Properties
-		copiedProperties.MessageID = autoGenMessageID
-
-		copiedMessage.Properties = &copiedProperties
-	}
-
-	return &copiedMessage, autoGenMessageID, nil
-}
-
-func isClosedError(err error) bool {
-	var detachError *amqp.DetachError
-
-	return errors.Is(err, amqp.ErrLinkClosed) ||
-		errors.As(err, &detachError) ||
-		errors.Is(err, amqp.ErrConnClosed) ||
-		errors.Is(err, amqp.ErrSessionClosed)
-}
diff --git a/sdk/messaging/internal/tracing/tracing.go b/sdk/messaging/internal/tracing/tracing.go
deleted file mode 100644
index 17b8bd1543..0000000000
--- a/sdk/messaging/internal/tracing/tracing.go
+++ /dev/null
@@ -1,33 +0,0 @@
-// Copyright (c) Microsoft Corporation. All rights reserved.
-// Licensed under the MIT License.
-package tracing
-
-import (
-	"context"
-	"os"
-
-	"github.com/Azure/azure-sdk-for-go/sdk/messaging/internal"
-	"github.com/devigned/tab"
-)
-
-// StartSpanFromContext starts a span given a context and applies common library information
-func StartSpanFromContext(ctx context.Context, operationName string) (context.Context, tab.Spanner) {
-	ctx, span := tab.StartSpan(ctx, operationName)
-	ApplyComponentInfo(span)
-	return ctx, span
-}
-
-// ApplyComponentInfo applies eventhub library and network info to the span
-func ApplyComponentInfo(span tab.Spanner) {
-	span.AddAttributes(
-		tab.StringAttribute("component", "github.com/Azure/azure-amqp-common-go"),
-		tab.StringAttribute("version", internal.Version))
-	applyNetworkInfo(span)
-}
-
-func applyNetworkInfo(span tab.Spanner) {
-	hostname, err := os.Hostname()
-	if err == nil {
-		span.AddAttributes(tab.StringAttribute("peer.hostname", hostname))
-	}
-}
diff --git a/sdk/messaging/internal/version.go b/sdk/messaging/internal/version.go
deleted file mode 100644
index c35df1698c..0000000000
--- a/sdk/messaging/internal/version.go
+++ /dev/null
@@ -1,8 +0,0 @@
-// Copyright (c) Microsoft Corporation. All rights reserved.
-// Licensed under the MIT License.
-package internal
-
-const (
-	// Version is the semantic version of the library
-	Version = "v0.1.2"
-)


Test patch:
diff --git a/eng/common/TestResources/remove-test-resources.yml b/eng/common/TestResources/remove-test-resources.yml
index 6c02706d22..cf5c0c5ac4 100644
--- a/eng/common/TestResources/remove-test-resources.yml
+++ b/eng/common/TestResources/remove-test-resources.yml
@@ -34,5 +34,5 @@ steps:
         -Force `
         -Verbose
     displayName: Remove test resources
-    condition: eq(variables['CI_HAS_DEPLOYED_RESOURCES'], 'true')
+    condition: and(eq(variables['CI_HAS_DEPLOYED_RESOURCES'], 'true'), ne(variables['Skip.RemoveTestResources'], 'true'))
     continueOnError: true
diff --git a/eng/scripts/MgmtTestLib.ps1 b/eng/scripts/MgmtTestLib.ps1
index 5b26b98636..f529c892db 100644
--- a/eng/scripts/MgmtTestLib.ps1
+++ b/eng/scripts/MgmtTestLib.ps1
@@ -19,8 +19,8 @@ function Invoke-MgmtTestgen ()
         [string]$autorestPath = "",
         [string]$config = "autorest.md",
         [string]$autorestVersion = "3.8.2",
-        [string]$goExtension = "@autorest/go@4.0.0-preview.39",
-        [string]$testExtension = "@autorest/gotest@3.1.1",
+        [string]$goExtension = "@autorest/go@4.0.0-preview.41",
+        [string]$testExtension = "@autorest/gotest@4.0.0",
         [string]$outputFolder
     )
     if ($clean)
diff --git a/sdk/data/azcosmos/cosmos_policy_bearer_token_test.go b/sdk/data/azcosmos/cosmos_policy_bearer_token_test.go
index e5d3eecade..fb986ccc1d 100644
--- a/sdk/data/azcosmos/cosmos_policy_bearer_token_test.go
+++ b/sdk/data/azcosmos/cosmos_policy_bearer_token_test.go
@@ -5,54 +5,148 @@ package azcosmos
 
 import (
 	"context"
+	"fmt"
+
+	"errors"
 	"net/http"
 	"testing"
+	"time"
 
+	"github.com/Azure/azure-sdk-for-go/sdk/azcore"
 	"github.com/Azure/azure-sdk-for-go/sdk/azcore/policy"
-	azruntime "github.com/Azure/azure-sdk-for-go/sdk/azcore/runtime"
+	"github.com/Azure/azure-sdk-for-go/sdk/azcore/runtime"
 	"github.com/Azure/azure-sdk-for-go/sdk/internal/mock"
 )
 
-func TestConvertBearerToken(t *testing.T) {
-	srv, close := mock.NewTLSServer()
-	defer close()
-	srv.SetResponse(mock.WithStatusCode(http.StatusOK))
+const (
+	tokenValue                = "***"
+	accessTokenRespSuccess    = `{"access_token": "` + tokenValue + `", "expires_in": 3600}`
+	accessTokenRespShortLived = `{"access_token": "` + tokenValue + `", "expires_in": 0}`
+	scope                     = "scope"
+)
 
-	verifier := bearerTokenVerify{}
-	pl := azruntime.NewPipeline("azcosmostest", "v1.0.0", azruntime.PipelineOptions{PerCall: []policy.Policy{&mockAuthPolicy{}, &cosmosBearerTokenPolicy{}, &verifier}}, &policy.ClientOptions{Transport: srv})
-	req, err := azruntime.NewRequest(context.Background(), http.MethodGet, srv.URL())
-	req.SetOperationValue(pipelineRequestOptions{
-		isWriteOperation: true,
-	})
+type mockCredential struct {
+	getTokenImpl func(ctx context.Context, options policy.TokenRequestOptions) (azcore.AccessToken, error)
+}
 
-	if err != nil {
-		t.Fatalf("unexpected error: %v", err)
+func (mc mockCredential) GetToken(ctx context.Context, options policy.TokenRequestOptions) (azcore.AccessToken, error) {
+	if mc.getTokenImpl != nil {
+		return mc.getTokenImpl(ctx, options)
 	}
+	return azcore.AccessToken{Token: "***", ExpiresOn: time.Now().Add(time.Hour)}, nil
+}
 
-	_, err = pl.Do(req)
-	if err != nil {
-		t.Fatalf("unexpected error: %v", err)
-	}
+func (mc mockCredential) Do(req *policy.Request) (*http.Response, error) {
+	return nil, nil
+}
 
-	if verifier.authHeaderContent != "type=aad&ver=1.0&sig=this is a test token" {
-		t.Fatalf("Expected auth header content to be 'type=aad&ver=1.0&sig=this is a test token', got %s", verifier.authHeaderContent)
+func defaultTestPipeline(srv policy.Transporter, scope string) runtime.Pipeline {
+	retryOpts := policy.RetryOptions{
+		MaxRetryDelay: 500 * time.Millisecond,
+		RetryDelay:    time.Millisecond,
 	}
+	b := newCosmosBearerTokenPolicy(mockCredential{}, []string{scope}, nil)
+	return runtime.NewPipeline(
+		"azcosmostest",
+		"v1.0.0",
+		runtime.PipelineOptions{PerRetry: []policy.Policy{b}},
+		&policy.ClientOptions{Retry: retryOpts, Transport: srv},
+	)
 }
 
-type bearerTokenVerify struct {
-	authHeaderContent string
+func TestBearerPolicy_SuccessGetToken(t *testing.T) {
+	srv, close := mock.NewTLSServer()
+	defer close()
+	srv.AppendResponse(mock.WithBody([]byte(accessTokenRespSuccess)))
+	srv.AppendResponse(mock.WithStatusCode(http.StatusOK))
+	pipeline := defaultTestPipeline(srv, scope)
+	req, err := runtime.NewRequest(context.Background(), http.MethodGet, srv.URL())
+	if err != nil {
+		t.Fatal(err)
+	}
+	resp, err := pipeline.Do(req)
+	if err != nil {
+		t.Fatalf("Expected nil error but received one")
+	}
+	expectedToken := fmt.Sprintf("type=aad&ver=1.0&sig=%v", tokenValue)
+	if token := resp.Request.Header.Get(headerAuthorization); token != expectedToken {
+		t.Fatalf("expected token '%s', got '%s'", expectedToken, token)
+	}
 }
 
-func (p *bearerTokenVerify) Do(req *policy.Request) (*http.Response, error) {
-	p.authHeaderContent = req.Raw().Header.Get(headerAuthorization)
-
-	return req.Next()
+func TestBearerPolicy_CredentialFailGetToken(t *testing.T) {
+	srv, close := mock.NewTLSServer()
+	defer close()
+	expectedErr := errors.New("oops")
+	failCredential := mockCredential{}
+	failCredential.getTokenImpl = func(ctx context.Context, options policy.TokenRequestOptions) (azcore.AccessToken, error) {
+		return azcore.AccessToken{}, expectedErr
+	}
+	b := newCosmosBearerTokenPolicy(failCredential, nil, nil)
+	pipeline := runtime.NewPipeline("azcosmostest", "v1.0.0", runtime.PipelineOptions{}, &policy.ClientOptions{
+		Transport: srv,
+		Retry: policy.RetryOptions{
+			RetryDelay: 10 * time.Millisecond,
+		},
+		PerRetryPolicies: []policy.Policy{b}})
+	req, err := runtime.NewRequest(context.Background(), http.MethodGet, srv.URL())
+	if err != nil {
+		t.Fatal(err)
+	}
+	resp, err := pipeline.Do(req)
+	if err != expectedErr {
+		t.Fatalf("unexpected error: %v", err)
+	}
+	if resp != nil {
+		t.Fatal("expected nil response")
+	}
 }
 
-type mockAuthPolicy struct{}
-
-func (p *mockAuthPolicy) Do(req *policy.Request) (*http.Response, error) {
-	req.Raw().Header.Set(headerAuthorization, "Bearer this is a test token")
+func TestBearerTokenPolicy_TokenExpired(t *testing.T) {
+	srv, close := mock.NewTLSServer()
+	defer close()
+	srv.AppendResponse(mock.WithBody([]byte(accessTokenRespShortLived)))
+	srv.AppendResponse(mock.WithStatusCode(http.StatusOK))
+	pipeline := defaultTestPipeline(srv, scope)
+	req, err := runtime.NewRequest(context.Background(), http.MethodGet, srv.URL())
+	if err != nil {
+		t.Fatal(err)
+	}
+	_, err = pipeline.Do(req)
+	if err != nil {
+		t.Fatalf("unexpected error %v", err)
+	}
+	_, err = pipeline.Do(req)
+	if err != nil {
+		t.Fatalf("unexpected error %v", err)
+	}
+}
 
-	return req.Next()
+func TestBearerPolicy_GetTokenFailsNoDeadlock(t *testing.T) {
+	srv, close := mock.NewTLSServer()
+	defer close()
+	srv.AppendResponse(mock.WithBody([]byte(accessTokenRespSuccess)))
+	retryOpts := policy.RetryOptions{
+		// use a negative try timeout to trigger a deadline exceeded error causing GetToken() to fail
+		TryTimeout:    -1 * time.Nanosecond,
+		MaxRetryDelay: 500 * time.Millisecond,
+		RetryDelay:    50 * time.Millisecond,
+		MaxRetries:    3,
+	}
+	b := newCosmosBearerTokenPolicy(mockCredential{}, nil, nil)
+	pipeline := runtime.NewPipeline("azcosmostest", "v1.0.0", runtime.PipelineOptions{}, &policy.ClientOptions{
+		Transport:        srv,
+		Retry:            retryOpts,
+		PerRetryPolicies: []policy.Policy{b}})
+	req, err := runtime.NewRequest(context.Background(), http.MethodGet, srv.URL())
+	if err != nil {
+		t.Fatal(err)
+	}
+	resp, err := pipeline.Do(req)
+	if err == nil {
+		t.Fatal("unexpected nil error")
+	}
+	if resp != nil {
+		t.Fatal("expected nil response")
+	}
 }
diff --git a/sdk/messaging/azservicebus/admin/admin_client_test.go b/sdk/messaging/azservicebus/admin/admin_client_test.go
index cd5210614b..18f8201b96 100644
--- a/sdk/messaging/azservicebus/admin/admin_client_test.go
+++ b/sdk/messaging/azservicebus/admin/admin_client_test.go
@@ -22,10 +22,10 @@ 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/azservicebus/internal/atom"
+	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azservicebus/internal/auth"
+	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azservicebus/internal/conn"
 	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azservicebus/internal/test"
 	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azservicebus/internal/utils"
-	"github.com/Azure/azure-sdk-for-go/sdk/messaging/internal/auth"
-	"github.com/Azure/azure-sdk-for-go/sdk/messaging/internal/conn"
 	"github.com/stretchr/testify/require"
 )
 
diff --git a/sdk/messaging/azservicebus/admin_client_test.go b/sdk/messaging/azservicebus/admin_client_test.go
index 56e3947dc2..6c11f3d068 100644
--- a/sdk/messaging/azservicebus/admin_client_test.go
+++ b/sdk/messaging/azservicebus/admin_client_test.go
@@ -11,8 +11,8 @@ import (
 
 	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azservicebus/admin"
 	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azservicebus/internal/atom"
+	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azservicebus/internal/conn"
 	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azservicebus/internal/test"
-	"github.com/Azure/azure-sdk-for-go/sdk/messaging/internal/conn"
 	"github.com/stretchr/testify/require"
 )
 
diff --git a/sdk/messaging/azservicebus/client_test.go b/sdk/messaging/azservicebus/client_test.go
index 7628314f53..c6f26055fe 100644
--- a/sdk/messaging/azservicebus/client_test.go
+++ b/sdk/messaging/azservicebus/client_test.go
@@ -17,8 +17,8 @@ import (
 	"github.com/Azure/azure-sdk-for-go/sdk/azidentity"
 	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azservicebus/admin"
 	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azservicebus/internal"
+	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azservicebus/internal/sas"
 	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azservicebus/internal/test"
-	"github.com/Azure/azure-sdk-for-go/sdk/messaging/internal/sas"
 	"github.com/stretchr/testify/require"
 	"nhooyr.io/websocket"
 )
diff --git a/sdk/messaging/azservicebus/internal/namespace_test.go b/sdk/messaging/azservicebus/internal/namespace_test.go
index ceaeb1e6d6..0f069dc598 100644
--- a/sdk/messaging/azservicebus/internal/namespace_test.go
+++ b/sdk/messaging/azservicebus/internal/namespace_test.go
@@ -11,9 +11,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/azservicebus/internal/auth"
 	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azservicebus/internal/sbauth"
 	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azservicebus/internal/test"
-	"github.com/Azure/azure-sdk-for-go/sdk/messaging/internal/auth"
 	"github.com/Azure/go-amqp"
 	"github.com/stretchr/testify/require"
 )
diff --git a/sdk/messaging/internal/sas/sas_test.go b/sdk/messaging/azservicebus/internal/sas/sas_test.go
similarity index 97%
rename from sdk/messaging/internal/sas/sas_test.go
rename to sdk/messaging/azservicebus/internal/sas/sas_test.go
index ff0a47912c..e3b0b72a99 100644
--- a/sdk/messaging/internal/sas/sas_test.go
+++ b/sdk/messaging/azservicebus/internal/sas/sas_test.go
@@ -10,7 +10,7 @@ import (
 	"testing"
 	"time"
 
-	"github.com/Azure/azure-sdk-for-go/sdk/messaging/internal/auth"
+	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azservicebus/internal/auth"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 )
diff --git a/sdk/messaging/azservicebus/internal/stress/tests/constant_detachment.go b/sdk/messaging/azservicebus/internal/stress/tests/constant_detachment.go
index 8c1749b07f..b57045e6d3 100644
--- a/sdk/messaging/azservicebus/internal/stress/tests/constant_detachment.go
+++ b/sdk/messaging/azservicebus/internal/stress/tests/constant_detachment.go
@@ -7,6 +7,8 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"log"
+	"sync"
 	"time"
 
 	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azservicebus"
@@ -30,19 +32,34 @@ func ConstantDetachment(remainingArgs []string) {
 	sender, err := client.NewSender(queueName, nil)
 	sc.PanicOnError("create a sender", err)
 
-	const maxMessages = 10000
+	// this number isn't too special, but it gives us long enough so that
+	// we are guaranteed that these detaches will interfere with our receiving.
+	// Bug filed for this particular test, pertaining to settlement counting:
+	//   https://github.com/Azure/azure-sdk-for-go/issues/17945)
+	const maxMessages = 20000
 
 	shared.MustGenerateMessages(sc, sender, maxMessages, 1024, stats)
 
+	// We'll give it a little time for the messages to show up in the runtime properties.
+	// Just a simple sanity check that we don't have a bug in our message generator and that
+	// they all exist on the remote entity.
+	time.Sleep(10 * time.Second)
+
+	adminClient, err := admin.NewClientFromConnectionString(sc.ConnectionString, nil)
+	sc.PanicOnError("creating admin client to check properties", err)
+
+	queueProps, err := adminClient.GetQueueRuntimeProperties(sc.Context, queueName, nil)
+	sc.PanicOnError("failed to get queue runtime properties", err)
+
+	if queueProps.TotalMessageCount != maxMessages {
+		sc.PanicOnError("", fmt.Errorf("incorrect number of messages (expected: %d, actual:%d)", maxMessages, queueProps.TotalMessageCount))
+	}
+
 	// now attempt to receive messages, while constant detaching is happening
 	receiver, err := client.NewReceiverForQueue(queueName, &azservicebus.ReceiverOptions{
-		ReceiveMode: azservicebus.ReceiveModeReceiveAndDelete,
+		ReceiveMode: azservicebus.ReceiveModePeekLock,
 	})
 	sc.PanicOnError("failed to create receiver", err)
-
-	adminClient, err := admin.NewClientFromConnectionString(sc.ConnectionString, nil)
-	sc.PanicOnError("failed to create admin client", err)
-
 	go func() {
 		// keep updating the definition of the queue, which will cause the service to detach us.
 		err := shared.ConstantlyUpdateQueue(sc.Context, adminClient, queueName, 30*time.Second)
@@ -55,14 +72,19 @@ func ConstantDetachment(remainingArgs []string) {
 	}()
 
 InfiniteLoop:
-	for stats.Received != maxMessages {
+	for stats.Completed != maxMessages {
 		select {
 		case <-sc.Done():
 			break InfiniteLoop
 		default:
 		}
 
-		messages, err := receiver.ReceiveMessages(sc.Context, 10, nil)
+		messages, err := receiver.ReceiveMessages(sc.Context, 1000, nil)
+
+		if errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded) {
+			log.Printf("Application is done, cancelling...")
+			break InfiniteLoop
+		}
 
 		if err != nil {
 			sc.LogIfFailed("receive failed, continuing", err, stats)
@@ -70,9 +92,29 @@ InfiniteLoop:
 		}
 
 		stats.AddReceived(int32(len(messages)))
+
+		wg := sync.WaitGroup{}
+		wg.Add(len(messages))
+
+		for _, m := range messages {
+			go func(m *azservicebus.ReceivedMessage) {
+				defer wg.Done()
+
+				if err := receiver.CompleteMessage(sc.Context, m, nil); err != nil {
+					sc.LogIfFailed("Failed completing message", err, stats)
+					return
+				}
+
+				stats.AddCompleted(int32(1))
+			}(m)
+		}
+
+		wg.Wait()
 	}
 
-	if stats.Received != maxMessages {
+	if stats.Completed != maxMessages {
 		sc.PanicOnError("constantdetach failed", fmt.Errorf("not all messages received (got %d, wanted %d)", stats.Received, maxMessages))
+	} else {
+		log.Printf("All messages received")
 	}
 }
diff --git a/sdk/messaging/azservicebus/internal/stress/tests/constant_detachment_sender.go b/sdk/messaging/azservicebus/internal/stress/tests/constant_detachment_sender.go
index 8083976f15..95de045f35 100644
--- a/sdk/messaging/azservicebus/internal/stress/tests/constant_detachment_sender.go
+++ b/sdk/messaging/azservicebus/internal/stress/tests/constant_detachment_sender.go
@@ -109,6 +109,12 @@ func checkMessages(sc *shared.StressContext, queueName string, numSent int, stat
 		defer cancel()
 
 		messages, err := receiver.ReceiveMessages(ctx, numSent, nil)
+
+		if errors.Is(err, context.DeadlineExceeded) || errors.Is(err, context.Canceled) {
+			// test is over or we received nothing (ie, we're probably done)
+			break
+		}
+
 		sc.PanicOnError("failed to receive messages", err)
 
 		if len(messages) == 0 {
diff --git a/sdk/messaging/azservicebus/internal/stress/tests/finite_send_and_receive.go b/sdk/messaging/azservicebus/internal/stress/tests/finite_send_and_receive.go
index 9625db3d2d..2b19ca8124 100644
--- a/sdk/messaging/azservicebus/internal/stress/tests/finite_send_and_receive.go
+++ b/sdk/messaging/azservicebus/internal/stress/tests/finite_send_and_receive.go
@@ -39,23 +39,22 @@ func FiniteSendAndReceiveTest(remainingArgs []string) {
 	sender, err := client.NewSender(queueName, nil)
 	sc.PanicOnError("failed to create sender", err)
 
-	const messageLimit = 500
+	// 50000 isn't particularly special, but it does give us a decent # of receives
+	// so we get a decent view into our performance.
+	const messageLimit = 50000
 
 	log.Printf("Sending %d messages (all messages will be sent before receiving begins)", messageLimit)
-	shared.MustGenerateMessages(sc, sender, messageLimit, 100, sc.NewStat("sender"))
+	stats := sc.NewStat("finite")
+	shared.MustGenerateMessages(sc, sender, messageLimit, 100, stats)
 
 	log.Printf("Starting receiving...")
 
 	receiver, err := client.NewReceiverForQueue(queueName, nil)
 	sc.PanicOnError("Failed to create receiver", err)
 
-	completions := make(chan struct{}, 100)
-
-	receiverStats := sc.NewStat("receiver")
-
-	for receiverStats.Received < messageLimit {
+	for stats.Received < messageLimit {
 		log.Printf("[start] Receiving messages...")
-		messages, err := receiver.ReceiveMessages(context.Background(), 100, nil)
+		messages, err := receiver.ReceiveMessages(context.Background(), 1000, nil)
 		log.Printf("[done] Receiving messages... %v, %v", len(messages), err)
 		sc.PanicOnError("failed to create client", err)
 
@@ -68,26 +67,22 @@ func FiniteSendAndReceiveTest(remainingArgs []string) {
 			wg.Add(1)
 
 			go func(msg *azservicebus.ReceivedMessage) {
-				completions <- struct{}{}
 				defer wg.Done()
-				defer func() { <-completions }()
 
 				ctx, cancel := context.WithTimeout(context.Background(), 2*time.Minute)
 				defer cancel()
 
 				err := receiver.CompleteMessage(ctx, msg, nil)
 
-				var rpcCodeErr interface{ RPCCode() int }
+				var sbErr *azservicebus.Error
 
-				if errors.As(err, &rpcCodeErr) {
-					if rpcCodeErr.RPCCode() == 410 {
-						receiverStats.AddError("lock lost", err)
-						return
-					}
+				if errors.As(err, &sbErr) && sbErr.Code == azservicebus.CodeLockLost {
+					stats.AddError("lock lost", err)
+					return
 				}
 
 				sc.PanicOnError("failed to complete message", err)
-				receiverStats.AddReceived(1)
+				stats.AddReceived(1)
 			}(msg)
 		}
 
diff --git a/sdk/messaging/azservicebus/internal/stress/tests/long_running_renew_lock.go b/sdk/messaging/azservicebus/internal/stress/tests/long_running_renew_lock.go
index 3f1089bbc8..18fdbea633 100644
--- a/sdk/messaging/azservicebus/internal/stress/tests/long_running_renew_lock.go
+++ b/sdk/messaging/azservicebus/internal/stress/tests/long_running_renew_lock.go
@@ -5,6 +5,7 @@ package tests
 
 import (
 	"context"
+	"errors"
 	"fmt"
 	"log"
 	"time"
@@ -50,7 +51,7 @@ func LongRunningRenewLockTest(remainingArgs []string) {
 		i++
 		err := receiver.RenewMessageLock(ctx, messages[0], nil)
 
-		if err == context.Canceled || err == context.DeadlineExceeded {
+		if errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded) {
 			log.Printf("Cancellation/deadline exceeded. Can stop, will complete message now")
 
 			err = receiver.CompleteMessage(context.Background(), messages[0], nil)
diff --git a/sdk/messaging/azservicebus/internal/stress/tests/mostly_idle_receiver.go b/sdk/messaging/azservicebus/internal/stress/tests/mostly_idle_receiver.go
index b6bd3791e1..e730ef984c 100644
--- a/sdk/messaging/azservicebus/internal/stress/tests/mostly_idle_receiver.go
+++ b/sdk/messaging/azservicebus/internal/stress/tests/mostly_idle_receiver.go
@@ -32,6 +32,10 @@ func MostlyIdleReceiver(remainingArgs []string) {
 		2 * time.Hour,
 		3 * time.Hour,
 		24 * time.Hour,
+		36 * time.Hour,
+		2 * 24 * time.Hour,
+		3 * 24 * time.Hour,
+		4 * 24 * time.Hour,
 	}
 
 	wg := sync.WaitGroup{}
diff --git a/sdk/messaging/azservicebus/internal/stress/tests/receive_cancellation.go b/sdk/messaging/azservicebus/internal/stress/tests/receive_cancellation.go
index 2670ba6ca9..d1a152f2b9 100644
--- a/sdk/messaging/azservicebus/internal/stress/tests/receive_cancellation.go
+++ b/sdk/messaging/azservicebus/internal/stress/tests/receive_cancellation.go
@@ -5,6 +5,7 @@ package tests
 
 import (
 	"context"
+	"errors"
 	"fmt"
 	"time"
 
@@ -34,14 +35,20 @@ func ReceiveCancellation(remainingArgs []string) {
 
 			// from a cold receiver link
 			_, err = receiver.ReceiveMessages(ctx, 95, nil)
-			sc.PanicOnError("failed to receive messages (1)", err)
+
+			if !errors.Is(err, context.Canceled) && !errors.Is(err, context.DeadlineExceeded) {
+				sc.PanicOnError("failed to receive messages (1)", err)
+			}
 
 			ctx, cancel = context.WithTimeout(context.Background(), time.Duration(i)*time.Millisecond)
 			defer cancel()
 
 			// and one more time, now that the link has been warmed up
 			_, err = receiver.ReceiveMessages(ctx, 95, nil)
-			sc.PanicOnError("failed to receive messages (2)", err)
+
+			if !errors.Is(err, context.Canceled) && !errors.Is(err, context.DeadlineExceeded) {
+				sc.PanicOnError("failed to receive messages (2)", err)
+			}
 		}()
 	}
 }
diff --git a/sdk/messaging/azservicebus/internal/stress/tests/send_and_receive drain.go b/sdk/messaging/azservicebus/internal/stress/tests/send_and_receive drain.go
new file mode 100644
index 0000000000..97626dcb45
--- /dev/null
+++ b/sdk/messaging/azservicebus/internal/stress/tests/send_and_receive drain.go
@@ -0,0 +1,97 @@
+// Copyright (c) Microsoft Corporation. All rights reserved.
+// Licensed under the MIT License.
+
+package tests
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"log"
+	"strings"
+	"time"
+
+	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azservicebus"
+	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azservicebus/admin"
+	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azservicebus/internal/stress/shared"
+)
+
+func SendAndReceiveDrain(remainingArgs []string) {
+	sc := shared.MustCreateStressContext("SendAndReceiveDrainTest")
+
+	sc.TrackEvent("Start")
+	defer sc.End()
+
+	queueName := strings.ToLower(fmt.Sprintf("queue-%X", time.Now().UnixNano()))
+
+	log.Printf("Creating queue")
+
+	// set a long lock duration to make it obvious when a message is being lost in our
+	// internal buffer or somewhere along the way.
+	// This mimics the scenario mentioned in this issue filed by a customer:
+	// https://github.com/Azure/azure-sdk-for-go/issues/17853
+	lockDuration := "PT5M"
+
+	shared.MustCreateAutoDeletingQueue(sc, queueName, &admin.QueueProperties{
+		LockDuration: &lockDuration,
+	})
+
+	client, err := azservicebus.NewClientFromConnectionString(sc.ConnectionString, nil)
+	sc.PanicOnError("failed to create client", err)
+
+	sender, err := client.NewSender(queueName, nil)
+	sc.PanicOnError("failed to create sender", err)
+
+	receiver, err := client.NewReceiverForQueue(queueName, nil)
+	sc.PanicOnError("Failed to create receiver", err)
+
+	for i := 0; i < 1000; i++ {
+		log.Printf("=====> Round [%d] <====", i)
+
+		const numToSend = 2000
+		const bodyLen = 4096
+		shared.MustGenerateMessages(sc, sender, numToSend, bodyLen, nil)
+
+		totalCompleted := 0
+
+		for totalCompleted < numToSend {
+			log.Printf("Receiving messages [%d/%d]...", totalCompleted, numToSend)
+			ctx, cancel := context.WithTimeout(sc.Context, time.Minute)
+			defer cancel()
+
+			messages, err := receiver.ReceiveMessages(ctx, numToSend+100, nil)
+
+			if errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded) {
+				// this is bad - it means we didn't get _any_ messages within an entire
+				// minute and might indicate that we're hitting the customer bug.
+
+				log.Printf("Exceeded the timeout, trying one more time real fast")
+
+				// let's see if there is some other momentary issue happening here by doing a quick receive again.
+				ctx, cancel := context.WithTimeout(sc.Context, time.Minute)
+				defer cancel()
+				messages, err = receiver.ReceiveMessages(ctx, numToSend+100, nil)
+				sc.PanicOnError("Exceeded a minute while waiting for messages", err)
+			}
+
+			log.Printf("Got %d messages, completing...", len(messages))
+
+			for _, m := range messages {
+				if len(m.Body) != bodyLen {
+					sc.PanicOnError("Body length issue", fmt.Errorf("Invalid body length - expected %d, got %d", bodyLen, len(m.Body)))
+				}
+
+				if err := receiver.CompleteMessage(sc.Context, m, nil); err != nil {
+					sc.PanicOnError("Failed to complete message", err)
+				}
+				totalCompleted++
+			}
+
+			if err != nil {
+				panic(err)
+			}
+		}
+
+		log.Printf("[end] Receiving messages (all received)")
+	}
+}
diff --git a/sdk/messaging/azservicebus/internal/stress/tests/tests.go b/sdk/messaging/azservicebus/internal/stress/tests/tests.go
index b2cd7e6e32..f35bbc171c 100644
--- a/sdk/messaging/azservicebus/internal/stress/tests/tests.go
+++ b/sdk/messaging/azservicebus/internal/stress/tests/tests.go
@@ -42,6 +42,7 @@ func Run(remainingArgs []string) {
 		"receiveCancellation":      ReceiveCancellation,
 		"idleFastReconnect":        IdleFastReconnect,
 		"mostlyIdleReceiver":       MostlyIdleReceiver,
+		"sendAndReceiveDrain":      SendAndReceiveDrain,
 	}
 
 	if len(remainingArgs) == 0 {
diff --git a/sdk/messaging/internal/rpc/rpc_test.go b/sdk/messaging/internal/rpc/rpc_test.go
deleted file mode 100644
index 4ee848dcd2..0000000000
--- a/sdk/messaging/internal/rpc/rpc_test.go
+++ /dev/null
@@ -1,331 +0,0 @@
-// Copyright (c) Microsoft Corporation. All rights reserved.
-// Licensed under the MIT License.
-package rpc
-
-import (
-	"context"
-	"errors"
-	"fmt"
-	"sync"
-	"testing"
-	"time"
-
-	"github.com/Azure/azure-sdk-for-go/sdk/internal/uuid"
-	"github.com/Azure/go-amqp"
-	"github.com/stretchr/testify/require"
-)
-
-func TestResponseRouterBasic(t *testing.T) {
-	receiver := &fakeReceiver{
-		Responses: []rpcResponse{
-			{amqpMessageWithCorrelationId("my message id"), nil},
-			{nil, amqp.ErrLinkClosed},
-		},
-	}
-
-	link := &Link{
-		responseMap: map[string]chan rpcResponse{
-			"my message id": make(chan rpcResponse, 1),
-		},
-		receiver: receiver,
-	}
-
-	ch := link.responseMap["my message id"]
-
-	link.startResponseRouter()
-	result := <-ch
-	require.EqualValues(t, result.message.Data[0], []byte("ID was my message id"))
-	require.Empty(t, receiver.Responses)
-	require.Nil(t, link.responseMap, "Response map is nil'd out after we get a closed error")
-}
-
-func TestResponseRouterMissingMessageID(t *testing.T) {
-	receiver := &fakeReceiver{
-		Responses: []rpcResponse{
-			{amqpMessageWithCorrelationId("my message id"), nil},
-			{nil, amqp.ErrLinkClosed},
-		},
-	}
-
-	link := &Link{
-		responseMap: map[string]chan rpcResponse{},
-		receiver:    receiver,
-	}
-
-	link.startResponseRouter()
-	require.Empty(t, receiver.Responses)
-}
-
-func TestResponseRouterBadCorrelationID(t *testing.T) {
-	messageWithBadCorrelationID := &amqp.Message{
-		Properties: &amqp.MessageProperties{
-			CorrelationID: uint64(1),
-		},
-	}
-
-	receiver := &fakeReceiver{
-		Responses: []rpcResponse{
-			{messageWithBadCorrelationID, nil},
-			{nil, amqp.ErrLinkClosed},
-		},
-	}
-
-	link := &Link{
-		responseMap: map[string]chan rpcResponse{},
-		receiver:    receiver,
-	}
-
-	link.startResponseRouter()
-	require.Empty(t, receiver.Responses)
-}
-
-func TestResponseRouterFatalErrors(t *testing.T) {
-	fatalErrors := []error{
-		amqp.ErrLinkClosed,
-		&amqp.DetachError{},
-		amqp.ErrConnClosed,
-		amqp.ErrSessionClosed,
-	}
-
-	for _, fatalError := range fatalErrors {
-		t.Run(fatalError.Error(), func(t *testing.T) {
-			receiver := &fakeReceiver{
-				Responses: []rpcResponse{
-					{nil, fatalError},
-				},
-			}
-			sentinelCh := make(chan rpcResponse, 1)
-
-			link := &Link{
-				responseMap: map[string]chan rpcResponse{
-					"sentinel": sentinelCh,
-				},
-				receiver: receiver,
-			}
-
-			link.startResponseRouter()
-			require.Empty(t, receiver.Responses)
-
-			// also, we should have broadcasted that the link is closed to anyone else
-			// that had not yet received a response but was still waiting.
-			select {
-			case rpcResponse := <-sentinelCh:
-				require.Error(t, rpcResponse.err, fatalError.Error())
-				require.Nil(t, rpcResponse.message)
-			case <-time.After(time.Second * 5):
-				require.Fail(t, "sentinel channel should have received a message")
-			}
-		})
-	}
-}
-
-func TestResponseRouterNoResponse(t *testing.T) {
-	receiver := &fakeReceiver{
-		Responses: []rpcResponse{
-			{nil, errors.New("Some other error that will get ignored since we can't route it to anyone (ie: no message ID)")},
-			{nil, amqp.ErrConnClosed},
-		},
-	}
-
-	link := &Link{
-		responseMap: map[string]chan rpcResponse{},
-		receiver:    receiver,
-	}
-
-	link.startResponseRouter()
-	require.Empty(t, receiver.Responses)
-}
-
-func TestAddMessageID(t *testing.T) {
-	message, id, err := addMessageID(&amqp.Message{}, uuid.New)
-	require.NoError(t, err)
-	require.NotEmpty(t, id)
-	require.EqualValues(t, message.Properties.MessageID, id)
-
-	m := &amqp.Message{
-		Data: [][]byte{[]byte("hello world")},
-		Properties: &amqp.MessageProperties{
-			UserID:    []byte("my user ID"),
-			MessageID: "is that will not be copied"},
-	}
-	message, id, err = addMessageID(m, uuid.New)
-	require.NoError(t, err)
-	require.NotEmpty(t, id)
-	require.EqualValues(t, message.Properties.MessageID, id)
-	require.EqualValues(t, message.Properties.UserID, []byte("my user ID"))
-	require.EqualValues(t, message.Data[0], []byte("hello world"))
-}
-
-func TestRPCBasic(t *testing.T) {
-	fakeUUID := uuid.UUID([16]byte{0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15})
-	replyMessage := amqpMessageWithCorrelationId(fakeUUID.String())
-	replyMessage.ApplicationProperties = map[string]interface{}{
-		"status-code": int32(200),
-	}
-
-	ch := make(chan struct{})
-	sender := &fakeSender{ch: ch}
-	receiver := &fakeReceiver{
-		Responses: []rpcResponse{
-			{replyMessage, nil},
-			{nil, amqp.ErrConnClosed},
-		},
-		ch: ch,
-	}
-
-	l := &Link{
-		receiver:                receiver,
-		sender:                  sender,
-		startResponseRouterOnce: &sync.Once{},
-		responseMap:             map[string]chan rpcResponse{},
-
-		uuidNewV4: func() (uuid.UUID, error) {
-			return fakeUUID, nil
-		},
-		messageAccept: func(ctx context.Context, message *amqp.Message) error {
-			return nil
-		},
-	}
-
-	messageToSend := &amqp.Message{}
-	resp, err := l.RPC(context.Background(), messageToSend)
-	require.NoError(t, err)
-
-	require.EqualValues(t, fakeUUID.String(), sender.Sent[0].Properties.MessageID, "Sent message contains a uniquely generated ID")
-	require.EqualValues(t, fakeUUID.String(), resp.Message.Properties.CorrelationID, "Correlation ID matches our originally sent message")
-
-	require.Nil(t, replyMessage.Properties.MessageID, "Original message not modified")
-}
-
-func TestRPCFailedSend(t *testing.T) {
-	// important bit is that we clean up the channel we stored in the map
-	// since we're no longer waiting for the response.
-	fakeUUID := uuid.UUID([16]byte{0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15})
-	replyMessage := amqpMessageWithCorrelationId(fakeUUID.String())
-	replyMessage.ApplicationProperties = map[string]interface{}{
-		"status-code": int32(200),
-	}
-
-	ch := make(chan struct{})
-
-	sender := &fakeSender{
-		ch: ch,
-	}
-	receiver := &fakeReceiver{
-		ch: ch,
-		Responses: []rpcResponse{
-			{nil, amqp.ErrConnClosed},
-		},
-	}
-
-	l := &Link{
-		receiver:                receiver,
-		sender:                  sender,
-		startResponseRouterOnce: &sync.Once{},
-		responseMap:             map[string]chan rpcResponse{},
-
-		uuidNewV4: func() (uuid.UUID, error) {
-			return fakeUUID, nil
-		},
-		messageAccept: func(ctx context.Context, message *amqp.Message) error {
-			panic("Should not be called")
-		},
-	}
-
-	messageToSend := &amqp.Message{}
-	cancelledContext, cancel := context.WithCancel(context.Background())
-	cancel()
-
-	resp, err := l.RPC(cancelledContext, messageToSend)
-	require.Nil(t, resp)
-	require.EqualError(t, err, context.Canceled.Error())
-
-	require.EqualValues(t, fakeUUID.String(), sender.Sent[0].Properties.MessageID, "Sent message contains a uniquely generated ID")
-}
-
-func TestRPCNilMessageMap(t *testing.T) {
-	fakeSender := &fakeSender{}
-	fakeReceiver := &fakeReceiver{
-		Responses: []rpcResponse{
-			// this should let us see what deleteChannelFromMap does
-			{amqpMessageWithCorrelationId("hello"), nil},
-			{nil, amqp.ErrLinkClosed},
-		},
-	}
-
-	link := &Link{
-		sender:   fakeSender,
-		receiver: fakeReceiver,
-		// responseMap is nil if the broadcastError() function is called. Since this can be
-		// at any time our individual map functions need to handle the map not being
-		// there.
-		responseMap:             nil,
-		startResponseRouterOnce: &sync.Once{},
-		uuidNewV4:               uuid.New,
-	}
-
-	// sanity check - all the map/channel functions are returning nil
-	require.Nil(t, link.addChannelToMap("hello"))
-	require.Nil(t, link.deleteChannelFromMap("hello"))
-
-	link.startResponseRouter()
-
-	require.Empty(t, fakeReceiver.Responses, "All responses are used")
-
-	// we're not testing the responseRouter for this second part, so just short-circuit
-	// the running.
-	link.startResponseRouterOnce.Do(func() {})
-
-	// now check that sending can handle it.
-	resp, err := link.RPC(context.Background(), &amqp.Message{})
-	require.Error(t, err, amqp.ErrLinkClosed.Error())
-	require.Nil(t, resp)
-}
-
-func amqpMessageWithCorrelationId(id string) *amqp.Message {
-	return &amqp.Message{
-		Data: [][]byte{[]byte(fmt.Sprintf("ID was %s", id))},
-		Properties: &amqp.MessageProperties{
-			CorrelationID: id,
-		},
-	}
-}
-
-type fakeReceiver struct {
-	Responses []rpcResponse
-	ch        <-chan struct{}
-}
-
-func (fr *fakeReceiver) Receive(ctx context.Context) (*amqp.Message, error) {
-	// wait until the actual send if we're simulating request/response
-	if fr.ch != nil {
-		<-fr.ch
-	}
-
-	resp := fr.Responses[0]
-	fr.Responses = fr.Responses[1:]
-	return resp.message, resp.err
-}
-
-func (fr *fakeReceiver) Close(ctx context.Context) error {
-	panic("Not used for this test")
-}
-
-type fakeSender struct {
-	Sent []*amqp.Message
-	ch   chan<- struct{}
-}
-
-func (s *fakeSender) Send(ctx context.Context, msg *amqp.Message) error {
-	s.Sent = append(s.Sent, msg)
-
-	if s.ch != nil {
-		close(s.ch)
-	}
-
-	return nil
-}
-
-func (fs *fakeSender) Close(ctx context.Context) error {
-	panic("Not used for this test")
-}