Mark test as inconclusive if doc publish does not start (#21859)

* Add asserts to verify all actions are counted for pending queue

* Increase timeout of test

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

@@ -364,6 +364,8 @@ public async Task Champion_BasicCheckpointing()
                         AutoFlushInterval = null
                     });
 
+            int removeFailedCount = 0;
+
             List<IndexDocumentsAction<SimpleDocument>> pending = new List<IndexDocumentsAction<SimpleDocument>>();
             indexer.ActionAdded +=
                 (IndexActionEventArgs<SimpleDocument> e) =>
@@ -374,25 +376,31 @@ public async Task Champion_BasicCheckpointing()
             indexer.ActionCompleted +=
                 (IndexActionCompletedEventArgs<SimpleDocument> e) =>
                 {
-                    pending.Remove(e.Action);
+                    if (!pending.Remove(e.Action))
+                    { removeFailedCount++; }
                     return Task.CompletedTask;
                 };
             indexer.ActionFailed +=
                 (IndexActionFailedEventArgs<SimpleDocument> e) =>
                 {
-                    pending.Remove(e.Action);
+                    if (!pending.Remove(e.Action))
+                    { removeFailedCount++; }
                     return Task.CompletedTask;
                 };
 
             await indexer.UploadDocumentsAsync(data.Take(500));
             await indexer.MergeDocumentsAsync(new[] { new SimpleDocument { Id = "Fake" } });
             await indexer.UploadDocumentsAsync(data.Skip(500));
 
-            await DelayAsync(TimeSpan.FromSeconds(5), TimeSpan.FromMilliseconds(250));
-            Assert.AreEqual(1001 - BatchSize, pending.Count);
+            int expectedPendingQueueSize = 1001 - BatchSize;
+
+            await ConditionallyDelayAsync(() => (pending.Count == expectedPendingQueueSize), TimeSpan.FromSeconds(2), TimeSpan.FromMilliseconds(250), 5);
+            Assert.AreEqual(expectedPendingQueueSize, pending.Count);
+            Assert.AreEqual(0, removeFailedCount);
 
             await indexer.FlushAsync();
             Assert.AreEqual(0, pending.Count);
+            Assert.AreEqual(0, removeFailedCount);
         }
         #endregion
 

sdk/search/Azure.Search.Documents/tests/Utilities/SearchTestBase.cs

@@ -107,6 +107,54 @@ public async Task DelayAsync(TimeSpan? delay = null, TimeSpan? playbackDelay = n
             }
         }
 
+        /// <summary>
+        /// A number of our tests have built in delays while we wait an expected
+        /// amount of time for a service operation to complete and this method
+        /// allows us to wait (unless we're playing back recordings, which can
+        /// complete immediately).
+        /// <para>This method allows us to return early if the <paramref name="predicate"/> condition evaluates to true.
+        /// It evaluates the predicate after every <paramref name="delayPerIteration"/> or <paramref name="playbackDelayPerIteration"/> time span.
+        /// It returns when the condition evaluates to <c>true</c>, or if the condition stays false after <paramref name="maxIterations"/> checks.</para>
+        /// </summary>
+        /// <param name="predicate">Condition that will result in early end of delay when it evaluates to <c>true</c>.</param>
+        /// <param name="delayPerIteration">The time to wait per iteration.  Defaults to 1s.</param>
+        /// <param name="playbackDelayPerIteration">
+        /// An optional time wait if we're playing back a recorded test.  This
+        /// is useful for allowing client side events to get processed.
+        /// </param>
+        /// <param name="maxIterations">Maximum number of iterations of the wait-and-check cycle.</param>
+        /// <param name="cancellationToken">Optional <see cref="CancellationToken"/> to check.</param>
+        /// <returns>A task that will (optionally) delay.</returns>
+        /// <exception cref="OperationCanceledException">The <paramref name="cancellationToken"/> was signaled.</exception>
+        public async Task ConditionallyDelayAsync(Func<bool> predicate, TimeSpan ? delayPerIteration = null, TimeSpan? playbackDelayPerIteration = null, uint maxIterations = 1, CancellationToken cancellationToken = default)
+        {
+            TimeSpan waitPeriod = TimeSpan.Zero;
+
+            for (int i = 0; i < maxIterations; i++)
+            {
+                if (predicate())
+                {
+                    TestContext.WriteLine($"{nameof(ConditionallyDelayAsync)}: Condition evaluated to true in {waitPeriod.TotalSeconds} seconds.");
+                    return;
+                }
+
+                cancellationToken.ThrowIfCancellationRequested();
+
+                if (Mode != RecordedTestMode.Playback)
+                {
+                    waitPeriod += delayPerIteration ?? TimeSpan.FromSeconds(1);
+                    await Task.Delay(delayPerIteration ?? TimeSpan.FromSeconds(1));
+                }
+                else if (playbackDelayPerIteration != null)
+                {
+                    waitPeriod += playbackDelayPerIteration.Value;
+                    await Task.Delay(playbackDelayPerIteration.Value);
+                }
+            }
+
+            TestContext.WriteLine($"{nameof(ConditionallyDelayAsync)}: Condition did not evaluate to true in {waitPeriod.TotalSeconds} seconds.");
+        }
+
         /// <summary>
         /// Assert that we can catch the desired exception.  NUnit's default
         /// forces everything to be sync.