Repository: Azure/azure-sdk-for-go
Pull request #17795 (2022-05-03 21:25:03)

Cosmos DB - Adds TransactionalBatch support
## APIView

https://apiview.dev/Assemblies/Review/7e272b28ee30452aaa7e3e77ad8cd4b9/23b1fc488ec84e9e855cce2bb2ab94d8?diffRevisionId=6b6ad7e3a9b74575915fb585bfc99929&doc=False&diffOnly=True

## Description

The exposed new types are:

* TransactionalBatch - Represents the batch itself
* TransactionalBatchOptions - Options for execution of the batch
* TransactionalBatchItemOptions - Options at the individual operation level
* TransactionalBatchResponse - The batch response
* TransactionalBatchResult - The individual operation result within the batch response

### Creating a Transactional Batch

```
func (c *ContainerClient) NewTransactionalBatch(partitionKey PartitionKey) TransactionalBatch
```

On the Container, we can create the TransactionalBatch like so:

```
partitionKey := NewPartitionKeyString("myPartitionKeyValue")
batch := container.NewTransactionalBatch(partitionKey)
```

The batch is scoped to a Partition Key Value.

### Populating operations

A batch contains a list of operations to be executed as a single unit:

```
// CreateItem adds a create operation to the batch.
func (b *TransactionalBatch) CreateItem(item []byte, o *TransactionalBatchItemOptions)

// DeleteItem adds a delete operation to the batch.
func (b *TransactionalBatch) DeleteItem(itemID string, o *TransactionalBatchItemOptions)

// ReplaceItem adds a replace operation to the batch.
func (b *TransactionalBatch) ReplaceItem(itemID string, item []byte, o *TransactionalBatchItemOptions)

// UpsertItem adds an upsert operation to the batch.
func (b *TransactionalBatch) UpsertItem(item []byte, o *TransactionalBatchItemOptions)

// ReadItem adds a read operation to the batch.
func (b *TransactionalBatch) ReadItem(itemID string, o *TransactionalBatchItemOptions) 
```

Each operation that can be added contains some potential options:

```
type TransactionalBatchItemOptions struct {
	// IfMatchETag is used to ensure optimistic concurrency control.
	// https://docs.microsoft.com/azure/cosmos-db/sql/database-transactions-optimistic-concurrency#optimistic-concurrency-control
	IfMatchETag *azcore.ETag
}
```

### Executing a batch

After a batch is created, operations need to be added, and then the ExecuteTransactionalBatch method is used:

```
func (c *ContainerClient) ExecuteTransactionalBatch(ctx context.Context, b TransactionalBatch, o *TransactionalBatchOptions) (TransactionalBatchResponse, error)
```

Example:

```
partitionKey := NewPartitionKeyString("myPartitionKeyValue")
batch := container.NewTransactionalBatch(partitionKey)
batch.CreateItem(bodyOfAnItemToCreate, nil)
batch.ReadItem("idOfAnItemToRead", nil)
//... other operations

batchResponse, err := container.ExecuteTransactionalBatch(context, batch, nil)
```

The ExecuteTransactionalBatch method contains options for the execution:

```
// TransactionalBatchOptions includes options for transactional batch operations.
type TransactionalBatchOptions struct {
	SessionToken string
	ConsistencyLevel *ConsistencyLevel
	EnableContentResponseOnWrite bool
}
```

### Handling batch response

The batch response contains the raw HTTP response plus Request Charge, SessionToken, ActivityId and 2 important properties:

```
// QueryItemsResponse contains response from the query operation.
type TransactionalBatchResponse struct {
	Response
	// SessionToken contains the value from the session token header to be used on session consistency.
	SessionToken string
	// OperationResults contains the individual batch operation results.
	// The order of the results is the same as the order of the operations in the batch.
	OperationResults []TransactionalBatchResponseOperationResult
	// Success indicates if the transaction was successfully committed.
	// If false, one of the operations in the batch failed.
	// Inspect the OperationResults, any operation with status code http.StatusFailedDependency is a dependency failure.
	// The cause of the batch failure is the first operation with status code different from http.StatusFailedDependency.
	Success bool
}
```

`OperationResults` contain all the individual operation results, each operation in a batch has its own response.

Each operation result exposes:

```
// TransactionalBatchResponseOperationResult represents the result of a single operation in a batch.
type TransactionalBatchResult struct {
	// StatusCode contains the status code of the operation.
	StatusCode int32
	// RequestCharge contains the request charge for the operation.
	RequestCharge float32
	// ResourceBody contains the body response of the operation.
	// This property is available depending on the EnableContentResponseOnWrite option.
	ResourceBody []byte
	// ETag contains the ETag of the operation.
	ETag azcore.ETag
}
```

`Success` is used to determine if the batch committed all operations or if it had to be rolled back due to one operation failing.

After receiving the response, we need to inspect if the batch committed, and if not, which operation was the source of the failure:

```
if batchResponse.Success {
		// Transaction succeeded
		// We can inspect the individual operation results 
		for index, operation := range batchResponse.OperationResults {
			fmt.Printf("Operation %v completed with status code %v consumed %v RU", index, operation.StatusCode, operation.RequestCharge)
			if index == 1 {
				// Read operation would have body available
				var itemResponseBody map[string]string
				err = json.Unmarshal(operation.ResourceBody, &itemResponseBody)
				if err != nil {
					panic(err)
				}
			}
		}
	} else {
		// Transaction failed, look for the offending operation
		for index, operation := range batchResponse.OperationResults {
			if operation.StatusCode != http.StatusFailedDependency {
				fmt.Printf("Transaction failed due to operation %v which failed with status code %v", index, operation.StatusCode)
			}
		}
	}
```


Closes https://github.com/Azure/azure-sdk-for-go/issues/17771


Patch:
diff --git a/sdk/data/azcosmos/CHANGELOG.md b/sdk/data/azcosmos/CHANGELOG.md
index 3af5d5d7ad..f3ad8a6de3 100644
--- a/sdk/data/azcosmos/CHANGELOG.md
+++ b/sdk/data/azcosmos/CHANGELOG.md
@@ -4,6 +4,8 @@
 
 ### Features Added
 
+* Added Transactional Batch support
+
 ### Breaking Changes
 
 ### Bugs Fixed
diff --git a/sdk/data/azcosmos/README.md b/sdk/data/azcosmos/README.md
index 50792db7b4..1663009c05 100644
--- a/sdk/data/azcosmos/README.md
+++ b/sdk/data/azcosmos/README.md
@@ -8,7 +8,7 @@ This client library enables client applications to connect to Azure Cosmos via t
 
 ### Prerequisites
 
-* Go versions 1.16 or higher
+* Go versions 1.18 or higher
 * An Azure subscription or free Azure Cosmos DB trial account
 
 Note: If you don't have an Azure subscription, create a free account before you begin.
diff --git a/sdk/data/azcosmos/cosmos_client.go b/sdk/data/azcosmos/cosmos_client.go
index b1d62a73cb..200097565b 100644
--- a/sdk/data/azcosmos/cosmos_client.go
+++ b/sdk/data/azcosmos/cosmos_client.go
@@ -238,6 +238,26 @@ func (c *Client) sendDeleteRequest(
 	return c.executeAndEnsureSuccessResponse(req)
 }
 
+func (c *Client) sendBatchRequest(
+	ctx context.Context,
+	path string,
+	batch []batchOperation,
+	operationContext pipelineRequestOptions,
+	requestOptions cosmosRequestOptions,
+	requestEnricher func(*policy.Request)) (*http.Response, error) {
+	req, err := c.createRequest(path, ctx, http.MethodPost, operationContext, requestOptions, requestEnricher)
+	if err != nil {
+		return nil, err
+	}
+
+	err = c.attachContent(batch, req)
+	if err != nil {
+		return nil, err
+	}
+
+	return c.executeAndEnsureSuccessResponse(req)
+}
+
 func (c *Client) createRequest(
 	path string,
 	ctx context.Context,
@@ -288,7 +308,6 @@ func (c *Client) attachContent(content interface{}, req *policy.Request) error {
 	default:
 		// Otherwise, we need to marshal it
 		err = azruntime.MarshalAsJSON(req, content)
-
 	}
 
 	if err != nil {
diff --git a/sdk/data/azcosmos/cosmos_container.go b/sdk/data/azcosmos/cosmos_container.go
index ad70384311..e9a594cbb0 100644
--- a/sdk/data/azcosmos/cosmos_container.go
+++ b/sdk/data/azcosmos/cosmos_container.go
@@ -5,6 +5,7 @@ package azcosmos
 
 import (
 	"context"
+	"errors"
 
 	"github.com/Azure/azure-sdk-for-go/sdk/azcore/policy"
 	"github.com/Azure/azure-sdk-for-go/sdk/azcore/runtime"
@@ -456,6 +457,66 @@ func (c *ContainerClient) NewQueryItemsPager(query string, partitionKey Partitio
 	})
 }
 
+// NewTransactionalBatch creates a batch of operations to be committed as a single unit.
+// See https://docs.microsoft.com/azure/cosmos-db/sql/transactional-batch
+func (c *ContainerClient) NewTransactionalBatch(partitionKey PartitionKey) TransactionalBatch {
+	return TransactionalBatch{partitionKey: partitionKey}
+}
+
+// ExecuteTransactionalBatch executes a transactional batch.
+// Once executed, verify the Success property of the response to determine if the batch was committed
+func (c *ContainerClient) ExecuteTransactionalBatch(ctx context.Context, b TransactionalBatch, o *TransactionalBatchOptions) (TransactionalBatchResponse, error) {
+	if len(b.operations) == 0 {
+		return TransactionalBatchResponse{}, errors.New("no operations in batch")
+	}
+
+	h := headerOptionsOverride{
+		partitionKey: &b.partitionKey,
+	}
+
+	if o == nil {
+		o = &TransactionalBatchOptions{}
+	} else {
+		h.enableContentResponseOnWrite = &o.EnableContentResponseOnWrite
+	}
+
+	// If contentResponseOnWrite is not enabled at the client level the
+	// service will not even send a batch response payload
+	// Instead we should automatically enforce contentResponseOnWrite for all
+	// batch requests whenever at least one of the item operations requires a content response (read operation)
+	enableContentResponseOnWriteForReadOperations := true
+	for _, op := range b.operations {
+		if op.getOperationType() == operationTypeRead {
+			h.enableContentResponseOnWrite = &enableContentResponseOnWriteForReadOperations
+			break
+		}
+	}
+
+	operationContext := pipelineRequestOptions{
+		resourceType:          resourceTypeDocument,
+		resourceAddress:       c.link,
+		isWriteOperation:      true,
+		headerOptionsOverride: &h}
+
+	path, err := generatePathForNameBased(resourceTypeDocument, operationContext.resourceAddress, true)
+	if err != nil {
+		return TransactionalBatchResponse{}, err
+	}
+
+	azResponse, err := c.database.client.sendBatchRequest(
+		ctx,
+		path,
+		b.operations,
+		operationContext,
+		o,
+		nil)
+	if err != nil {
+		return TransactionalBatchResponse{}, err
+	}
+
+	return newTransactionalBatchResponse(azResponse)
+}
+
 func (c *ContainerClient) getRID(ctx context.Context) (string, error) {
 	containerResponse, err := c.Read(ctx, nil)
 	if err != nil {
diff --git a/sdk/data/azcosmos/cosmos_headers.go b/sdk/data/azcosmos/cosmos_headers.go
index c0c7faebc4..d054bd652b 100644
--- a/sdk/data/azcosmos/cosmos_headers.go
+++ b/sdk/data/azcosmos/cosmos_headers.go
@@ -30,6 +30,9 @@ const (
 	cosmosHeaderQueryMetrics                       string = "x-ms-documentdb-query-metrics"
 	cosmosHeaderIndexUtilization                   string = "x-ms-cosmos-index-utilization"
 	cosmosHeaderCorrelatedActivityId               string = "x-ms-cosmos-correlated-activityid"
+	cosmosHeaderIsBatchRequest                     string = "x-ms-cosmos-is-batch-request"
+	cosmosHeaderIsBatchAtomic                      string = "x-ms-cosmos-batch-atomic"
+	cosmosHeaderIsBatchOrdered                     string = "x-ms-cosmos-batch-ordered"
 	headerXmsDate                                  string = "x-ms-date"
 	headerAuthorization                            string = "Authorization"
 	headerContentType                              string = "Content-Type"
diff --git a/sdk/data/azcosmos/cosmos_transactional_batch.go b/sdk/data/azcosmos/cosmos_transactional_batch.go
new file mode 100644
index 0000000000..167fb55890
--- /dev/null
+++ b/sdk/data/azcosmos/cosmos_transactional_batch.go
@@ -0,0 +1,201 @@
+// Copyright (c) Microsoft Corporation. All rights reserved.
+// Licensed under the MIT License.
+
+package azcosmos
+
+import (
+	"bytes"
+	"encoding/json"
+	"fmt"
+
+	"github.com/Azure/azure-sdk-for-go/sdk/azcore"
+)
+
+// TransactionalBatch is a batch of operations to be executed in a single transaction.
+// See https://docs.microsoft.com/azure/cosmos-db/sql/transactional-batch
+type TransactionalBatch struct {
+	partitionKey PartitionKey
+	operations   []batchOperation
+}
+
+// CreateItem adds a create operation to the batch.
+func (b *TransactionalBatch) CreateItem(item []byte, o *TransactionalBatchItemOptions) {
+	b.operations = append(b.operations,
+		batchOperationCreate{
+			operationType: "Create",
+			resourceBody:  item})
+}
+
+// DeleteItem adds a delete operation to the batch.
+func (b *TransactionalBatch) DeleteItem(itemID string, o *TransactionalBatchItemOptions) {
+	if o == nil {
+		o = &TransactionalBatchItemOptions{}
+	}
+	b.operations = append(b.operations,
+		batchOperationDelete{
+			operationType: "Delete",
+			id:            itemID,
+			ifMatch:       o.IfMatchETag})
+}
+
+// ReplaceItem adds a replace operation to the batch.
+func (b *TransactionalBatch) ReplaceItem(itemID string, item []byte, o *TransactionalBatchItemOptions) {
+	if o == nil {
+		o = &TransactionalBatchItemOptions{}
+	}
+	b.operations = append(b.operations,
+		batchOperationReplace{
+			operationType: "Replace",
+			id:            itemID,
+			resourceBody:  item,
+			ifMatch:       o.IfMatchETag})
+}
+
+// UpsertItem adds an upsert operation to the batch.
+func (b *TransactionalBatch) UpsertItem(item []byte, o *TransactionalBatchItemOptions) {
+	if o == nil {
+		o = &TransactionalBatchItemOptions{}
+	}
+	b.operations = append(b.operations,
+		batchOperationUpsert{
+			operationType: "Upsert",
+			resourceBody:  item,
+			ifMatch:       o.IfMatchETag})
+}
+
+// ReadItem adds a read operation to the batch.
+func (b *TransactionalBatch) ReadItem(itemID string, o *TransactionalBatchItemOptions) {
+	b.operations = append(b.operations,
+		batchOperationRead{
+			operationType: "Read",
+			id:            itemID})
+}
+
+type batchOperation interface {
+	getOperationType() operationType
+}
+
+type batchOperationCreate struct {
+	operationType string
+	resourceBody  []byte
+}
+
+func (b batchOperationCreate) getOperationType() operationType {
+	return operationTypeCreate
+}
+
+// MarshalJSON implements the json.Marshaler interface
+func (b batchOperationCreate) MarshalJSON() ([]byte, error) {
+	buffer := bytes.NewBufferString("{")
+	buffer.WriteString(fmt.Sprintf("\"operationType\":\"%s\"", b.operationType))
+	buffer.WriteString(",\"resourceBody\":")
+	buffer.Write(b.resourceBody)
+	buffer.WriteString("}")
+	return buffer.Bytes(), nil
+}
+
+type batchOperationDelete struct {
+	operationType string
+	ifMatch       *azcore.ETag
+	id            string
+}
+
+func (b batchOperationDelete) getOperationType() operationType {
+	return operationTypeDelete
+}
+
+// MarshalJSON implements the json.Marshaler interface
+func (b batchOperationDelete) MarshalJSON() ([]byte, error) {
+	buffer := bytes.NewBufferString("{")
+	buffer.WriteString(fmt.Sprintf("\"operationType\":\"%s\"", b.operationType))
+	buffer.WriteString(fmt.Sprintf(",\"id\":\"%s\"", b.id))
+	if b.ifMatch != nil {
+		buffer.WriteString(",\"ifMatch\":")
+		etag, err := json.Marshal(b.ifMatch)
+		if err != nil {
+			return nil, err
+		}
+		buffer.Write(etag)
+	}
+
+	buffer.WriteString("}")
+	return buffer.Bytes(), nil
+}
+
+type batchOperationReplace struct {
+	operationType string
+	ifMatch       *azcore.ETag
+	id            string
+	resourceBody  []byte
+}
+
+func (b batchOperationReplace) getOperationType() operationType {
+	return operationTypeReplace
+}
+
+// MarshalJSON implements the json.Marshaler interface
+func (b batchOperationReplace) MarshalJSON() ([]byte, error) {
+	buffer := bytes.NewBufferString("{")
+	buffer.WriteString(fmt.Sprintf("\"operationType\":\"%s\"", b.operationType))
+	if b.ifMatch != nil {
+		buffer.WriteString(",\"ifMatch\":")
+		etag, err := json.Marshal(b.ifMatch)
+		if err != nil {
+			return nil, err
+		}
+		buffer.Write(etag)
+	}
+
+	buffer.WriteString(fmt.Sprintf(",\"id\":\"%s\"", b.id))
+	buffer.WriteString(",\"resourceBody\":")
+	buffer.Write(b.resourceBody)
+	buffer.WriteString("}")
+	return buffer.Bytes(), nil
+}
+
+type batchOperationUpsert struct {
+	operationType string
+	ifMatch       *azcore.ETag
+	resourceBody  []byte
+}
+
+func (b batchOperationUpsert) getOperationType() operationType {
+	return operationTypeUpsert
+}
+
+// MarshalJSON implements the json.Marshaler interface
+func (b batchOperationUpsert) MarshalJSON() ([]byte, error) {
+	buffer := bytes.NewBufferString("{")
+	buffer.WriteString(fmt.Sprintf("\"operationType\":\"%s\"", b.operationType))
+	if b.ifMatch != nil {
+		buffer.WriteString(",\"ifMatch\":")
+		etag, err := json.Marshal(b.ifMatch)
+		if err != nil {
+			return nil, err
+		}
+		buffer.Write(etag)
+	}
+
+	buffer.WriteString(",\"resourceBody\":")
+	buffer.Write(b.resourceBody)
+	buffer.WriteString("}")
+	return buffer.Bytes(), nil
+}
+
+type batchOperationRead struct {
+	operationType string
+	id            string
+}
+
+func (b batchOperationRead) getOperationType() operationType {
+	return operationTypeRead
+}
+
+// MarshalJSON implements the json.Marshaler interface
+func (b batchOperationRead) MarshalJSON() ([]byte, error) {
+	buffer := bytes.NewBufferString("{")
+	buffer.WriteString(fmt.Sprintf("\"operationType\":\"%s\"", b.operationType))
+	buffer.WriteString(fmt.Sprintf(",\"id\":\"%s\"", b.id))
+	buffer.WriteString("}")
+	return buffer.Bytes(), nil
+}
diff --git a/sdk/data/azcosmos/cosmos_transactional_batch_options.go b/sdk/data/azcosmos/cosmos_transactional_batch_options.go
new file mode 100644
index 0000000000..22ed6ddd57
--- /dev/null
+++ b/sdk/data/azcosmos/cosmos_transactional_batch_options.go
@@ -0,0 +1,51 @@
+// Copyright (c) Microsoft Corporation. All rights reserved.
+// Licensed under the MIT License.
+
+package azcosmos
+
+import (
+	"github.com/Azure/azure-sdk-for-go/sdk/azcore"
+)
+
+// TransactionalBatchOptions includes options for transactional batch operations.
+type TransactionalBatchOptions struct {
+	// SessionToken to be used when using Session consistency on the account.
+	// When working with Session consistency, each new write request to Azure Cosmos DB is assigned a new SessionToken.
+	// The client instance will use this token internally with each read/query request to ensure that the set consistency level is maintained.
+	// In some scenarios you need to manage this Session yourself: Consider a web application with multiple nodes, each node will have its own client instance.
+	// If you wanted these nodes to participate in the same session (to be able read your own writes consistently across web tiers),
+	// you would have to send the SessionToken from the response of the write action on one node to the client tier, using a cookie or some other mechanism, and have that token flow back to the web tier for subsequent reads.
+	// If you are using a round-robin load balancer which does not maintain session affinity between requests, such as the Azure Load Balancer,the read could potentially land on a different node to the write request, where the session was created.
+	SessionToken string
+	// ConsistencyLevel overrides the account defined consistency level for this operation.
+	// Consistency can only be relaxed.
+	ConsistencyLevel *ConsistencyLevel
+	// When EnableContentResponseOnWrite is false, the operations in the batch response will have no body, except when they are Read operations.
+	// The default is false.
+	EnableContentResponseOnWrite bool
+}
+
+// TransactionalBatchItemOptions includes options for the specific operation inside a TransactionalBatch
+type TransactionalBatchItemOptions struct {
+	// IfMatchETag is used to ensure optimistic concurrency control.
+	// https://docs.microsoft.com/azure/cosmos-db/sql/database-transactions-optimistic-concurrency#optimistic-concurrency-control
+	IfMatchETag *azcore.ETag
+}
+
+func (options *TransactionalBatchOptions) toHeaders() *map[string]string {
+	headers := make(map[string]string, 2)
+
+	if options.ConsistencyLevel != nil {
+		headers[cosmosHeaderConsistencyLevel] = string(*options.ConsistencyLevel)
+	}
+
+	if options.SessionToken != "" {
+		headers[cosmosHeaderSessionToken] = options.SessionToken
+	}
+
+	headers[cosmosHeaderIsBatchRequest] = "True"
+	headers[cosmosHeaderIsBatchAtomic] = "True"
+	headers[cosmosHeaderIsBatchOrdered] = "True"
+
+	return &headers
+}
diff --git a/sdk/data/azcosmos/cosmos_transactional_batch_response.go b/sdk/data/azcosmos/cosmos_transactional_batch_response.go
new file mode 100644
index 0000000000..09f8ee5790
--- /dev/null
+++ b/sdk/data/azcosmos/cosmos_transactional_batch_response.go
@@ -0,0 +1,89 @@
+// Copyright (c) Microsoft Corporation. All rights reserved.
+// Licensed under the MIT License.
+
+package azcosmos
+
+import (
+	"encoding/json"
+	"net/http"
+
+	"github.com/Azure/azure-sdk-for-go/sdk/azcore"
+	"github.com/Azure/azure-sdk-for-go/sdk/azcore/runtime"
+)
+
+// TransactionalBatchResponse contains response from a transactional batch operation.
+type TransactionalBatchResponse struct {
+	Response
+	// SessionToken contains the value from the session token header to be used on session consistency.
+	SessionToken string
+	// OperationResults contains the individual batch operation results.
+	// The order of the results is the same as the order of the operations in the batch.
+	OperationResults []TransactionalBatchResult
+	// Success indicates if the transaction was successfully committed.
+	// If false, one of the operations in the batch failed.
+	// Inspect the OperationResults, any operation with status code http.StatusFailedDependency is a dependency failure.
+	// The cause of the batch failure is the first operation with status code different from http.StatusFailedDependency.
+	Success bool
+}
+
+func newTransactionalBatchResponse(resp *http.Response) (TransactionalBatchResponse, error) {
+	response := TransactionalBatchResponse{
+		Response: newResponse(resp),
+	}
+
+	response.SessionToken = resp.Header.Get(cosmosHeaderSessionToken)
+
+	response.Success = resp.StatusCode != http.StatusMultiStatus
+
+	if err := runtime.UnmarshalAsJSON(resp, &response.OperationResults); err != nil {
+		return TransactionalBatchResponse{}, err
+	}
+
+	return response, nil
+}
+
+// TransactionalBatchResult represents the result of a single operation in a batch.
+type TransactionalBatchResult struct {
+	// StatusCode contains the status code of the operation.
+	StatusCode int32
+	// RequestCharge contains the request charge for the operation.
+	RequestCharge float32
+	// ResourceBody contains the body response of the operation.
+	// This property is available depending on the EnableContentResponseOnWrite option.
+	ResourceBody []byte
+	// ETag contains the ETag of the operation.
+	ETag azcore.ETag
+}
+
+// UnmarshalJSON implements the json.Unmarshaler interface
+func (or *TransactionalBatchResult) UnmarshalJSON(b []byte) error {
+	var attributes map[string]json.RawMessage
+	err := json.Unmarshal(b, &attributes)
+	if err != nil {
+		return err
+	}
+
+	if statusCode, ok := attributes["statusCode"]; ok {
+		if err := json.Unmarshal(statusCode, &or.StatusCode); err != nil {
+			return err
+		}
+	}
+
+	if requestCharge, ok := attributes["requestCharge"]; ok {
+		if err := json.Unmarshal(requestCharge, &or.RequestCharge); err != nil {
+			return err
+		}
+	}
+
+	if etag, ok := attributes["eTag"]; ok {
+		if err := json.Unmarshal(etag, &or.ETag); err != nil {
+			return err
+		}
+	}
+
+	if body, ok := attributes["resourceBody"]; ok {
+		or.ResourceBody = body
+	}
+
+	return nil
+}
diff --git a/sdk/data/azcosmos/doc.go b/sdk/data/azcosmos/doc.go
index b8689fa183..622300a0d2 100644
--- a/sdk/data/azcosmos/doc.go
+++ b/sdk/data/azcosmos/doc.go
@@ -49,6 +49,7 @@ The following sections provide several code snippets covering some of the most c
 	- Creating a container
 	- Creating, reading, and deleting items
 	- Querying items
+	- Using Transactional Batch
 
 
 Creating a database
@@ -135,5 +136,48 @@ Querying items
 			json.Unmarshal(item, &itemResponseBody)
 		}
 	}
+
+Using Transactional batch
+
+	pk := azcosmos.NewPartitionKeyString("myPartitionKeyValue")
+	batch := container.NewTransactionalBatch(pk)
+
+	item := map[string]string{
+		"id":    "1",
+		"myPartitionKeyProperty": "myPartitionKeyValue",
+		"otherValue": 10
+	}
+	marshalled, err := json.Marshal(item)
+	handle(err)
+
+	batch.CreateItem(marshalled, nil)
+	batch.ReadItem("otherExistingId", nil)
+	batch.DeleteItem("yetAnotherExistingId", nil)
+
+	batchResponse, err  := container.ExecuteTransactionalBatch(context, batch, nil)
+	handle(err)
+
+	if batchResponse.Success {
+		// Transaction succeeded
+		// We can inspect the individual operation results
+		for index, operation := range batchResponse.OperationResults {
+			fmt.Printf("Operation %v completed with status code %v consumed %v RU", index, operation.StatusCode, operation.RequestCharge)
+			if index == 1 {
+				// Read operation would have body available
+				var itemResponseBody map[string]string
+				err = json.Unmarshal(operation.ResourceBody, &itemResponseBody)
+				if err != nil {
+					panic(err)
+				}
+			}
+		}
+	} else {
+		// Transaction failed, look for the offending operation
+		for index, operation := range batchResponse.OperationResults {
+			if operation.StatusCode != http.StatusFailedDependency {
+				fmt.Printf("Transaction failed due to operation %v which failed with status code %v", index, operation.StatusCode)
+			}
+		}
+	}
 */
 package azcosmos
diff --git a/sdk/data/azcosmos/operation_type.go b/sdk/data/azcosmos/operation_type.go
new file mode 100644
index 0000000000..a37b64df99
--- /dev/null
+++ b/sdk/data/azcosmos/operation_type.go
@@ -0,0 +1,15 @@
+// Copyright (c) Microsoft Corporation. All rights reserved.
+// Licensed under the MIT License.
+
+package azcosmos
+
+// operationType defines supported values for operations.
+type operationType int
+
+const (
+	operationTypeCreate  operationType = 0
+	operationTypeRead    operationType = 2
+	operationTypeReplace operationType = 5
+	operationTypeDelete  operationType = 4
+	operationTypeUpsert  operationType = 20
+)
diff --git a/sdk/messaging/azservicebus/CHANGELOG.md b/sdk/messaging/azservicebus/CHANGELOG.md
index 9019bf3acd..be1e6508fa 100644
--- a/sdk/messaging/azservicebus/CHANGELOG.md
+++ b/sdk/messaging/azservicebus/CHANGELOG.md
@@ -21,6 +21,7 @@
 
 - `admin.Client` can now be configured using `azcore.Options`. (#17796)
 - `ReceivedMessage.TransactionPartitionKey` has been removed as this library doesn't support transactions.
+- `ReceivedMessage.Body()` is now a field. `Body` will be nil in the cases where it would have returned an error (where the underlying AMQP message had a payload in .Value, .Sequence or had multiple byte slices in .Data). (#17888)
 
 ### Bugs Fixed
 
diff --git a/sdk/messaging/azservicebus/README.md b/sdk/messaging/azservicebus/README.md
index 3eca9c09b8..bff21758a0 100644
--- a/sdk/messaging/azservicebus/README.md
+++ b/sdk/messaging/azservicebus/README.md
@@ -219,13 +219,30 @@ if err != nil {
 }
 
 for _, message := range messages {
+  // The message body is a []byte. For this example we're just assuming that the body
+  // was a string, converted to bytes but any []byte payload is valid.
+  var body []byte = message.Body
+  fmt.Printf("Message received with body: %s\n", string(body))
+
   // For more information about settling messages:
   // https://docs.microsoft.com/azure/service-bus-messaging/message-transfers-locks-settlement#settling-receive-operations
   err = receiver.CompleteMessage(context.TODO(), message, nil)
 
   if err != nil {
+    var sbErr *azservicebus.Error
+
+    if errors.As(err, &sbErr) && sbErr.Code == azservicebus.CodeLockLost {
+      // The message lock has expired. This isn't fatal for the client, but it does mean
+      // that this message can be received by another Receiver (or potentially this one!).
+      fmt.Printf("Message lock expired\n")
+
+      // You can extend the message lock by calling receiver.RenewMessageLock(msg) before the
+      // message lock has expired.
+      continue
+    }
+
     panic(err)
-  }
+	}
 
   fmt.Printf("Received and completed the message\n")
 }
diff --git a/sdk/messaging/azservicebus/message.go b/sdk/messaging/azservicebus/message.go
index 43a4106e6f..1111d711e8 100644
--- a/sdk/messaging/azservicebus/message.go
+++ b/sdk/messaging/azservicebus/message.go
@@ -4,7 +4,6 @@
 package azservicebus
 
 import (
-	"errors"
 	"fmt"
 	"time"
 
@@ -18,6 +17,9 @@ type ReceivedMessage struct {
 	// ApplicationProperties can be used to store custom metadata for a message.
 	ApplicationProperties map[string]interface{}
 
+	// Body is the payload for a message.
+	Body []byte
+
 	// ContentType describes the payload of the message, with a descriptor following
 	// the format of Content-Type, specified by RFC2045 (ex: "application/json").
 	ContentType *string
@@ -138,17 +140,6 @@ const (
 	MessageStateScheduled MessageState = 2
 )
 
-// Body returns the body for this received message.
-// If the body not compatible with ReceivedMessage this function will return an error.
-func (rm *ReceivedMessage) Body() ([]byte, error) {
-	// TODO: does this come back as a zero length array if the body is empty (which is allowed)
-	if rm.rawAMQPMessage.Data == nil || len(rm.rawAMQPMessage.Data) != 1 {
-		return nil, errors.New("AMQP message Data section is improperly encoded for ReceivedMessage")
-	}
-
-	return rm.rawAMQPMessage.Data[0], nil
-}
-
 // Message is a message with a body and commonly used properties.
 // Properties that are pointers are optional.
 type Message struct {
@@ -323,6 +314,10 @@ func newReceivedMessage(amqpMsg *amqp.Message) *ReceivedMessage {
 		State:          MessageStateActive,
 	}
 
+	if len(msg.rawAMQPMessage.Data) == 1 {
+		msg.Body = msg.rawAMQPMessage.Data[0]
+	}
+
 	if amqpMsg.Properties != nil {
 		if id, ok := amqpMsg.Properties.MessageID.(string); ok {
 			msg.MessageID = id


Test patch:
diff --git a/sdk/data/azcosmos/cosmos_client_test.go b/sdk/data/azcosmos/cosmos_client_test.go
index 6bda337e82..93567db308 100644
--- a/sdk/data/azcosmos/cosmos_client_test.go
+++ b/sdk/data/azcosmos/cosmos_client_test.go
@@ -354,6 +354,50 @@ func TestSendQuery(t *testing.T) {
 	}
 }
 
+func TestSendBatch(t *testing.T) {
+	srv, close := mock.NewTLSServer()
+	defer close()
+	srv.SetResponse(
+		mock.WithStatusCode(200))
+	verifier := pipelineVerifier{}
+	pl := azruntime.NewPipeline("azcosmostest", "v1.0.0", azruntime.PipelineOptions{PerCall: []policy.Policy{&verifier}}, &policy.ClientOptions{Transport: srv})
+	client := &Client{endpoint: srv.URL(), pipeline: pl}
+	operationContext := pipelineRequestOptions{
+		resourceType:    resourceTypeDocument,
+		resourceAddress: "",
+	}
+
+	batch := TransactionalBatch{}
+	batch.partitionKey = NewPartitionKeyString("foo")
+
+	body := map[string]string{
+		"foo": "bar",
+	}
+
+	itemMarshall, _ := json.Marshal(body)
+
+	batch.CreateItem(itemMarshall, nil)
+	batch.ReadItem("someId", nil)
+
+	marshalled, err := json.Marshal(batch.operations)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	_, err = client.sendBatchRequest(context.Background(), "/", batch.operations, operationContext, &TransactionalBatchOptions{}, nil)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	if verifier.requests[0].method != http.MethodPost {
+		t.Errorf("Expected %v, but got %v", http.MethodPost, verifier.requests[0].method)
+	}
+
+	if verifier.requests[0].body != string(marshalled) {
+		t.Errorf("Expected %v, but got %v", string(marshalled), verifier.requests[0].body)
+	}
+}
+
 func TestCreateScopeFromEndpoint(t *testing.T) {
 	url := "https://foo.documents.azure.com:443/"
 	scope, err := createScopeFromEndpoint(url)
diff --git a/sdk/data/azcosmos/cosmos_container_test.go b/sdk/data/azcosmos/cosmos_container_test.go
index 7cb83c431c..f06d0da427 100644
--- a/sdk/data/azcosmos/cosmos_container_test.go
+++ b/sdk/data/azcosmos/cosmos_container_test.go
@@ -520,3 +520,72 @@ func TestContainerQueryItems(t *testing.T) {
 		}
 	}
 }
+
+func TestContainerExecuteBatch(t *testing.T) {
+	batchResponseRaw := []map[string]interface{}{
+		{"statusCode": 200, "requestCharge": 10.0, "eTag": "someETag", "resourceBody": "someBody"},
+		{"statusCode": 201, "requestCharge": 11.0, "eTag": "someETag2"},
+	}
+
+	jsonString, err := json.Marshal(batchResponseRaw)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	srv, close := mock.NewTLSServer()
+	defer close()
+	srv.SetResponse(
+		mock.WithBody(jsonString),
+		mock.WithStatusCode(http.StatusOK),
+		mock.WithHeader(cosmosHeaderEtag, "someEtag"),
+		mock.WithHeader(cosmosHeaderActivityId, "someActivityId"),
+		mock.WithHeader(cosmosHeaderRequestCharge, "13.42"))
+
+	verifier := pipelineVerifier{}
+
+	pl := azruntime.NewPipeline("azcosmostest", "v1.0.0", azruntime.PipelineOptions{PerCall: []policy.Policy{&verifier}}, &policy.ClientOptions{Transport: srv})
+	client := &Client{endpoint: srv.URL(), pipeline: pl}
+
+	database, _ := newDatabase("databaseId", client)
+	container, _ := newContainer("containerId", database)
+
+	pk := NewPartitionKeyString("pk")
+	batch := container.NewTransactionalBatch(pk)
+	_, err = container.ExecuteTransactionalBatch(context.TODO(), batch, nil)
+	if err == nil {
+		t.Fatal("Expected error, but got nil")
+	}
+
+	batch.ReadItem("someId", nil)
+
+	body := map[string]string{
+		"foo": "bar",
+	}
+
+	itemMarshall, _ := json.Marshal(body)
+	batch.CreateItem(itemMarshall, nil)
+
+	_, err = container.ExecuteTransactionalBatch(context.TODO(), batch, nil)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	if len(verifier.requests) != 1 {
+		t.Fatalf("Expected 1 request, got %d", len(verifier.requests))
+	}
+
+	request := verifier.requests[0]
+
+	if request.method != http.MethodPost {
+		t.Errorf("Expected method to be %s, but got %s", http.MethodPost, request.method)
+	}
+
+	if request.url.RequestURI() != "/dbs/databaseId/colls/containerId/docs" {
+		t.Errorf("Expected url to be %s, but got %s", "/dbs/databaseId/colls/containerId/docs", request.url.RequestURI())
+	}
+
+	marshalledOperations, _ := json.Marshal(batch.operations)
+	if request.body != string(marshalledOperations) {
+		t.Errorf("Expected %v, but got %v", string(marshalledOperations), request.body)
+	}
+}
diff --git a/sdk/data/azcosmos/cosmos_transactional_batch_options_test.go b/sdk/data/azcosmos/cosmos_transactional_batch_options_test.go
new file mode 100644
index 0000000000..cac1b51fa1
--- /dev/null
+++ b/sdk/data/azcosmos/cosmos_transactional_batch_options_test.go
@@ -0,0 +1,35 @@
+// Copyright (c) Microsoft Corporation. All rights reserved.
+// Licensed under the MIT License.
+
+package azcosmos
+
+import (
+	"testing"
+)
+
+func TestTransactionalBatchOptionsToHeaders(t *testing.T) {
+	options := &TransactionalBatchOptions{}
+	options.ConsistencyLevel = ConsistencyLevelSession.ToPtr()
+	options.SessionToken = "sessionToken"
+	header := options.toHeaders()
+	if header == nil {
+		t.Fatal("toHeaders should return non-nil")
+	}
+
+	headers := *header
+	if headers[cosmosHeaderConsistencyLevel] != "Session" {
+		t.Errorf("ConsistencyLevel should be Session but got %v", headers[cosmosHeaderConsistencyLevel])
+	}
+	if headers[cosmosHeaderSessionToken] != "sessionToken" {
+		t.Errorf("SessionToken should be sessionToken but got %v", headers[cosmosHeaderSessionToken])
+	}
+	if headers[cosmosHeaderIsBatchAtomic] != "True" {
+		t.Fatal("IsBatchAtomic should be true")
+	}
+	if headers[cosmosHeaderIsBatchRequest] != "True" {
+		t.Fatal("IsBatchRequest should be true")
+	}
+	if headers[cosmosHeaderIsBatchOrdered] != "True" {
+		t.Fatal("IsBatchOrdered should be true")
+	}
+}
diff --git a/sdk/data/azcosmos/cosmos_transactional_batch_response_test.go b/sdk/data/azcosmos/cosmos_transactional_batch_response_test.go
new file mode 100644
index 0000000000..77f89fd8a5
--- /dev/null
+++ b/sdk/data/azcosmos/cosmos_transactional_batch_response_test.go
@@ -0,0 +1,144 @@
+// Copyright (c) Microsoft Corporation. All rights reserved.
+// Licensed under the MIT License.
+
+package azcosmos
+
+import (
+	"context"
+	"encoding/json"
+	"net/http"
+	"testing"
+
+	"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/internal/mock"
+)
+
+func TestTransactionalBatchResponseParsing(t *testing.T) {
+	batchResponseRaw := []map[string]interface{}{
+		{"statusCode": 200, "requestCharge": 10.0, "eTag": "someETag", "resourceBody": "someBody"},
+		{"statusCode": 201, "requestCharge": 11.0, "eTag": "someETag2"},
+	}
+
+	jsonString, err := json.Marshal(batchResponseRaw)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	srv, close := mock.NewTLSServer()
+	defer close()
+	srv.SetResponse(
+		mock.WithBody(jsonString),
+		mock.WithStatusCode(http.StatusOK),
+		mock.WithHeader(cosmosHeaderEtag, "someEtag"),
+		mock.WithHeader(cosmosHeaderActivityId, "someActivityId"),
+		mock.WithHeader(cosmosHeaderRequestCharge, "13.42"))
+
+	req, err := azruntime.NewRequest(context.Background(), http.MethodGet, srv.URL())
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	pl := azruntime.NewPipeline("azcosmostest", "v1.0.0", azruntime.PipelineOptions{}, &policy.ClientOptions{Transport: srv})
+	resp, _ := pl.Do(req)
+	parsedResponse, err := newTransactionalBatchResponse(resp)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	if !parsedResponse.Success {
+		t.Errorf("Expected Success to be true, but got false")
+	}
+
+	if parsedResponse.RawResponse == nil {
+		t.Fatal("parsedResponse.RawResponse is nil")
+	}
+
+	if parsedResponse.ActivityID != "someActivityId" {
+		t.Errorf("Expected ActivityId to be %s, but got %s", "someActivityId", parsedResponse.ActivityID)
+	}
+
+	if parsedResponse.RequestCharge != 13.42 {
+		t.Errorf("Expected RequestCharge to be %f, but got %f", 13.42, parsedResponse.RequestCharge)
+	}
+
+	if parsedResponse.ETag != "someEtag" {
+		t.Errorf("Expected ETag to be %s, but got %s", "someEtag", parsedResponse.ETag)
+	}
+
+	if len(parsedResponse.OperationResults) != 2 {
+		t.Errorf("Expected 2 documents, but got %d", len(parsedResponse.OperationResults))
+	}
+
+	for index, item := range parsedResponse.OperationResults {
+		if index == 0 && item.ETag != "someETag" {
+			t.Errorf("Expected ETag to be %s, but got %s", "someETag", item.ETag)
+		}
+
+		if index == 1 && item.ETag != "someETag2" {
+			t.Errorf("Expected ETag to be %s, but got %s", "someETag2", item.ETag)
+		}
+
+		if index == 0 && item.StatusCode != http.StatusOK {
+			t.Errorf("Expected StatusCode to be %d, but got %d", http.StatusOK, item.StatusCode)
+		}
+
+		if index == 1 && item.StatusCode != http.StatusCreated {
+			t.Errorf("Expected StatusCode to be %d, but got %d", http.StatusCreated, item.StatusCode)
+		}
+
+		if index == 0 && item.RequestCharge != 10 {
+			t.Errorf("Expected RequestCharge to be %f, but got %f", 10.0, item.RequestCharge)
+		}
+
+		if index == 1 && item.RequestCharge != 11 {
+			t.Errorf("Expected RequestCharge to be %f, but got %f", 11.0, item.RequestCharge)
+		}
+
+		if index == 0 && string(item.ResourceBody) != "\"someBody\"" {
+			t.Errorf("Expected ResourceBody to be %s, but got %s", "someBody", item.ResourceBody)
+		}
+
+		if index == 1 && item.ResourceBody != nil {
+			t.Errorf("Expected ResourceBody to be nil, but got %s", item.ResourceBody)
+		}
+
+	}
+}
+
+func TestTransactionalBatchResponseParsing_Failed(t *testing.T) {
+	batchResponseRaw := []map[string]interface{}{
+		{"statusCode": 424},
+		{"statusCode": 409},
+	}
+
+	jsonString, err := json.Marshal(batchResponseRaw)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	srv, close := mock.NewTLSServer()
+	defer close()
+	srv.SetResponse(
+		mock.WithBody(jsonString),
+		mock.WithStatusCode(http.StatusMultiStatus),
+		mock.WithHeader(cosmosHeaderEtag, "someEtag"),
+		mock.WithHeader(cosmosHeaderActivityId, "someActivityId"),
+		mock.WithHeader(cosmosHeaderRequestCharge, "13.42"))
+
+	req, err := azruntime.NewRequest(context.Background(), http.MethodGet, srv.URL())
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	pl := azruntime.NewPipeline("azcosmostest", "v1.0.0", azruntime.PipelineOptions{}, &policy.ClientOptions{Transport: srv})
+	resp, _ := pl.Do(req)
+	parsedResponse, err := newTransactionalBatchResponse(resp)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	if parsedResponse.Success {
+		t.Errorf("Expected Success to be false, but got true")
+	}
+}
diff --git a/sdk/data/azcosmos/cosmos_transactional_batch_test.go b/sdk/data/azcosmos/cosmos_transactional_batch_test.go
new file mode 100644
index 0000000000..84b03b59fa
--- /dev/null
+++ b/sdk/data/azcosmos/cosmos_transactional_batch_test.go
@@ -0,0 +1,176 @@
+// Copyright (c) Microsoft Corporation. All rights reserved.
+// Licensed under the MIT License.
+
+package azcosmos
+
+import (
+	"encoding/json"
+	"testing"
+
+	"github.com/Azure/azure-sdk-for-go/sdk/azcore"
+)
+
+func TestTransactionalBatchCreateItem(t *testing.T) {
+	batch := &TransactionalBatch{}
+	batch.partitionKey = NewPartitionKeyString("foo")
+	body := map[string]string{
+		"foo": "bar",
+	}
+
+	itemMarshall, _ := json.Marshal(body)
+	batch.CreateItem(itemMarshall, nil)
+
+	if len(batch.operations) != 1 {
+		t.Errorf("Expected 1 operation, but got %v", len(batch.operations))
+	}
+
+	if batch.operations[0].getOperationType() != operationTypeCreate {
+		t.Errorf("Expected operation type %v, but got %v", operationTypeCreate, batch.operations[0].getOperationType())
+	}
+
+	asCreate := batch.operations[0].(batchOperationCreate)
+
+	if asCreate.operationType != "Create" {
+		t.Errorf("Expected operation type %v, but got %v", "Create", asCreate.operationType)
+	}
+
+	if string(asCreate.resourceBody) != string(itemMarshall) {
+		t.Errorf("Expected body %v, but got %v", string(itemMarshall), string(asCreate.resourceBody))
+	}
+}
+
+func TestTransactionalBatchReadItem(t *testing.T) {
+	batch := &TransactionalBatch{}
+	batch.partitionKey = NewPartitionKeyString("foo")
+	itemId := "bar"
+	batch.ReadItem(itemId, nil)
+
+	if len(batch.operations) != 1 {
+		t.Errorf("Expected 1 operation, but got %v", len(batch.operations))
+	}
+
+	if batch.operations[0].getOperationType() != operationTypeRead {
+		t.Errorf("Expected operation type %v, but got %v", operationTypeRead, batch.operations[0].getOperationType())
+	}
+
+	asRead := batch.operations[0].(batchOperationRead)
+
+	if asRead.operationType != "Read" {
+		t.Errorf("Expected operation type %v, but got %v", "Read", asRead.operationType)
+	}
+
+	if asRead.id != itemId {
+		t.Errorf("Expected id %v, but got %v", itemId, asRead.id)
+	}
+}
+
+func TestTransactionalBatchUpsertItem(t *testing.T) {
+	batch := &TransactionalBatch{}
+	batch.partitionKey = NewPartitionKeyString("foo")
+	body := map[string]string{
+		"foo": "bar",
+	}
+
+	itemMarshall, _ := json.Marshal(body)
+
+	options := &TransactionalBatchItemOptions{}
+	etag := azcore.ETag("someEtag")
+	options.IfMatchETag = &etag
+	batch.UpsertItem(itemMarshall, options)
+
+	if len(batch.operations) != 1 {
+		t.Errorf("Expected 1 operation, but got %v", len(batch.operations))
+	}
+
+	if batch.operations[0].getOperationType() != operationTypeUpsert {
+		t.Errorf("Expected operation type %v, but got %v", operationTypeUpsert, batch.operations[0].getOperationType())
+	}
+
+	asUpsert := batch.operations[0].(batchOperationUpsert)
+
+	if asUpsert.operationType != "Upsert" {
+		t.Errorf("Expected operation type %v, but got %v", "Upsert", asUpsert.operationType)
+	}
+
+	if asUpsert.ifMatch != options.IfMatchETag {
+		t.Errorf("Expected ifMatch %v, but got %v", etag, asUpsert.ifMatch)
+	}
+
+	if string(asUpsert.resourceBody) != string(itemMarshall) {
+		t.Errorf("Expected body %v, but got %v", string(itemMarshall), string(asUpsert.resourceBody))
+	}
+}
+
+func TestTransactionalBatchReplaceItem(t *testing.T) {
+	batch := &TransactionalBatch{}
+	batch.partitionKey = NewPartitionKeyString("foo")
+	body := map[string]string{
+		"foo": "bar",
+	}
+
+	itemMarshall, _ := json.Marshal(body)
+
+	options := &TransactionalBatchItemOptions{}
+	etag := azcore.ETag("someEtag")
+	options.IfMatchETag = &etag
+	itemId := "bar"
+	batch.ReplaceItem(itemId, itemMarshall, options)
+
+	if len(batch.operations) != 1 {
+		t.Errorf("Expected 1 operation, but got %v", len(batch.operations))
+	}
+
+	if batch.operations[0].getOperationType() != operationTypeReplace {
+		t.Errorf("Expected operation type %v, but got %v", operationTypeReplace, batch.operations[0].getOperationType())
+	}
+
+	asReplace := batch.operations[0].(batchOperationReplace)
+
+	if asReplace.operationType != "Replace" {
+		t.Errorf("Expected operation type %v, but got %v", "Replace", asReplace.operationType)
+	}
+
+	if asReplace.id != itemId {
+		t.Errorf("Expected id %v, but got %v", itemId, asReplace.id)
+	}
+
+	if asReplace.ifMatch != options.IfMatchETag {
+		t.Errorf("Expected ifMatch %v, but got %v", etag, asReplace.ifMatch)
+	}
+
+	if string(asReplace.resourceBody) != string(itemMarshall) {
+		t.Errorf("Expected body %v, but got %v", string(itemMarshall), string(asReplace.resourceBody))
+	}
+}
+
+func TestTransactionalBatchDeleteItem(t *testing.T) {
+	batch := &TransactionalBatch{}
+	batch.partitionKey = NewPartitionKeyString("foo")
+	options := &TransactionalBatchItemOptions{}
+	etag := azcore.ETag("someEtag")
+	options.IfMatchETag = &etag
+	itemId := "bar"
+	batch.DeleteItem(itemId, options)
+
+	if len(batch.operations) != 1 {
+		t.Errorf("Expected 1 operation, but got %v", len(batch.operations))
+	}
+
+	if batch.operations[0].getOperationType() != operationTypeDelete {
+		t.Errorf("Expected operation type %v, but got %v", operationTypeDelete, batch.operations[0].getOperationType())
+	}
+
+	asDelete := batch.operations[0].(batchOperationDelete)
+
+	if asDelete.operationType != "Delete" {
+		t.Errorf("Expected operation type %v, but got %v", "Delete", asDelete.operationType)
+	}
+
+	if asDelete.ifMatch != options.IfMatchETag {
+		t.Errorf("Expected ifMatch %v, but got %v", etag, asDelete.ifMatch)
+	}
+
+	if asDelete.id != itemId {
+		t.Errorf("Expected id %v, but got %v", itemId, asDelete.id)
+	}
+}
diff --git a/sdk/data/azcosmos/emulator_cosmos_batch_test.go b/sdk/data/azcosmos/emulator_cosmos_batch_test.go
new file mode 100644
index 0000000000..c44791ba5b
--- /dev/null
+++ b/sdk/data/azcosmos/emulator_cosmos_batch_test.go
@@ -0,0 +1,220 @@
+// Copyright (c) Microsoft Corporation. All rights reserved.
+// Licensed under the MIT License.
+
+package azcosmos
+
+import (
+	"context"
+	"net/http"
+	"testing"
+)
+
+func TestItemTransactionalBatch(t *testing.T) {
+	emulatorTests := newEmulatorTests(t)
+	client := emulatorTests.getClient(t)
+
+	database := emulatorTests.createDatabase(t, context.TODO(), client, "itemCRUD")
+	defer emulatorTests.deleteDatabase(t, context.TODO(), database)
+	properties := ContainerProperties{
+		ID: "aContainer",
+		PartitionKeyDefinition: PartitionKeyDefinition{
+			Paths: []string{"/pk"},
+		},
+	}
+
+	_, err := database.CreateContainer(context.TODO(), properties, nil)
+	if err != nil {
+		t.Fatalf("Failed to create container: %v", err)
+	}
+
+	container, _ := client.NewContainer("itemCRUD", "aContainer")
+
+	pkValue := NewPartitionKeyString("tBatch")
+	batch := container.NewTransactionalBatch(pkValue)
+
+	batch.CreateItem(emulatorTests.marshallItem("test", "tBatch"), nil)
+	batch.CreateItem(emulatorTests.marshallItem("test2", "tBatch"), nil)
+
+	// Default behavior has no content body
+	response, err := container.ExecuteTransactionalBatch(context.TODO(), batch, nil)
+	if err != nil {
+		t.Fatalf("Failed to execute batch: %v", err)
+	}
+
+	if len(response.OperationResults) != 2 {
+		t.Fatalf("Expected 2 operation results, got %v", len(response.OperationResults))
+	}
+
+	if !response.Success {
+		t.Fatalf("Expected committed to be true, got false")
+	}
+
+	for _, operationResult := range response.OperationResults {
+		if operationResult.StatusCode != http.StatusCreated {
+			t.Fatalf("Expected status code 201, got %v", operationResult.StatusCode)
+		}
+
+		if operationResult.RequestCharge <= 0 {
+			t.Fatalf("Expected RequestCharge to be greater than 0, got %v", operationResult.RequestCharge)
+		}
+
+		if operationResult.ETag == "" {
+			t.Fatalf("Expected ETag to be non-empty, got %v", operationResult.ETag)
+		}
+
+		if operationResult.ResourceBody != nil {
+			t.Fatalf("Expected ResourceBody to be nil, got %v", operationResult.ResourceBody)
+		}
+	}
+
+	batch2 := container.NewTransactionalBatch(pkValue)
+
+	batch2.CreateItem(emulatorTests.marshallItem("test3", "tBatch"), nil)
+	batch2.ReadItem("test2", nil)
+	batch2.DeleteItem("test", nil)
+
+	// If there is a read operation, body should be included
+	response2, err := container.ExecuteTransactionalBatch(context.TODO(), batch2, nil)
+	if err != nil {
+		t.Fatalf("Failed to execute batch: %v", err)
+	}
+
+	if !response2.Success {
+		t.Fatalf("Expected committed to be true, got false")
+	}
+
+	if len(response2.OperationResults) != 3 {
+		t.Fatalf("Expected 3 operation results, got %v", len(response2.OperationResults))
+	}
+
+	for index, operationResult := range response2.OperationResults {
+		if index == 0 && operationResult.StatusCode != http.StatusCreated {
+			t.Fatalf("Expected status code 201, got %v", operationResult.StatusCode)
+		}
+
+		if index == 1 && operationResult.StatusCode != http.StatusOK {
+			t.Fatalf("Expected status code 200, got %v", operationResult.StatusCode)
+		}
+
+		if index == 2 && operationResult.StatusCode != http.StatusNoContent {
+			t.Fatalf("Expected status code 204, got %v", operationResult.StatusCode)
+		}
+
+		if operationResult.RequestCharge <= 0 {
+			t.Fatalf("Expected RequestCharge to be greater than 0, got %v", operationResult.RequestCharge)
+		}
+
+		if index < 2 && operationResult.ETag == "" {
+			t.Fatalf("Expected ETag to be non-empty, got %v", operationResult.ETag)
+		}
+
+		if index < 2 && operationResult.ResourceBody == nil {
+			t.Fatalf("Expected ResourceBody to be not-nil, got %v", operationResult.ResourceBody)
+		}
+
+		if index == 2 && operationResult.ResourceBody != nil {
+			t.Fatalf("Expected ResourceBody to be nil, got %v", operationResult.ResourceBody)
+		}
+	}
+
+	// Forcing body through options
+	batch3 := container.NewTransactionalBatch(pkValue)
+
+	batch3.UpsertItem(emulatorTests.marshallItem("test4", "tBatch"), nil)
+	batch3.ReplaceItem("test3", emulatorTests.marshallItem("test3", "tBatch"), nil)
+
+	response3, err := container.ExecuteTransactionalBatch(context.TODO(), batch3, &TransactionalBatchOptions{EnableContentResponseOnWrite: true})
+	if err != nil {
+		t.Fatalf("Failed to execute batch: %v", err)
+	}
+
+	if !response3.Success {
+		t.Fatalf("Expected Success to be true, got false")
+	}
+
+	if len(response3.OperationResults) != 2 {
+		t.Fatalf("Expected 2 operation results, got %v", len(response3.OperationResults))
+	}
+
+	for index, operationResult := range response3.OperationResults {
+		if index == 0 && operationResult.StatusCode != http.StatusCreated {
+			t.Fatalf("Expected status code 201, got %v", operationResult.StatusCode)
+		}
+
+		if index == 1 && operationResult.StatusCode != http.StatusOK {
+			t.Fatalf("Expected status code 200, got %v", operationResult.StatusCode)
+		}
+
+		if operationResult.RequestCharge <= 0 {
+			t.Fatalf("Expected RequestCharge to be greater than 0, got %v", operationResult.RequestCharge)
+		}
+
+		if operationResult.ETag == "" {
+			t.Fatalf("Expected ETag to be non-empty, got %v", operationResult.ETag)
+		}
+
+		if operationResult.ResourceBody == nil {
+			t.Fatalf("Expected ResourceBody not to be nil, got %v", operationResult.ResourceBody)
+		}
+	}
+}
+
+func TestItemTransactionalBatchError(t *testing.T) {
+	emulatorTests := newEmulatorTests(t)
+	client := emulatorTests.getClient(t)
+
+	database := emulatorTests.createDatabase(t, context.TODO(), client, "itemCRUD")
+	defer emulatorTests.deleteDatabase(t, context.TODO(), database)
+	properties := ContainerProperties{
+		ID: "aContainer",
+		PartitionKeyDefinition: PartitionKeyDefinition{
+			Paths: []string{"/pk"},
+		},
+	}
+
+	_, err := database.CreateContainer(context.TODO(), properties, nil)
+	if err != nil {
+		t.Fatalf("Failed to create container: %v", err)
+	}
+
+	container, _ := client.NewContainer("itemCRUD", "aContainer")
+
+	pkValue := NewPartitionKeyString("tBatch")
+
+	_, err = container.CreateItem(context.TODO(), pkValue, emulatorTests.marshallItem("test", "tBatch"), nil)
+	if err != nil {
+		t.Fatalf("Failed to create item: %v", err)
+	}
+
+	batch := container.NewTransactionalBatch(pkValue)
+
+	batch.CreateItem(emulatorTests.marshallItem("test2", "tBatch"), nil)
+	batch.CreateItem(emulatorTests.marshallItem("test", "tBatch"), nil)
+
+	response, err := container.ExecuteTransactionalBatch(context.TODO(), batch, &TransactionalBatchOptions{EnableContentResponseOnWrite: true})
+	if err != nil {
+		t.Fatalf("Failed to execute batch: %v", err)
+	}
+
+	if response.RawResponse.StatusCode != http.StatusMultiStatus {
+		t.Fatalf("Expected status code 207, got %v", response.RawResponse.StatusCode)
+	}
+
+	if response.Success {
+		t.Fatalf("Expected Success to be false, got true")
+	}
+
+	if len(response.OperationResults) != 2 {
+		t.Fatalf("Expected 2 operation results, got %v", len(response.OperationResults))
+	}
+
+	for index, operationResult := range response.OperationResults {
+		if index == 0 && operationResult.StatusCode != http.StatusFailedDependency {
+			t.Fatalf("Expected status code 424, got %v", operationResult.StatusCode)
+		}
+
+		if index == 1 && operationResult.StatusCode != http.StatusConflict {
+			t.Fatalf("Expected status code 409, got %v", operationResult.StatusCode)
+		}
+	}
+}
diff --git a/sdk/data/azcosmos/emulator_tests.go b/sdk/data/azcosmos/emulator_tests.go
index c8994ab939..d36c2a19fc 100644
--- a/sdk/data/azcosmos/emulator_tests.go
+++ b/sdk/data/azcosmos/emulator_tests.go
@@ -6,6 +6,7 @@ package azcosmos
 import (
 	"context"
 	"encoding/base64"
+	"encoding/json"
 	"os"
 	"strconv"
 	"testing"
@@ -81,6 +82,16 @@ func (e *emulatorTests) deleteDatabase(
 	}
 }
 
+func (e *emulatorTests) marshallItem(id string, pk string) []byte {
+	item := map[string]string{
+		"id": id,
+		"pk": pk,
+	}
+
+	marshalled, _ := json.Marshal(item)
+	return marshalled
+}
+
 type emulatorTokenCredential struct {
 }
 
diff --git a/sdk/data/azcosmos/example_test.go b/sdk/data/azcosmos/example_test.go
index 32d5161468..79611af665 100644
--- a/sdk/data/azcosmos/example_test.go
+++ b/sdk/data/azcosmos/example_test.go
@@ -8,6 +8,7 @@ import (
 	"encoding/json"
 	"errors"
 	"fmt"
+	"net/http"
 	"os"
 	"time"
 
@@ -600,6 +601,80 @@ func ExampleContainerClient_NewQueryItemsPager() {
 	}
 }
 
+func ExampleContainerClient_NewTransactionalBatch() {
+	endpoint, ok := os.LookupEnv("AZURE_COSMOS_ENDPOINT")
+	if !ok {
+		panic("AZURE_COSMOS_ENDPOINT could not be found")
+	}
+
+	key, ok := os.LookupEnv("AZURE_COSMOS_KEY")
+	if !ok {
+		panic("AZURE_COSMOS_KEY could not be found")
+	}
+
+	cred, err := azcosmos.NewKeyCredential(key)
+	if err != nil {
+		panic(err)
+	}
+
+	client, err := azcosmos.NewClientWithKey(endpoint, cred, nil)
+	if err != nil {
+		panic(err)
+	}
+
+	container, err := client.NewContainer("databaseName", "aContainer")
+	if err != nil {
+		panic(err)
+	}
+
+	pk := azcosmos.NewPartitionKeyString("newPartitionKey")
+
+	batch := container.NewTransactionalBatch(pk)
+
+	item := map[string]string{
+		"id":             "anId",
+		"value":          "2",
+		"myPartitionKey": "newPartitionKey",
+	}
+
+	marshalledItem, err := json.Marshal(item)
+	if err != nil {
+		panic(err)
+	}
+
+	batch.CreateItem(marshalledItem, nil)
+	batch.ReadItem("anIdThatExists", nil)
+	batch.DeleteItem("yetAnotherExistingId", nil)
+
+	batchResponse, err := container.ExecuteTransactionalBatch(context.Background(), batch, nil)
+	if err != nil {
+		panic(err)
+	}
+
+	if batchResponse.Success {
+		// Transaction succeeded
+		// We can inspect the individual operation results
+		for index, operation := range batchResponse.OperationResults {
+			fmt.Printf("Operation %v completed with status code %v consumed %v RU", index, operation.StatusCode, operation.RequestCharge)
+			if index == 1 {
+				// Read operation would have body available
+				var itemResponseBody map[string]string
+				err = json.Unmarshal(operation.ResourceBody, &itemResponseBody)
+				if err != nil {
+					panic(err)
+				}
+			}
+		}
+	} else {
+		// Transaction failed, look for the offending operation
+		for index, operation := range batchResponse.OperationResults {
+			if operation.StatusCode != http.StatusFailedDependency {
+				fmt.Printf("Transaction failed due to operation %v which failed with status code %v", index, operation.StatusCode)
+			}
+		}
+	}
+}
+
 func retryOptimisticConcurrency(retryAttempts int, wait time.Duration, retry func() (bool, error)) (result error) {
 	for i := 0; ; i++ {
 		retryResult, err := retry()
diff --git a/sdk/messaging/azservicebus/admin_client_test.go b/sdk/messaging/azservicebus/admin_client_test.go
index 5ca6818dbb..56e3947dc2 100644
--- a/sdk/messaging/azservicebus/admin_client_test.go
+++ b/sdk/messaging/azservicebus/admin_client_test.go
@@ -75,9 +75,7 @@ func TestAdminClient_Queue_Forwarding(t *testing.T) {
 	forwardedMessages, err := receiver.ReceiveMessages(context.Background(), 1, nil)
 	require.NoError(t, err)
 
-	body, err := forwardedMessages[0].Body()
-	require.NoError(t, err)
-	require.EqualValues(t, "this message will be auto-forwarded", string(body))
+	require.EqualValues(t, "this message will be auto-forwarded", string(forwardedMessages[0].Body))
 }
 
 func TestAdminClient_GetQueueRuntimeProperties(t *testing.T) {
diff --git a/sdk/messaging/azservicebus/client_test.go b/sdk/messaging/azservicebus/client_test.go
index 7ce444d41b..7628314f53 100644
--- a/sdk/messaging/azservicebus/client_test.go
+++ b/sdk/messaging/azservicebus/client_test.go
@@ -120,9 +120,7 @@ func TestNewClientWithWebsockets(t *testing.T) {
 	messages, err := receiver.ReceiveMessages(context.Background(), 1, nil)
 	require.NoError(t, err)
 
-	bytes, err := messages[0].Body()
-	require.NoError(t, err)
-	require.EqualValues(t, "hello world", string(bytes))
+	require.EqualValues(t, "hello world", string(messages[0].Body))
 }
 
 func TestNewClientUsingSharedAccessSignature(t *testing.T) {
@@ -154,9 +152,7 @@ func TestNewClientUsingSharedAccessSignature(t *testing.T) {
 	messages, err := receiver.ReceiveMessages(context.Background(), 1, nil)
 	require.NoError(t, err)
 
-	bytes, err := messages[0].Body()
-	require.NoError(t, err)
-	require.EqualValues(t, "hello world", string(bytes))
+	require.EqualValues(t, "hello world", string(messages[0].Body))
 }
 
 const fastNotFoundDuration = 10 * time.Second
diff --git a/sdk/messaging/azservicebus/example_receiver_test.go b/sdk/messaging/azservicebus/example_receiver_test.go
index 203f6fdf35..9b8ae0a9af 100644
--- a/sdk/messaging/azservicebus/example_receiver_test.go
+++ b/sdk/messaging/azservicebus/example_receiver_test.go
@@ -88,6 +88,11 @@ func ExampleReceiver_ReceiveMessages() {
 	}
 
 	for _, message := range messages {
+		// The message body is a []byte. For this example we're just assuming that the body
+		// was a string, converted to bytes but any []byte payload is valid.
+		var body []byte = message.Body
+		fmt.Printf("Message received with body: %s\n", string(body))
+
 		// For more information about settling messages:
 		// https://docs.microsoft.com/azure/service-bus-messaging/message-transfers-locks-settlement#settling-receive-operations
 		err = receiver.CompleteMessage(context.TODO(), message, nil)
diff --git a/sdk/messaging/azservicebus/message_test.go b/sdk/messaging/azservicebus/message_test.go
index 70d360f9af..28668d2178 100644
--- a/sdk/messaging/azservicebus/message_test.go
+++ b/sdk/messaging/azservicebus/message_test.go
@@ -60,6 +60,9 @@ func TestAMQPMessageToReceivedMessage(t *testing.T) {
 		scheduledEnqueueTime := time.Now().Add(3 * time.Hour)
 
 		amqpMessage := &amqp.Message{
+			Data: [][]byte{
+				[]byte("hello"),
+			},
 			Annotations: map[interface{}]interface{}{
 				"x-opt-locked-until":            lockedUntil,
 				"x-opt-sequence-number":         int64(101),
@@ -72,6 +75,7 @@ func TestAMQPMessageToReceivedMessage(t *testing.T) {
 
 		receivedMessage := newReceivedMessage(amqpMessage)
 
+		require.Equal(t, []byte("hello"), receivedMessage.Body)
 		require.EqualValues(t, lockedUntil, *receivedMessage.LockedUntil)
 		require.EqualValues(t, int64(101), *receivedMessage.SequenceNumber)
 		require.EqualValues(t, "partitionKey1", *receivedMessage.PartitionKey)
@@ -143,9 +147,7 @@ func TestAMQPMessageToMessage(t *testing.T) {
 	require.EqualValues(t, msg.To, amqpMsg.Properties.To, "to")
 	require.EqualValues(t, MessageStateActive, msg.State)
 
-	body, err := msg.Body()
-	require.NoError(t, err)
-	require.EqualValues(t, body, amqpMsg.Data[0], "data")
+	require.EqualValues(t, msg.Body, amqpMsg.Data[0], "data")
 
 	expectedAMQPEncodedLockTokenGUID := [16]byte{187, 49, 89, 205, 253, 254, 205, 77, 162, 38, 172, 76, 45, 235, 91, 225}
 
@@ -189,3 +191,28 @@ func TestMessageState(t *testing.T) {
 		require.EqualValues(t, MessageStateActive, m.State)
 	})
 }
+
+func TestMessageWithIncorrectBody(t *testing.T) {
+	// these are cases where the simple ReceivedMessage can't represent the AMQP message's
+	// payload.
+	message := newReceivedMessage(&amqp.Message{})
+	require.Nil(t, message.Body)
+
+	message = newReceivedMessage(&amqp.Message{
+		Value: "hello",
+	})
+	require.Nil(t, message.Body)
+
+	message = newReceivedMessage(&amqp.Message{
+		Sequence: [][]any{},
+	})
+	require.Nil(t, message.Body)
+
+	message = newReceivedMessage(&amqp.Message{
+		Data: [][]byte{
+			[]byte("hello"),
+			[]byte("world"),
+		},
+	})
+	require.Nil(t, message.Body)
+}
diff --git a/sdk/messaging/azservicebus/receiver_test.go b/sdk/messaging/azservicebus/receiver_test.go
index c0608a47af..fd55b3d203 100644
--- a/sdk/messaging/azservicebus/receiver_test.go
+++ b/sdk/messaging/azservicebus/receiver_test.go
@@ -60,12 +60,9 @@ func TestReceiverSendFiveReceiveFive(t *testing.T) {
 	require.EqualValues(t, 5, len(messages))
 
 	for i := 0; i < 5; i++ {
-		body, err := messages[i].Body()
-		require.NoError(t, err)
-
 		require.EqualValues(t,
 			fmt.Sprintf("[%d]: send five, receive five", i),
-			string(body))
+			string(messages[i].Body))
 
 		require.NoError(t, receiver.CompleteMessage(context.Background(), messages[i], nil))
 	}
@@ -380,11 +377,8 @@ func TestReceiverPeek(t *testing.T) {
 	require.NoError(t, err)
 	require.EqualValues(t, 1, len(repeekedMessages))
 
-	body, err := peekedMessages2[0].Body()
-	require.NoError(t, err)
-
 	require.EqualValues(t, []string{
-		string(body),
+		string(peekedMessages2[0].Body),
 	}, getSortedBodies(repeekedMessages))
 
 	// and peek again (note it won't reset so there'll be "nothing")
@@ -789,19 +783,7 @@ func (messages receivedMessageSlice) Len() int {
 }
 
 func (messages receivedMessageSlice) Less(i, j int) bool {
-	bodyI, err := messages[i].Body()
-
-	if err != nil {
-		panic(err)
-	}
-
-	bodyJ, err := messages[j].Body()
-
-	if err != nil {
-		panic(err)
-	}
-
-	return string(bodyI) < string(bodyJ)
+	return string(messages[i].Body) < string(messages[j].Body)
 }
 
 func (messages receivedMessageSlice) Swap(i, j int) {
diff --git a/sdk/messaging/azservicebus/sender_test.go b/sdk/messaging/azservicebus/sender_test.go
index 36abb705fb..67154510a4 100644
--- a/sdk/messaging/azservicebus/sender_test.go
+++ b/sdk/messaging/azservicebus/sender_test.go
@@ -198,9 +198,7 @@ func Test_Sender_SendMessages_resend(t *testing.T) {
 		require.NoError(t, err)
 
 		require.EqualValues(t, "first send", messages[0].ApplicationProperties["Status"])
-		body, err := messages[0].Body()
-		require.NoError(t, err)
-		require.EqualValues(t, "ResendableMessage", string(body))
+		require.EqualValues(t, "ResendableMessage", string(messages[0].Body))
 
 		if complete {
 			require.NoError(t, receiver.CompleteMessage(ctx, messages[0], nil))
@@ -215,9 +213,7 @@ func Test_Sender_SendMessages_resend(t *testing.T) {
 		messages, err = receiver.ReceiveMessages(ctx, 1, nil)
 		require.NoError(t, err)
 		require.EqualValues(t, "resend", messages[0].ApplicationProperties["Status"])
-		body, err = messages[0].Body()
-		require.NoError(t, err)
-		require.EqualValues(t, "ResendableMessage", string(body))
+		require.EqualValues(t, "ResendableMessage", string(messages[0].Body))
 
 		if complete {
 			require.NoError(t, receiver.CompleteMessage(ctx, messages[0], nil))
@@ -229,14 +225,11 @@ func Test_Sender_SendMessages_resend(t *testing.T) {
 }
 
 func messageFromReceivedMessage(t *testing.T, receivedMessage *ReceivedMessage) *Message {
-	body, err := receivedMessage.Body()
-	require.NoError(t, err)
-
 	newMsg := &Message{
 		MessageID:             &receivedMessage.MessageID,
 		ContentType:           receivedMessage.ContentType,
 		CorrelationID:         receivedMessage.CorrelationID,
-		Body:                  body,
+		Body:                  receivedMessage.Body,
 		SessionID:             receivedMessage.SessionID,
 		Subject:               receivedMessage.Subject,
 		ReplyTo:               receivedMessage.ReplyTo,
@@ -430,13 +423,7 @@ func getSortedBodies(messages []*ReceivedMessage) []string {
 	var bodies []string
 
 	for _, msg := range messages {
-		body, err := msg.Body()
-
-		if err != nil {
-			panic(err)
-		}
-
-		bodies = append(bodies, string(body))
+		bodies = append(bodies, string(msg.Body))
 	}
 
 	return bodies
@@ -450,19 +437,7 @@ func (rm receivedMessages) Len() int {
 
 // Less compares the messages assuming the .Body field is a valid string.
 func (rm receivedMessages) Less(i, j int) bool {
-	bodyI, err := rm[i].Body()
-
-	if err != nil {
-		panic(err)
-	}
-
-	bodyJ, err := rm[j].Body()
-
-	if err != nil {
-		panic(err)
-	}
-
-	return string(bodyI) < string(bodyJ)
+	return string(rm[i].Body) < string(rm[j].Body)
 }
 
 func (rm receivedMessages) Swap(i, j int) {
diff --git a/sdk/messaging/azservicebus/session_receiver_test.go b/sdk/messaging/azservicebus/session_receiver_test.go
index 84563b6d22..cfc8348933 100644
--- a/sdk/messaging/azservicebus/session_receiver_test.go
+++ b/sdk/messaging/azservicebus/session_receiver_test.go
@@ -43,10 +43,7 @@ func TestSessionReceiver_acceptSession(t *testing.T) {
 	messages, err := receiver.inner.ReceiveMessages(ctx, 1, nil)
 	require.NoError(t, err)
 
-	body, err := messages[0].Body()
-	require.NoError(t, err)
-
-	require.EqualValues(t, "session-based message", body)
+	require.EqualValues(t, "session-based message", messages[0].Body)
 	require.EqualValues(t, "session-1", *messages[0].SessionID)
 	require.NoError(t, receiver.CompleteMessage(ctx, messages[0], nil))
 
@@ -112,11 +109,8 @@ func TestSessionReceiver_blankSessionIDs(t *testing.T) {
 	}
 
 	for _, msg := range received {
-		body, err := msg.Body()
-		require.NoError(t, err)
 		require.EqualValues(t, "", *msg.SessionID)
-
-		require.EqualValues(t, "session-based message", string(body))
+		require.EqualValues(t, "session-based message", string(msg.Body))
 	}
 }
 
@@ -165,9 +159,7 @@ func TestSessionReceiver_acceptNextSession(t *testing.T) {
 	messages, err := receiver.inner.ReceiveMessages(ctx, 1, nil)
 	require.NoError(t, err)
 
-	body, err := messages[0].Body()
-	require.NoError(t, err)
-	require.EqualValues(t, "session-based message", body)
+	require.EqualValues(t, "session-based message", messages[0].Body)
 	require.EqualValues(t, "acceptnextsession-test", *messages[0].SessionID)
 	require.NoError(t, receiver.CompleteMessage(ctx, messages[0], nil))