Add configuration options to tune SearchIndexingBufferedSender<T>. (Azure#16695)

Add configuration options to tune SearchIndexingBufferedSender<T>

Also adds better behavior around swallowing notification errors and fixes Azure#16291 to make tests less flaky.

Author: tg-msft

sdk/search/Azure.Search.Documents/CHANGELOG.md

@@ -10,6 +10,7 @@
 ### Added
 
 - Added `EncryptionKey` to `SearchIndexer`, `SearchIndexerDataSourceConnection`, and `SearchIndexerSkillset`.
+- Add configuration options to tune the performance of `SearchIndexingBufferedSender<T>`.
 
 ## 11.2.0-beta.1 (2020-10-09)
 

sdk/search/Azure.Search.Documents/api/Azure.Search.Documents.netstandard2.0.cs

@@ -80,7 +80,11 @@ public SearchIndexingBufferedSenderOptions() { }
         public bool AutoFlush { get { throw null; } set { } }
         public System.TimeSpan? AutoFlushInterval { get { throw null; } set { } }
         public System.Threading.CancellationToken FlushCancellationToken { get { throw null; } set { } }
+        public int? InitialBatchActionCount { get { throw null; } set { } }
         public System.Func<T, string> KeyFieldAccessor { get { throw null; } set { } }
+        public int MaxRetries { get { throw null; } set { } }
+        public System.TimeSpan MaxRetryDelay { get { throw null; } set { } }
+        public System.TimeSpan RetryDelay { get { throw null; } set { } }
     }
     public partial class SearchIndexingBufferedSender<T> : System.IAsyncDisposable, System.IDisposable
     {

sdk/search/Azure.Search.Documents/src/Batching/Publisher{T}.cs

@@ -71,7 +71,7 @@ internal abstract partial class Publisher<T> : IDisposable, IAsyncDisposable
         /// Manual retry policy to add exponential back-off after throttled
         /// requests.
         /// </summary>
-        private ManualRetryDelay _manualRetries = new ManualRetryDelay();
+        private ManualRetryDelay _manualRetries;
 
         /// <summary>
         /// Gets the number of indexing actions currently awaiting submission.
@@ -104,7 +104,7 @@ public int IndexingActionsCount
         /// Gets a value indicating the number of actions to group into a batch
         /// when tuning the behavior of the publisher.
         /// </summary>
-        protected int BatchActionSize { get; }  // TODO: Not automatically tuning yet
+        protected int BatchActionCount { get; }  // TODO: Not automatically tuning yet
 
         /// <summary>
         /// Gets a value indicating the number of bytes to use when tuning the
@@ -115,7 +115,7 @@ public int IndexingActionsCount
         /// <summary>
         /// Gets the number of times to retry a failed document.
         /// </summary>
-        protected int RetryCount { get; }  // TODO: Not configurable yet
+        protected int MaxRetries { get; }
 
         /// <summary>
         /// Creates a new Publisher which immediately starts listening to
@@ -134,9 +134,16 @@ public int IndexingActionsCount
         /// The number of bytes to use when tuning the behavior of the
         /// publisher.
         /// </param>
-        /// <param name="retryCount">
+        /// <param name="maxRetries">
         /// The number of times to retry a failed document.
         /// </param>
+        /// <param name="retryDelay">
+        /// The initial retry delay on which to base calculations for a
+        /// backoff-based approach.
+        /// </param>
+        /// <param name="maxRetryDelay">
+        /// The maximum permissible delay between retry attempts.
+        /// </param>
         /// <param name="publisherCancellationToken">
         /// A <see cref="CancellationToken"/> to use when publishing.
         /// </param>
@@ -145,15 +152,20 @@ public Publisher(
             TimeSpan? autoFlushInterval,
             int? batchActionSize,
             int? batchPayloadSize,
-            int? retryCount,
+            int maxRetries,
+            TimeSpan retryDelay,
+            TimeSpan maxRetryDelay,
             CancellationToken publisherCancellationToken)
         {
             AutoFlush = autoFlush;
             AutoFlushInterval = autoFlushInterval <= TimeSpan.Zero ? null : autoFlushInterval;
             PublisherCancellationToken = publisherCancellationToken;
-            BatchActionSize = batchActionSize ?? SearchIndexingBufferedSenderOptions<T>.DefaultBatchActionSize;
+            BatchActionCount = batchActionSize ?? SearchIndexingBufferedSenderOptions<T>.DefaultInitialBatchActionCount;
             BatchPayloadSize = batchPayloadSize ?? SearchIndexingBufferedSenderOptions<T>.DefaultBatchPayloadSize;
-            RetryCount = retryCount ??  SearchIndexingBufferedSenderOptions<T>.DefaultRetryCount;
+            MaxRetries = maxRetries;
+
+            // Setup manual retries
+            _manualRetries = new ManualRetryDelay { Delay = retryDelay, MaxDelay = maxRetryDelay };
 
             // Start the message loop
             _readerLoop = Task.Run(ProcessMessagesAsync, publisherCancellationToken);
@@ -372,7 +384,7 @@ private void EnsureNotDisposed()
         /// <param name="flush">Whether we're flushing.</param>
         /// <returns>If we have a full batch ready to send.</returns>
         private bool HasBatch(bool flush = false) =>
-            IndexingActionsCount > (flush ? 0 : BatchActionSize);
+            IndexingActionsCount > (flush ? 0 : BatchActionCount);
 
         /// <summary>
         /// Publish as many batches are ready.
@@ -389,7 +401,7 @@ private async Task PublishAsync(bool flush, CancellationToken cancellationToken)
             do
             {
                 List<PublisherAction<T>> batch = new List<PublisherAction<T>>(
-                    capacity: Math.Min(BatchActionSize, IndexingActionsCount));
+                    capacity: Math.Min(BatchActionCount, IndexingActionsCount));
 
                 // Prefer pulling from the _retry queue first
                 if (!FillBatchFromQueue(batch, _retry))
@@ -415,7 +427,7 @@ bool FillBatchFromQueue(List<PublisherAction<T>> batch, Queue< PublisherAction<T
 
                 while (queue.Count > 0)
                 {
-                    if (batch.Count < BatchActionSize)
+                    if (batch.Count < BatchActionCount)
                     {
                         batch.Add(queue.Dequeue());
                     }
@@ -471,7 +483,7 @@ protected bool EnqueueRetry(
             PublisherAction<T> action,
             bool skipIncrement = false)
         {
-            bool retriable = skipIncrement || action.RetryAttempts++ < RetryCount;
+            bool retriable = skipIncrement || action.RetryAttempts++ < MaxRetries;
             if (retriable)
             {
                 _retry.Enqueue(action);

sdk/search/Azure.Search.Documents/src/Batching/SearchIndexingBufferedSenderOptions{T}.cs

@@ -40,12 +40,13 @@ public class SearchIndexingBufferedSenderOptions<T>
         public CancellationToken FlushCancellationToken { get; set; }
 
         /// <summary>
-        /// Gets or sets a value indicating the number of actions to group into
-        /// a batch when tuning the behavior of the sender.  The default value
-        /// is 512.  The current service maximum is 32000.
+        /// Gets or sets a value indicating the initial number of actions to
+        /// group into a batch when tuning the behavior of the sender.  The
+        /// default value will be 512 if unset.  The current service maximum is
+        /// 32000.
         /// </summary>
-        internal int? BatchActionSize { get; set; } = DefaultBatchActionSize;
-        internal const int DefaultBatchActionSize = 512;
+        public int? InitialBatchActionCount { get; set; } = null;
+        internal const int DefaultInitialBatchActionCount = 512;
 
         /// <summary>
         /// Gets or sets a value indicating the number of bytes to use when
@@ -56,10 +57,32 @@ public class SearchIndexingBufferedSenderOptions<T>
         internal const int DefaultBatchPayloadSize = 500 * 1024;
 
         /// <summary>
-        /// Gets or sets the number of times to retry a failed document.
+        /// Gets or sets the number of times to retry a failed document.  Note
+        /// that this is different than <see cref="Azure.Core.RetryOptions.MaxRetries"/>
+        /// which will try to resend the same request.  This property is used
+        /// to control the number of attempts we will make to submit an indexing
+        /// action.
         /// </summary>
-        internal int? RetryCount { get; set; } = DefaultRetryCount;
-        internal const int DefaultRetryCount = 3;
+        public int MaxRetries { get; set; } = 3;
+
+        /// <summary>
+        /// The initial retry delay. The delay will increase exponentially with
+        /// subsequent retries and add random jitter.  Note that this is
+        /// different than <see cref="Azure.Core.RetryOptions.Delay"/> which
+        /// will only delay before resending the same request.  This property
+        /// is used to add delay between additional batch submissions when our
+        /// requests are being throttled by the service.
+        /// </summary>
+        public TimeSpan RetryDelay { get; set; } = TimeSpan.FromSeconds(0.8);
+
+        /// <summary>
+        /// The maximum permissible delay between retry attempts.  Note that
+        /// this is different than <see cref="Azure.Core.RetryOptions.MaxDelay"/>
+        /// which will only delay before resending the same request.  This
+        /// property is used to add delay between additional batch
+        /// submissions when our requests are being throttled by the service.
+        /// </summary>
+        public TimeSpan MaxRetryDelay { get; set; } = TimeSpan.FromMinutes(1);
 
         /// <summary>
         /// Gets or sets a function that can be used to access the index key

sdk/search/Azure.Search.Documents/src/Batching/SearchIndexingBufferedSender{T}.cs

@@ -127,9 +127,11 @@ internal SearchIndexingBufferedSender(
                 this,
                 options.AutoFlush,
                 options.AutoFlushInterval,
-                options.BatchActionSize,
+                options.InitialBatchActionCount,
                 options.BatchPayloadSize,
-                options.RetryCount,
+                options.MaxRetries,
+                options.RetryDelay,
+                options.MaxRetryDelay,
                 options.FlushCancellationToken);
         }
 
@@ -331,10 +333,20 @@ static Func<T, string> CompileAccessor(string key)
         /// </returns>
         internal async Task RaiseActionAddedAsync(
             IndexDocumentsAction<T> action,
-            CancellationToken cancellationToken) =>
-            await ActionAddedAsync
-                .RaiseAsync(action, cancellationToken)
-                .ConfigureAwait(false);
+            CancellationToken cancellationToken)
+        {
+            try
+            {
+                await ActionAddedAsync
+                    .RaiseAsync(action, cancellationToken)
+                    .ConfigureAwait(false);
+            }
+            catch
+            {
+                // TODO: #16706 - Log any exceptions raised from async events
+                // we can't let bubble out because they'd tear down the process
+            }
+        }
 
         /// <summary>
         /// Raise the <see cref="ActionSentAsync"/> event.
@@ -346,10 +358,20 @@ await ActionAddedAsync
         /// </returns>
         internal async Task RaiseActionSentAsync(
             IndexDocumentsAction<T> action,
-            CancellationToken cancellationToken) =>
-            await ActionSentAsync
-                .RaiseAsync(action, cancellationToken)
-                .ConfigureAwait(false);
+            CancellationToken cancellationToken)
+        {
+            try
+            {
+                await ActionSentAsync
+                    .RaiseAsync(action, cancellationToken)
+                    .ConfigureAwait(false);
+            }
+            catch
+            {
+                // TODO: #16706 - Log any exceptions raised from async events
+                // we can't let bubble out because they'd tear down the process
+            }
+        }
 
         /// <summary>
         /// Raise the <see cref="ActionCompletedAsync"/> event.
@@ -363,10 +385,20 @@ await ActionSentAsync
         internal async Task RaiseActionCompletedAsync(
             IndexDocumentsAction<T> action,
             IndexingResult result,
-            CancellationToken cancellationToken) =>
-            await ActionCompletedAsync
-                .RaiseAsync(action, result, cancellationToken)
-                .ConfigureAwait(false);
+            CancellationToken cancellationToken)
+        {
+            try
+            {
+                await ActionCompletedAsync
+                    .RaiseAsync(action, result, cancellationToken)
+                    .ConfigureAwait(false);
+            }
+            catch
+            {
+                // TODO: #16706 - Log any exceptions raised from async events
+                // we can't let bubble out because they'd tear down the process
+            }
+        }
 
         /// <summary>
         /// Raise the <see cref="ActionFailedAsync"/> event.
@@ -382,10 +414,20 @@ internal async Task RaiseActionFailedAsync(
             IndexDocumentsAction<T> action,
             IndexingResult result,
             Exception exception,
-            CancellationToken cancellationToken) =>
-            await ActionFailedAsync
-                .RaiseAsync(action, result, exception, cancellationToken)
-                .ConfigureAwait(false);
+            CancellationToken cancellationToken)
+        {
+            try
+            {
+                await ActionFailedAsync
+                    .RaiseAsync(action, result, exception, cancellationToken)
+                    .ConfigureAwait(false);
+            }
+            catch
+            {
+                // TODO: #16706 - Log any exceptions raised from async events
+                // we can't let bubble out because they'd tear down the process
+            }
+        }
         #endregion
 
         #region Index Documents

sdk/search/Azure.Search.Documents/src/Batching/SearchIndexingPublisher{T}.cs

@@ -45,9 +45,16 @@ internal class SearchIndexingPublisher<T> : Publisher<IndexDocumentsAction<T>>
         /// The number of bytes to use when tuning the behavior of the
         /// publisher.
         /// </param>
-        /// <param name="retryCount">
+        /// <param name="maxRetries">
         /// The number of times to retry a failed document.
         /// </param>
+        /// <param name="retryDelay">
+        /// The initial retry delay on which to base calculations for a
+        /// backoff-based approach.
+        /// </param>
+        /// <param name="maxRetryDelay">
+        /// The maximum permissible delay between retry attempts.
+        /// </param>
         /// <param name="publisherCancellationToken">
         /// A <see cref="CancellationToken"/> to use when publishing.
         /// </param>
@@ -57,9 +64,19 @@ public SearchIndexingPublisher(
             TimeSpan? autoFlushInterval,
             int? batchActionSize,
             int? batchPayloadSize,
-            int? retryCount,
+            int maxRetries,
+            TimeSpan retryDelay,
+            TimeSpan maxRetryDelay,
             CancellationToken publisherCancellationToken)
-            : base(autoFlush, autoFlushInterval, batchActionSize, batchPayloadSize, retryCount, publisherCancellationToken)
+            : base(
+                autoFlush,
+                autoFlushInterval,
+                batchActionSize,
+                batchPayloadSize,
+                maxRetries,
+                retryDelay,
+                maxRetryDelay,
+                publisherCancellationToken)
         {
             _sender = sender;
         }

sdk/search/Azure.Search.Documents/tests/Batching/AsyncEventExtensionsTests.cs

@@ -369,7 +369,7 @@ public async Task Cancels_AlreadyFinished()
         [Test]
         public async Task Cancels_StillRunning()
         {
-            TestHandler test = new TestHandler() { Delay = TimeSpan.FromMilliseconds(100) };
+            TestHandler test = new TestHandler() { Delay = TimeSpan.FromMilliseconds(500) };
             CancellationTokenSource cancellation = new CancellationTokenSource();
 
             Func<EventArgs, CancellationToken, Task> handler = null;
@@ -391,9 +391,9 @@ await Task.WhenAll(
         [Test]
         public async Task Cancels_All()
         {
-            TestHandler first = new TestHandler() { Delay = TimeSpan.FromMilliseconds(100) };
-            TestHandler second = new TestHandler() { Delay = TimeSpan.FromMilliseconds(100) };
-            TestHandler third = new TestHandler() { Delay = TimeSpan.FromMilliseconds(100) };
+            TestHandler first = new TestHandler() { Delay = TimeSpan.FromMilliseconds(500) };
+            TestHandler second = new TestHandler() { Delay = TimeSpan.FromMilliseconds(500) };
+            TestHandler third = new TestHandler() { Delay = TimeSpan.FromMilliseconds(500) };
             CancellationTokenSource cancellation = new CancellationTokenSource();
 
             Func<EventArgs, CancellationToken, Task> handler = null;