Cross entity transactions (Azure#18402)

* cross entity txn

* Add processor support and address pr fb

* Export API

* PR FB

* Export API

Author: JoshLove-msft

sdk/servicebus/Azure.Messaging.ServiceBus/MigrationGuide.md

@@ -14,6 +14,7 @@ We assume that you are familiar with the `Microsoft.Azure.ServiceBus` library. I
   - [Sending messages](#sending-messages)
   - [Receiving messages](#receiving-messages)
   - [Working with sessions](#working-with-sessions)
+  - [Cross-entity transactions](#cross-entity-transactions)
 - [Known gaps](#known-gaps-from-previous-library)
 - [Additional samples](#additional-samples)
 
@@ -284,7 +285,7 @@ Previously, in `Microsoft.Azure.ServiceBus`, you had the below options to receiv
 
 While the first option is similar to what you would do in a non-session scenario, the second that allows you finer-grained control is very different from any other pattern used in the library.
 
-Now in `Azure.Messaging.ServiceBus`, we simplfify this by giving session variants of the same methods and classes that are available when working with queues/subscriptions that do not have sessions enabled.
+Now in `Azure.Messaging.ServiceBus`, we simplify this by giving session variants of the same methods and classes that are available when working with queues/subscriptions that do not have sessions enabled.
 
 The below code snippet shows you the session variation of the `ServiceBusProcessor`.
 
@@ -367,6 +368,49 @@ ServiceBusReceivedMessage receivedMessage = await receiver.ReceiveMessageAsync()
 Console.WriteLine(receivedMessage.SessionId);
 ```
 
+### Cross-Entity transactions (Unreleased)
+
+Previously, in `Microsoft.Azure.ServiceBus`, when performing a transaction that spanned multiple queues, topics, or subscriptions you would need to use the "Send-Via" option
+in the `MessageSender`. 
+
+Now in `Azure.Messaging.ServiceBus`, there is a `TransactionGroup` property on the sender, receiver, and processor options bags. This property is used to group senders, receivers, and processors that will be involved in entities that span transactions. When using this property, the first operation that occurs among this group implicitly becomes the send-via entity. Because of this, subsequent operations must either be by senders, or if they are by receivers, the receiver must be receiving from the send-via entity. For this reason, it probably makes more sense to have your first operation be a receive rather than a send when using transaction groups.
+
+The below code snippet shows you how to use transaction groups.
+
+```C# Snippet:ServiceBusTransactionGroup
+// The first sender won't be part of our transaction group.
+ServiceBusSender senderA = client.CreateSender(queueA.QueueName);
+
+string transactionGroup = "myTxn";
+
+ServiceBusReceiver receiverA = client.CreateReceiver(queueA.QueueName, new ServiceBusReceiverOptions
+{
+    TransactionGroup = transactionGroup
+});
+ServiceBusSender senderB = client.CreateSender(queueB.QueueName, new ServiceBusSenderOptions
+{
+    TransactionGroup = transactionGroup
+});
+ServiceBusSender senderC = client.CreateSender(topicC.TopicName, new ServiceBusSenderOptions
+{
+    TransactionGroup = transactionGroup
+});
+
+var message = new ServiceBusMessage();
+
+await senderA.SendMessageAsync(message);
+
+ServiceBusReceivedMessage receivedMessage = await receiverA.ReceiveMessageAsync();
+
+using (var ts = new TransactionScope(TransactionScopeAsyncFlowOption.Enabled))
+{
+    await receiverA.CompleteMessageAsync(receivedMessage);
+    await senderB.SendMessageAsync(message);
+    await senderC.SendMessageAsync(message);
+    ts.Complete();
+}
+```
+
 ## Known Gaps from Previous Library
 There are a few features that are yet to be implemented in `Azure.Messaging.ServiceBus`, but were present in the previous library `Microsoft.Azure.ServiceBus`. The plan is to add these features in upcoming releases (unless otherwise noted), but they will not be available in the version 7.0.0:
 - **Cross entity transactions** - In the previous library, Microsoft.Azure.ServiceBus, transactions could work across multiple entities by leveraging the [`viaEntityPath`](https://github.com/Azure/azure-sdk-for-net/blob/master/sdk/servicebus/Microsoft.Azure.ServiceBus/src/Core/MessageSender.cs#L118) parameter of the `MessageSender` constructor. The service is planning to make backward compatible updates that would make it possible to do cross-entity transactions without specificing a "via" entity. The new library plans to take advantage of this new feature and therefore, the cross entity transactions feature will be available in the upcoming release instead of the current version 7.0.0. Support for this feature can be tracked via https://github.com/Azure/azure-sdk-for-net/issues/17355.

sdk/servicebus/Azure.Messaging.ServiceBus/api/Azure.Messaging.ServiceBus.netstandard2.0.cs

@@ -168,6 +168,7 @@ public ServiceBusMessage(string body) { }
         public string Subject { get { throw null; } set { } }
         public System.TimeSpan TimeToLive { get { throw null; } set { } }
         public string To { get { throw null; } set { } }
+        public string TransactionPartitionKey { get { throw null; } set { } }
         public Azure.Core.Amqp.AmqpAnnotatedMessage GetRawAmqpMessage() { throw null; }
         public override string ToString() { throw null; }
     }
@@ -201,6 +202,7 @@ protected ServiceBusProcessor() { }
         public virtual int MaxConcurrentCalls { get { throw null; } }
         public virtual int PrefetchCount { get { throw null; } }
         public virtual Azure.Messaging.ServiceBus.ServiceBusReceiveMode ReceiveMode { get { throw null; } }
+        public virtual string TransactionGroup { get { throw null; } }
         public event System.Func<Azure.Messaging.ServiceBus.ProcessErrorEventArgs, System.Threading.Tasks.Task> ProcessErrorAsync { add { } remove { } }
         public event System.Func<Azure.Messaging.ServiceBus.ProcessMessageEventArgs, System.Threading.Tasks.Task> ProcessMessageAsync { add { } remove { } }
         public virtual System.Threading.Tasks.Task CloseAsync(System.Threading.CancellationToken cancellationToken = default(System.Threading.CancellationToken)) { throw null; }
@@ -222,6 +224,7 @@ public ServiceBusProcessorOptions() { }
         public int MaxConcurrentCalls { get { throw null; } set { } }
         public int PrefetchCount { get { throw null; } set { } }
         public Azure.Messaging.ServiceBus.ServiceBusReceiveMode ReceiveMode { get { throw null; } set { } }
+        public string TransactionGroup { get { throw null; } set { } }
         [System.ComponentModel.EditorBrowsableAttribute(System.ComponentModel.EditorBrowsableState.Never)]
         public override bool Equals(object obj) { throw null; }
         [System.ComponentModel.EditorBrowsableAttribute(System.ComponentModel.EditorBrowsableState.Never)]
@@ -255,6 +258,7 @@ internal ServiceBusReceivedMessage() { }
         public string Subject { get { throw null; } }
         public System.TimeSpan TimeToLive { get { throw null; } }
         public string To { get { throw null; } }
+        public string TransactionPartitionKey { get { throw null; } }
         public Azure.Core.Amqp.AmqpAnnotatedMessage GetRawAmqpMessage() { throw null; }
         public override string ToString() { throw null; }
     }
@@ -271,6 +275,7 @@ protected ServiceBusReceiver() { }
         public virtual bool IsClosed { get { throw null; } }
         public virtual int PrefetchCount { get { throw null; } }
         public virtual Azure.Messaging.ServiceBus.ServiceBusReceiveMode ReceiveMode { get { throw null; } }
+        public virtual string TransactionGroup { get { throw null; } }
         public virtual System.Threading.Tasks.Task AbandonMessageAsync(Azure.Messaging.ServiceBus.ServiceBusReceivedMessage message, System.Collections.Generic.IDictionary<string, object> propertiesToModify = null, System.Threading.CancellationToken cancellationToken = default(System.Threading.CancellationToken)) { throw null; }
         public virtual System.Threading.Tasks.Task CloseAsync(System.Threading.CancellationToken cancellationToken = default(System.Threading.CancellationToken)) { throw null; }
         public virtual System.Threading.Tasks.Task CompleteMessageAsync(Azure.Messaging.ServiceBus.ServiceBusReceivedMessage message, System.Threading.CancellationToken cancellationToken = default(System.Threading.CancellationToken)) { throw null; }
@@ -299,6 +304,7 @@ public ServiceBusReceiverOptions() { }
         public int PrefetchCount { get { throw null; } set { } }
         public Azure.Messaging.ServiceBus.ServiceBusReceiveMode ReceiveMode { get { throw null; } set { } }
         public Azure.Messaging.ServiceBus.SubQueue SubQueue { get { throw null; } set { } }
+        public string TransactionGroup { get { throw null; } set { } }
         [System.ComponentModel.EditorBrowsableAttribute(System.ComponentModel.EditorBrowsableState.Never)]
         public override bool Equals(object obj) { throw null; }
         [System.ComponentModel.EditorBrowsableAttribute(System.ComponentModel.EditorBrowsableState.Never)]
@@ -339,6 +345,7 @@ protected ServiceBusSender() { }
         public virtual string EntityPath { get { throw null; } }
         public virtual string FullyQualifiedNamespace { get { throw null; } }
         public virtual bool IsClosed { get { throw null; } }
+        public virtual string TransactionGroup { get { throw null; } }
         public virtual System.Threading.Tasks.Task CancelScheduledMessageAsync(long sequenceNumber, System.Threading.CancellationToken cancellationToken = default(System.Threading.CancellationToken)) { throw null; }
         public virtual System.Threading.Tasks.Task CancelScheduledMessagesAsync(System.Collections.Generic.IEnumerable<long> sequenceNumbers, System.Threading.CancellationToken cancellationToken = default(System.Threading.CancellationToken)) { throw null; }
         public virtual System.Threading.Tasks.Task CloseAsync(System.Threading.CancellationToken cancellationToken = default(System.Threading.CancellationToken)) { throw null; }
@@ -357,6 +364,17 @@ protected ServiceBusSender() { }
         [System.ComponentModel.EditorBrowsableAttribute(System.ComponentModel.EditorBrowsableState.Never)]
         public override string ToString() { throw null; }
     }
+    public partial class ServiceBusSenderOptions
+    {
+        public ServiceBusSenderOptions() { }
+        public string TransactionGroup { get { throw null; } set { } }
+        [System.ComponentModel.EditorBrowsableAttribute(System.ComponentModel.EditorBrowsableState.Never)]
+        public override bool Equals(object obj) { throw null; }
+        [System.ComponentModel.EditorBrowsableAttribute(System.ComponentModel.EditorBrowsableState.Never)]
+        public override int GetHashCode() { throw null; }
+        [System.ComponentModel.EditorBrowsableAttribute(System.ComponentModel.EditorBrowsableState.Never)]
+        public override string ToString() { throw null; }
+    }
     public partial class ServiceBusSessionProcessor : System.IAsyncDisposable
     {
         protected ServiceBusSessionProcessor() { }
@@ -371,6 +389,7 @@ protected ServiceBusSessionProcessor() { }
         public virtual int PrefetchCount { get { throw null; } }
         public virtual Azure.Messaging.ServiceBus.ServiceBusReceiveMode ReceiveMode { get { throw null; } }
         public virtual System.TimeSpan? SessionIdleTimeout { get { throw null; } }
+        public virtual string TransactionGroup { get { throw null; } }
         public event System.Func<Azure.Messaging.ServiceBus.ProcessErrorEventArgs, System.Threading.Tasks.Task> ProcessErrorAsync { add { } remove { } }
         public event System.Func<Azure.Messaging.ServiceBus.ProcessSessionMessageEventArgs, System.Threading.Tasks.Task> ProcessMessageAsync { add { } remove { } }
         public event System.Func<Azure.Messaging.ServiceBus.ProcessSessionEventArgs, System.Threading.Tasks.Task> SessionClosingAsync { add { } remove { } }
@@ -397,6 +416,7 @@ public ServiceBusSessionProcessorOptions() { }
         public Azure.Messaging.ServiceBus.ServiceBusReceiveMode ReceiveMode { get { throw null; } set { } }
         public System.TimeSpan? SessionIdleTimeout { get { throw null; } set { } }
         public System.Collections.Generic.IList<string> SessionIds { get { throw null; } }
+        public string TransactionGroup { get { throw null; } set { } }
         [System.ComponentModel.EditorBrowsableAttribute(System.ComponentModel.EditorBrowsableState.Never)]
         public override bool Equals(object obj) { throw null; }
         [System.ComponentModel.EditorBrowsableAttribute(System.ComponentModel.EditorBrowsableState.Never)]
@@ -418,6 +438,7 @@ public partial class ServiceBusSessionReceiverOptions
         public ServiceBusSessionReceiverOptions() { }
         public int PrefetchCount { get { throw null; } set { } }
         public Azure.Messaging.ServiceBus.ServiceBusReceiveMode ReceiveMode { get { throw null; } set { } }
+        public string TransactionGroup { get { throw null; } set { } }
         [System.ComponentModel.EditorBrowsableAttribute(System.ComponentModel.EditorBrowsableState.Never)]
         public override bool Equals(object obj) { throw null; }
         [System.ComponentModel.EditorBrowsableAttribute(System.ComponentModel.EditorBrowsableState.Never)]

sdk/servicebus/Azure.Messaging.ServiceBus/samples/Sample06_Transactions.md

@@ -44,6 +44,84 @@ using (var ts = new TransactionScope(TransactionScopeAsyncFlowOption.Enabled))
 }
 ```
 
+### Transactions across entities (Unreleased)
+
+When creating senders and receivers that should be part of a cross-entity transaction, you can specify the same `TransactionGroup` string in the options as shown below. When using transaction groups, the first entity that an operation occurs on becomes the entity through which all subsequent sends will be routed through. This enables the service to perform a transaction that is meant to span multiple entities. This means that subsequent entities that perform their first operation need to either be senders, or if they are receivers they need to be on the same entity as the initial entity through which all sends are routed through (otherwise the service would not be able to ensure that the transaction is committed because it cannot route a receive operation through a different entity). For instance, if you have SenderA and ReceiverB that are part of the same transaction group, you would need to receive first with ReceiverB to allow this to work. If you first used SenderA to send to QueueA, and then attempted to receive from QueueB, an `InvalidOperationException` would be thrown. You could still add a ReceiverA to the same transaction group after initially sending to SenderA, since they are both using the same queue. This would be useful if you also had a SenderB that you want to include as part of the transaction group (otherwise there would be no need to use transaction groups as you would be dealing with only one entity).
+
+```C# Snippet:ServiceBusTransactionGroup
+// The first sender won't be part of our transaction group.
+ServiceBusSender senderA = client.CreateSender(queueA.QueueName);
+
+string transactionGroup = "myTxn";
+
+ServiceBusReceiver receiverA = client.CreateReceiver(queueA.QueueName, new ServiceBusReceiverOptions
+{
+    TransactionGroup = transactionGroup
+});
+ServiceBusSender senderB = client.CreateSender(queueB.QueueName, new ServiceBusSenderOptions
+{
+    TransactionGroup = transactionGroup
+});
+ServiceBusSender senderC = client.CreateSender(topicC.TopicName, new ServiceBusSenderOptions
+{
+    TransactionGroup = transactionGroup
+});
+
+var message = new ServiceBusMessage();
+
+await senderA.SendMessageAsync(message);
+
+ServiceBusReceivedMessage receivedMessage = await receiverA.ReceiveMessageAsync();
+
+using (var ts = new TransactionScope(TransactionScopeAsyncFlowOption.Enabled))
+{
+    await receiverA.CompleteMessageAsync(receivedMessage);
+    await senderB.SendMessageAsync(message);
+    await senderC.SendMessageAsync(message);
+    ts.Complete();
+}
+```
+In the following snippet, we will look at an incorrect ordering that would cause an `InvalidOperationException` to be thrown.
+Bad:
+```C# Snippet:ServiceBusTransactionGroupWrongOrder
+// The first sender won't be part of our transaction group.
+ServiceBusSender senderA = client.CreateSender(queueA.QueueName);
+
+string transactionGroup = "myTxn";
+
+ServiceBusReceiver receiverA = client.CreateReceiver(queueA.QueueName, new ServiceBusReceiverOptions
+{
+    TransactionGroup = transactionGroup
+});
+ServiceBusSender senderB = client.CreateSender(queueB.QueueName, new ServiceBusSenderOptions
+{
+    TransactionGroup = transactionGroup
+});
+ServiceBusSender senderC = client.CreateSender(topicC.TopicName, new ServiceBusSenderOptions
+{
+    TransactionGroup = transactionGroup
+});
+
+var message = new ServiceBusMessage();
+
+// SenderB becomes the entity through which subsequent "sends" are routed through.
+await senderB.SendMessageAsync(message);
+
+await senderA.SendMessageAsync(message);
+
+ServiceBusReceivedMessage receivedMessage = await receiverA.ReceiveMessageAsync();
+
+using (var ts = new TransactionScope(TransactionScopeAsyncFlowOption.Enabled))
+{
+    // This will through an InvalidOperationException because a "receive" cannot be
+    // routed through a different entity.
+    await receiverA.CompleteMessageAsync(receivedMessage);
+    await senderB.SendMessageAsync(message);
+    await senderC.SendMessageAsync(message);
+    ts.Complete();
+}
+```
+
 ## Source
 
 To see the full example source, see:

sdk/servicebus/Azure.Messaging.ServiceBus/src/Amqp/AmqpClient.cs

@@ -109,12 +109,14 @@ internal AmqpClient(
         /// <param name="entityPath">The entity path to send the message to.</param>
         /// <param name="retryPolicy">The policy which governs retry behavior and try timeouts.</param>
         /// <param name="identifier">The identifier for the sender.</param>
+        /// <param name="transactionGroup"></param>
         ///
         /// <returns>A <see cref="TransportSender"/> configured in the requested manner.</returns>
         public override TransportSender CreateSender(
             string entityPath,
             ServiceBusRetryPolicy retryPolicy,
-            string identifier)
+            string identifier,
+            string transactionGroup)
         {
             Argument.AssertNotDisposed(_closed, nameof(AmqpClient));
 
@@ -123,7 +125,8 @@ public override TransportSender CreateSender(
                 entityPath,
                 ConnectionScope,
                 retryPolicy,
-                identifier
+                identifier,
+                transactionGroup
             );
         }
 
@@ -139,6 +142,7 @@ public override TransportSender CreateSender(
         /// <param name="identifier"></param>
         /// <param name="sessionId"></param>
         /// <param name="isSessionReceiver"></param>
+        /// <param name="transactionGroup"></param>
         ///
         /// <returns>A <see cref="TransportReceiver" /> configured in the requested manner.</returns>
         ///
@@ -149,7 +153,8 @@ public override TransportReceiver CreateReceiver(
             uint prefetchCount,
             string identifier,
             string sessionId,
-            bool isSessionReceiver)
+            bool isSessionReceiver,
+            string transactionGroup)
         {
             Argument.AssertNotDisposed(_closed, nameof(AmqpClient));
 
@@ -162,7 +167,8 @@ public override TransportReceiver CreateReceiver(
                 retryPolicy,
                 identifier,
                 sessionId,
-                isSessionReceiver
+                isSessionReceiver,
+                transactionGroup
             );
         }