Log retries as Verbose when attempting to accept sessions (Azure#28302)

JoshLove-msft · web-flow · commit bb449821bd23 · 2022-04-24T23:20:24.000-07:00
* Log retries as Verbose when attempting to accept sessions

* Fix property name in exception

* Fix flaky test

* Revert to warning for other retries

* Add identifier/entity path filters to tests

* Add identifier to error handler event

* fix
diff --git a/sdk/servicebus/Azure.Messaging.ServiceBus/src/Administration/CreateSubscriptionOptions.cs b/sdk/servicebus/Azure.Messaging.ServiceBus/src/Administration/CreateSubscriptionOptions.cs
@@ -182,7 +182,7 @@ public int MaxDeliveryCount
                 Argument.AssertAtLeast(
                     value,
                     AdministrationClientConstants.MinAllowedMaxDeliveryCount,
-                    nameof(AutoDeleteOnIdle));
+                    nameof(MaxDeliveryCount));
 
                 _maxDeliveryCount = value;
             }
diff --git a/sdk/servicebus/Azure.Messaging.ServiceBus/src/Amqp/AmqpReceiver.cs b/sdk/servicebus/Azure.Messaging.ServiceBus/src/Amqp/AmqpReceiver.cs
@@ -1351,7 +1351,7 @@ private static TimeSpan UseMinimum(
             (firstOption < secondOption) ? firstOption : secondOption;
 
         /// <summary>
-        /// Opens an AMQP link for use with receiver operations.
+        /// Opens an AMQP link for use with session receiver operations.
         /// </summary>
         /// <param name="cancellationToken">An optional <see cref="CancellationToken"/> instance to signal the request to cancel the operation.</param>
         /// <returns>A task to be resolved on when the operation has completed.</returns>
@@ -1362,7 +1362,8 @@ static async (receiveLink, timeout, token) =>
                     await receiveLink.GetOrCreateAsync(timeout, token).ConfigureAwait(false),
                 _receiveLink,
                 _connectionScope,
-                cancellationToken).ConfigureAwait(false);
+                cancellationToken,
+                true).ConfigureAwait(false);
         }
 
         private bool HasLinkCommunicationError(ReceivingAmqpLink link) =>
diff --git a/sdk/servicebus/Azure.Messaging.ServiceBus/src/Diagnostics/ServiceBusEventSource.cs b/sdk/servicebus/Azure.Messaging.ServiceBus/src/Diagnostics/ServiceBusEventSource.cs
@@ -194,6 +194,8 @@ protected ServiceBusEventSource() : base(EventSourceName)
 
         internal const int ProcessorStoppingCancellationWarningEvent = 113;
 
+        internal const int RunOperationExceptionVerboseEvent = 114;
+
         #endregion
         // add new event numbers here incrementing from previous
 
@@ -844,12 +846,12 @@ public virtual void ProcessorRenewSessionLockException(string identifier, string
             }
         }
 
-        [Event(ProcessorErrorHandlerThrewExceptionEvent, Level = EventLevel.Error, Message = "ExceptionReceivedHandler threw exception. Exception:{0}")]
-        public void ProcessorErrorHandlerThrewException(string exception)
+        [Event(ProcessorErrorHandlerThrewExceptionEvent, Level = EventLevel.Error, Message = "{1}: ExceptionReceivedHandler threw exception. Exception:{0}")]
+        public void ProcessorErrorHandlerThrewException(string exception, string identifier)
         {
             if (IsEnabled())
             {
-                WriteEvent(ProcessorErrorHandlerThrewExceptionEvent, exception);
+                WriteEvent(ProcessorErrorHandlerThrewExceptionEvent, exception, identifier);
             }
         }
 
@@ -1357,14 +1359,23 @@ public virtual void RequestAuthorizationException(string identifier, string endp
 
         #region Retries
 
-        [Event(RunOperationExceptionEvent, Level = EventLevel.Informational, Message = "RunOperation encountered an exception and will retry. Exception: {0}")]
+        [Event(RunOperationExceptionEvent, Level = EventLevel.Warning, Message = "RunOperation encountered an exception and will retry. Exception: {0}")]
         public virtual void RunOperationExceptionEncountered(string exception)
         {
             if (IsEnabled())
             {
                 WriteEvent(RunOperationExceptionEvent, exception);
             }
         }
+
+        [Event(RunOperationExceptionVerboseEvent, Level = EventLevel.Verbose, Message = "RunOperation encountered an exception and will retry. Exception: {0}")]
+        public virtual void RunOperationExceptionEncounteredVerbose(string exception)
+        {
+            if (IsEnabled())
+            {
+                WriteEvent(RunOperationExceptionVerboseEvent, exception);
+            }
+        }
         #endregion
 
         #region Client lifecycle
diff --git a/sdk/servicebus/Azure.Messaging.ServiceBus/src/Primitives/ServiceBusRetryPolicy.cs b/sdk/servicebus/Azure.Messaging.ServiceBus/src/Primitives/ServiceBusRetryPolicy.cs
@@ -118,7 +118,8 @@ internal async ValueTask<TResult> RunOperation<T1, TResult>(
             Func<T1, TimeSpan, CancellationToken, ValueTask<TResult>> operation,
             T1 t1,
             TransportConnectionScope scope,
-            CancellationToken cancellationToken)
+            CancellationToken cancellationToken,
+            bool logRetriesAsVerbose = false)
       {
             var failedAttemptCount = 0;
 
@@ -160,7 +161,15 @@ internal async ValueTask<TResult> RunOperation<T1, TResult>(
                     TimeSpan? retryDelay = CalculateRetryDelay(activeEx, failedAttemptCount);
                     if (retryDelay.HasValue && !scope.IsDisposed && !cancellationToken.IsCancellationRequested)
                     {
-                        Logger.RunOperationExceptionEncountered(activeEx.ToString());
+                        if (logRetriesAsVerbose)
+                        {
+                            Logger.RunOperationExceptionEncounteredVerbose(activeEx.ToString());
+                        }
+                        else
+                        {
+                            Logger.RunOperationExceptionEncountered(activeEx.ToString());
+                        }
+
                         await Task.Delay(retryDelay.Value, cancellationToken).ConfigureAwait(false);
                         tryTimeout = CalculateTryTimeout(failedAttemptCount);
                     }
diff --git a/sdk/servicebus/Azure.Messaging.ServiceBus/src/Processor/ReceiverManager.cs b/sdk/servicebus/Azure.Messaging.ServiceBus/src/Processor/ReceiverManager.cs
@@ -346,7 +346,7 @@ protected virtual async Task RaiseExceptionReceived(ProcessErrorEventArgs eventA
             catch (Exception exception)
             {
                 // don't bubble up exceptions raised from customer exception handler
-                ServiceBusEventSource.Log.ProcessorErrorHandlerThrewException(exception.ToString());
+                ServiceBusEventSource.Log.ProcessorErrorHandlerThrewException(exception.ToString(), Processor.Identifier);
             }
         }
 
diff --git a/sdk/servicebus/Azure.Messaging.ServiceBus/src/Processor/ServiceBusProcessor.cs b/sdk/servicebus/Azure.Messaging.ServiceBus/src/Processor/ServiceBusProcessor.cs
@@ -892,7 +892,7 @@ await OnProcessErrorAsync(
                     catch (Exception ex)
                     {
                         // Don't bubble up exceptions raised from customer exception handler.
-                        Logger.ProcessorErrorHandlerThrewException(ex.ToString());
+                        Logger.ProcessorErrorHandlerThrewException(ex.ToString(), Identifier);
                     }
 
                     // This call will deadlock if awaited, as StopProcessingAsync awaits
diff --git a/sdk/servicebus/Azure.Messaging.ServiceBus/src/Processor/SessionReceiverManager.cs b/sdk/servicebus/Azure.Messaging.ServiceBus/src/Processor/SessionReceiverManager.cs
@@ -411,7 +411,7 @@ protected override async Task RaiseExceptionReceived(ProcessErrorEventArgs event
             catch (Exception exception)
             {
                 // don't bubble up exceptions raised from customer exception handler
-                ServiceBusEventSource.Log.ProcessorErrorHandlerThrewException(exception.ToString());
+                ServiceBusEventSource.Log.ProcessorErrorHandlerThrewException(exception.ToString(), Processor.Identifier);
             }
         }
 
diff --git a/sdk/servicebus/Azure.Messaging.ServiceBus/tests/Diagnostics/EventSourceLiveTests.cs b/sdk/servicebus/Azure.Messaging.ServiceBus/tests/Diagnostics/EventSourceLiveTests.cs
@@ -315,18 +315,62 @@ Task ProcessMessage(ProcessMessageEventArgs args)
 
                 Task ExceptionHandler(ProcessErrorEventArgs args)
                 {
+                    if (args.ErrorSource == ServiceBusErrorSource.ProcessMessageCallback)
+                    {
+                        throw new Exception();
+                    }
+
+                    return Task.CompletedTask;
+                }
+
+                processor.ProcessMessageAsync += ProcessMessage;
+                processor.ProcessErrorAsync += ExceptionHandler;
+
+                await processor.StartProcessingAsync();
+                await tcs.Task;
+                await processor.StopProcessingAsync();
+                _listener.SingleEventById(ServiceBusEventSource.ProcessorMessageHandlerStartEvent, args => args.Payload.Contains(processor.Identifier));
+                _listener.SingleEventById(ServiceBusEventSource.ProcessorMessageHandlerExceptionEvent, args => args.Payload.Contains(processor.Identifier));
+                _listener.SingleEventById(ServiceBusEventSource.ProcessorErrorHandlerThrewExceptionEvent, args => args.Payload.Contains(processor.Identifier));
+            }
+        }
+
+        [Test]
+        public async Task LogsSessionProcessorExceptionEvent()
+        {
+            await using (var scope = await ServiceBusScope.CreateWithQueue(enablePartitioning: false, enableSession: true))
+            {
+                await using var client = CreateClient();
+                var sender = client.CreateSender(scope.QueueName);
+                await sender.SendMessageAsync(ServiceBusTestUtilities.GetMessage("sessionId"));
+                await using var processor = client.CreateSessionProcessor(scope.QueueName);
+                var tcs = new TaskCompletionSource<bool>();
+
+                Task ProcessMessage(ProcessSessionMessageEventArgs args)
+                {
+                    tcs.SetResult(true);
                     throw new Exception();
                 }
 
+                Task ExceptionHandler(ProcessErrorEventArgs args)
+                {
+                    if (args.ErrorSource == ServiceBusErrorSource.ProcessMessageCallback)
+                    {
+                        throw new Exception();
+                    }
+
+                    return Task.CompletedTask;
+                }
+
                 processor.ProcessMessageAsync += ProcessMessage;
                 processor.ProcessErrorAsync += ExceptionHandler;
 
                 await processor.StartProcessingAsync();
                 await tcs.Task;
                 await processor.StopProcessingAsync();
-                _listener.SingleEventById(ServiceBusEventSource.ProcessorMessageHandlerStartEvent);
-                _listener.SingleEventById(ServiceBusEventSource.ProcessorMessageHandlerExceptionEvent);
-                _listener.SingleEventById(ServiceBusEventSource.ProcessorErrorHandlerThrewExceptionEvent);
+                _listener.SingleEventById(ServiceBusEventSource.ProcessorMessageHandlerStartEvent, args => args.Payload.Contains(processor.Identifier));
+                _listener.SingleEventById(ServiceBusEventSource.ProcessorMessageHandlerExceptionEvent, args => args.Payload.Contains(processor.Identifier));
+                _listener.SingleEventById(ServiceBusEventSource.ProcessorErrorHandlerThrewExceptionEvent, args => args.Payload.Contains(processor.Identifier));
             }
         }
 
@@ -366,7 +410,7 @@ public async Task LogsProcessorClientClosedExceptionEvent()
                     await Task.Delay(500, cancellationSource.Token);
                 }
 
-                _listener.SingleEventById(ServiceBusEventSource.ProcessorClientClosedExceptionEvent);
+                _listener.SingleEventById(ServiceBusEventSource.ProcessorClientClosedExceptionEvent, args => args.Payload.Contains(processor.Identifier));
             }
         }
 
@@ -375,22 +419,26 @@ public async Task DoesNotLogAcceptSessionTimeoutAsError()
         {
             await using (var scope = await ServiceBusScope.CreateWithQueue(enablePartitioning: false, enableSession: true))
             {
-                await using var client = CreateNoRetryClient(5);
+                await using var client = CreateClient(tryTimeout: 5, maxRetries: 1);
                 await using var processor = client.CreateSessionProcessor(scope.QueueName);
 
                 processor.ProcessMessageAsync += args => Task.CompletedTask;
                 processor.ProcessErrorAsync += args => Task.CompletedTask;
 
                 await processor.StartProcessingAsync();
 
-                // wait twice as long as the try timeout to ensure that the Accept session will timeout
-                await Task.Delay(TimeSpan.FromSeconds(10));
+                // wait long enough to ensure that the Accept session will timeout accounting for retries
+                await Task.Delay(TimeSpan.FromSeconds(20));
 
                 await processor.StopProcessingAsync();
 
-                Assert.False(_listener.EventsById(ServiceBusEventSource.CreateReceiveLinkExceptionEvent).Any());
-                Assert.False(_listener.EventsById(ServiceBusEventSource.ClientCreateExceptionEvent).Any());
-                Assert.True(_listener.EventsById(ServiceBusEventSource.ProcessorAcceptSessionTimeoutEvent).Any(e => e.Level == EventLevel.Verbose));
+                bool ContainsEntityPath(EventWrittenEventArgs args) => args.Payload.Contains(processor.EntityPath);
+
+                Assert.False(_listener.EventsById(ServiceBusEventSource.CreateReceiveLinkExceptionEvent).Any(ContainsEntityPath));
+                Assert.False(_listener.EventsById(ServiceBusEventSource.ClientCreateExceptionEvent).Any(ContainsEntityPath));
+                Assert.True(_listener.EventsById(ServiceBusEventSource.ProcessorAcceptSessionTimeoutEvent).Any(e => e.Level == EventLevel.Verbose && ContainsEntityPath(e)));
+                Assert.True(_listener.EventsById(ServiceBusEventSource.RunOperationExceptionVerboseEvent).Any(e => e.Level == EventLevel.Verbose));
+                Assert.False(_listener.EventsById(ServiceBusEventSource.RunOperationExceptionEvent).Any(ContainsEntityPath));
             }
         }
 
@@ -413,8 +461,9 @@ public async Task DoesNotLogStoppingAcceptSessionCanceledAsError()
                 await processor.StopProcessingAsync();
 
                 Assert.False(_listener.EventsById(ServiceBusEventSource.CreateReceiveLinkExceptionEvent).Any());
-                Assert.False(_listener.EventsById(ServiceBusEventSource.ClientCreateExceptionEvent).Any());
-                Assert.True(_listener.EventsById(ServiceBusEventSource.ProcessorStoppingAcceptSessionCanceledEvent).Any(e => e.Level == EventLevel.Verbose));
+                Assert.False(_listener.EventsById(ServiceBusEventSource.ClientCreateExceptionEvent).Any(e => e.Payload.Contains(processor.EntityPath)));
+                Assert.True(_listener.EventsById(ServiceBusEventSource.ProcessorStoppingAcceptSessionCanceledEvent)
+                    .Any(e => e.Level == EventLevel.Verbose));
             }
         }
 
@@ -437,7 +486,7 @@ public async Task StoppingProcessorDoesNotLogTaskCanceledExceptions()
                 await processor.StopProcessingAsync();
 
                 Assert.False(_listener.EventsById(ServiceBusEventSource.CreateReceiveLinkExceptionEvent).Any());
-                Assert.False(_listener.EventsById(ServiceBusEventSource.ClientCreateExceptionEvent).Any());
+                Assert.False(_listener.EventsById(ServiceBusEventSource.ClientCreateExceptionEvent).Any(e => e.Payload.Contains(processor.EntityPath)));
                 Assert.True(_listener.EventsById(ServiceBusEventSource.ProcessorStoppingReceiveCanceledEvent).Any());
             }
         }
@@ -467,7 +516,7 @@ public async Task StoppingSessionProcessorDoesNotLogTaskCanceledExceptions()
                 await processor.StopProcessingAsync();
 
                 Assert.False(_listener.EventsById(ServiceBusEventSource.CreateReceiveLinkExceptionEvent).Any());
-                Assert.False(_listener.EventsById(ServiceBusEventSource.ClientCreateExceptionEvent).Any());
+                Assert.False(_listener.EventsById(ServiceBusEventSource.ClientCreateExceptionEvent).Any(e => e.Payload.Contains(processor.EntityPath)));
                 Assert.True(_listener.EventsById(ServiceBusEventSource.ProcessorStoppingReceiveCanceledEvent).Any());
             }
         }
diff --git a/sdk/servicebus/Azure.Messaging.ServiceBus/tests/Infrastructure/ServiceBusLiveTestBase.cs b/sdk/servicebus/Azure.Messaging.ServiceBus/tests/Infrastructure/ServiceBusLiveTestBase.cs
@@ -18,30 +18,17 @@ public abstract class ServiceBusLiveTestBase : LiveTestBase<ServiceBusTestEnviro
 
         protected TimeSpan ShortLockDuration = TimeSpan.FromSeconds(10);
 
-        protected ServiceBusClient CreateNoRetryClient(int tryTimeout = DefaultTryTimeout)
-        {
-            var options =
-                new ServiceBusClientOptions
-                {
-                    RetryOptions = new ServiceBusRetryOptions
-                    {
-                        TryTimeout = TimeSpan.FromSeconds(tryTimeout),
-                        MaxRetries = 0
-                    }
-                };
-            return new ServiceBusClient(
-                TestEnvironment.ServiceBusConnectionString,
-                options);
-        }
+        protected ServiceBusClient CreateNoRetryClient(int tryTimeout = DefaultTryTimeout) => CreateClient(tryTimeout, 0);
 
-        protected ServiceBusClient CreateClient(int tryTimeout = DefaultTryTimeout)
+        protected ServiceBusClient CreateClient(int tryTimeout = DefaultTryTimeout, int maxRetries = 3)
         {
             var options =
                 new ServiceBusClientOptions
                 {
                     RetryOptions = new ServiceBusRetryOptions
                     {
                         TryTimeout = TimeSpan.FromSeconds(tryTimeout),
+                        MaxRetries = maxRetries
                     }
                 };
             return new ServiceBusClient(

Original file line number	Diff line number	Diff line change
`@@ -182,7 +182,7 @@ public int MaxDeliveryCount`
`182`	`182`	`Argument.AssertAtLeast(`
`183`	`183`	`value,`
`184`	`184`	`AdministrationClientConstants.MinAllowedMaxDeliveryCount,`
`185`		`- nameof(AutoDeleteOnIdle));`
	`185`	`+ nameof(MaxDeliveryCount));`
`186`	`186`
`187`	`187`	`_maxDeliveryCount = value;`
`188`	`188`	`}`
Original file line number	Diff line number	Diff line change
`@@ -194,6 +194,8 @@ protected ServiceBusEventSource() : base(EventSourceName)`
`194`	`194`
`195`	`195`	`internal const int ProcessorStoppingCancellationWarningEvent = 113;`
`196`	`196`
	`197`	`+ internal const int RunOperationExceptionVerboseEvent = 114;`
	`198`	`+`
`197`	`199`	`#endregion`
`198`	`200`	`// add new event numbers here incrementing from previous`
`199`	`201`
`@@ -844,12 +846,12 @@ public virtual void ProcessorRenewSessionLockException(string identifier, string`
`844`	`846`	`}`
`845`	`847`	`}`
`846`	`848`
`847`		`- [Event(ProcessorErrorHandlerThrewExceptionEvent, Level = EventLevel.Error, Message = "ExceptionReceivedHandler threw exception. Exception:{0}")]`
`848`		`- public void ProcessorErrorHandlerThrewException(string exception)`
	`849`	`+ [Event(ProcessorErrorHandlerThrewExceptionEvent, Level = EventLevel.Error, Message = "{1}: ExceptionReceivedHandler threw exception. Exception:{0}")]`
	`850`	`+ public void ProcessorErrorHandlerThrewException(string exception, string identifier)`
`849`	`851`	`{`
`850`	`852`	`if (IsEnabled())`
`851`	`853`	`{`
`852`		`- WriteEvent(ProcessorErrorHandlerThrewExceptionEvent, exception);`
	`854`	`+ WriteEvent(ProcessorErrorHandlerThrewExceptionEvent, exception, identifier);`
`853`	`855`	`}`
`854`	`856`	`}`
`855`	`857`
`@@ -1357,14 +1359,23 @@ public virtual void RequestAuthorizationException(string identifier, string endp`
`1357`	`1359`
`1358`	`1360`	`#region Retries`
`1359`	`1361`
`1360`		`- [Event(RunOperationExceptionEvent, Level = EventLevel.Informational, Message = "RunOperation encountered an exception and will retry. Exception: {0}")]`
	`1362`	`+ [Event(RunOperationExceptionEvent, Level = EventLevel.Warning, Message = "RunOperation encountered an exception and will retry. Exception: {0}")]`
`1361`	`1363`	`public virtual void RunOperationExceptionEncountered(string exception)`
`1362`	`1364`	`{`
`1363`	`1365`	`if (IsEnabled())`
`1364`	`1366`	`{`
`1365`	`1367`	`WriteEvent(RunOperationExceptionEvent, exception);`
`1366`	`1368`	`}`
`1367`	`1369`	`}`
	`1370`	`+`
	`1371`	`+ [Event(RunOperationExceptionVerboseEvent, Level = EventLevel.Verbose, Message = "RunOperation encountered an exception and will retry. Exception: {0}")]`
	`1372`	`+ public virtual void RunOperationExceptionEncounteredVerbose(string exception)`
	`1373`	`+ {`
	`1374`	`+ if (IsEnabled())`
	`1375`	`+ {`
	`1376`	`+ WriteEvent(RunOperationExceptionVerboseEvent, exception);`
	`1377`	`+ }`
	`1378`	`+ }`
`1368`	`1379`	`#endregion`
`1369`	`1380`
`1370`	`1381`	`#region Client lifecycle`
Original file line number	Diff line number	Diff line change
`@@ -346,7 +346,7 @@ protected virtual async Task RaiseExceptionReceived(ProcessErrorEventArgs eventA`
`346`	`346`	`catch (Exception exception)`
`347`	`347`	`{`
`348`	`348`	`// don't bubble up exceptions raised from customer exception handler`
`349`		`- ServiceBusEventSource.Log.ProcessorErrorHandlerThrewException(exception.ToString());`
	`349`	`+ ServiceBusEventSource.Log.ProcessorErrorHandlerThrewException(exception.ToString(), Processor.Identifier);`
`350`	`350`	`}`
`351`	`351`	`}`
`352`	`352`
Original file line number	Diff line number	Diff line change
`@@ -892,7 +892,7 @@ await OnProcessErrorAsync(`
`892`	`892`	`catch (Exception ex)`
`893`	`893`	`{`
`894`	`894`	`// Don't bubble up exceptions raised from customer exception handler.`
`895`		`- Logger.ProcessorErrorHandlerThrewException(ex.ToString());`
	`895`	`+ Logger.ProcessorErrorHandlerThrewException(ex.ToString(), Identifier);`
`896`	`896`	`}`
`897`	`897`
`898`	`898`	`// This call will deadlock if awaited, as StopProcessingAsync awaits`
Original file line number	Diff line number	Diff line change
`@@ -411,7 +411,7 @@ protected override async Task RaiseExceptionReceived(ProcessErrorEventArgs event`
`411`	`411`	`catch (Exception exception)`
`412`	`412`	`{`
`413`	`413`	`// don't bubble up exceptions raised from customer exception handler`
`414`		`- ServiceBusEventSource.Log.ProcessorErrorHandlerThrewException(exception.ToString());`
	`414`	`+ ServiceBusEventSource.Log.ProcessorErrorHandlerThrewException(exception.ToString(), Processor.Identifier);`
`415`	`415`	`}`
`416`	`416`	`}`
`417`	`417`