Fix #3954 make sure BulkAll() is aborted properly if the whole bulk request keeps returning a bad statuscode (#4014)

Ensure BulkAll() is aborted properly if the whole bulk request keeps returning a bad statuscode.

(cherry picked from commit c0293a0)

Author: Mpdreamz

src/Nest/CommonAbstractions/Extensions/Extensions.cs

@@ -5,10 +5,12 @@
 using System.Globalization;
 using System.Linq;
 using System.Reflection;
+using System.Runtime.ExceptionServices;
 using System.Runtime.Serialization;
 using System.Text;
 using System.Threading;
 using System.Threading.Tasks;
+using Elasticsearch.Net;
 using Newtonsoft.Json;
 using Newtonsoft.Json.Converters;
 using Newtonsoft.Json.Linq;
@@ -24,9 +26,9 @@ internal static class EmptyReadOnly<TKey, TValue>
 	{
 		public static readonly IReadOnlyDictionary<TKey, TValue> Dictionary = new ReadOnlyDictionary<TKey, TValue>(new Dictionary<TKey, TValue>(0));
 	}
-
 	internal static class Extensions
 	{
+
 		internal static bool NotWritable(this QueryContainer q) => q == null || !q.IsWritable;
 
 		internal static bool NotWritable(this IEnumerable<QueryContainer> qs) => qs == null || qs.All(q => q.NotWritable());
@@ -111,13 +113,13 @@ internal static string ToEnumValue<T>(this T enumValue) where T : struct
 			return null;
 		}
 
-#if !DOTNETCORE
-		internal static string Utf8String(this byte[] bytes) => bytes == null ? null : Encoding.UTF8.GetString(bytes);
-#else
-		internal static string Utf8String(this byte[] bytes) => bytes == null ? null : Encoding.UTF8.GetString(bytes, 0, bytes.Length);
-#endif
+        #if !DOTNETCORE
+				internal static string Utf8String(this byte[] bytes) => bytes == null ? null : Encoding.UTF8.GetString(bytes);
+        #else
+				internal static string Utf8String(this byte[] bytes) => bytes == null ? null : Encoding.UTF8.GetString(bytes, 0, bytes.Length);
+        #endif
 
-		internal static byte[] Utf8Bytes(this string s) => s.IsNullOrEmpty() ? null : Encoding.UTF8.GetBytes(s);
+				internal static byte[] Utf8Bytes(this string s) => s.IsNullOrEmpty() ? null : Encoding.UTF8.GetBytes(s);
 
 		internal static bool IsNullOrEmpty(this TypeName value) => value == null || value.GetHashCode() == 0;
 
@@ -174,7 +176,7 @@ internal static bool IsEmpty<T>(this IEnumerable<T> list)
 			return !enumerable.Any() || enumerable.All(t => t == null);
 		}
 
-		internal static void ThrowIfNull<T>(this T value, string name, string message = null)
+		internal static void ThrowIfNull<T>(this T value, string name, string message = null) 
 		{
 			if (value == null && message.IsNullOrEmpty()) throw new ArgumentNullException(name);
 			else if (value == null) throw new ArgumentNullException(name, "Argument can not be null when " + message);
@@ -250,6 +252,12 @@ internal static async Task ForEachAsync<TSource, TResult>(
 						continue;
 
 					var task = await Task.WhenAny(tasks).ConfigureAwait(false);
+					if (task.Exception != null
+						&& (task.IsFaulted && task.Exception.Flatten().InnerExceptions.First() is Exception e))
+					{
+						ExceptionDispatchInfo.Capture(e).Throw();
+						return;
+					}
 					tasks.Remove(task);
 					i++;
 				}

src/Nest/Document/Multiple/BulkAll/BulkAllObservable.cs

@@ -20,6 +20,7 @@ public class BulkAllObservable<T> : IDisposable, IObservable<IBulkAllResponse> w
 		private readonly int _maxDegreeOfParallelism;
 		private readonly IBulkAllRequest<T> _partitionedBulkRequest;
 		private readonly Func<IBulkResponseItem, T, bool> _retryPredicate;
+		private readonly Action<IBulkResponse> _bulkResponseCallback;
 		private Action _incrementFailed = () => { };
 		private Action _incrementRetries = () => { };
 
@@ -36,6 +37,8 @@ public BulkAllObservable(
 			_bulkSize = _partitionedBulkRequest.Size ?? CoordinatedRequestDefaults.BulkAllSizeDefault;
 			_retryPredicate = _partitionedBulkRequest.RetryDocumentPredicate ?? RetryBulkActionPredicate;
 			_droppedDocumentCallBack = _partitionedBulkRequest.DroppedDocumentCallback ?? DroppedDocumentCallbackDefault;
+			_bulkResponseCallback = _partitionedBulkRequest.BulkResponseCallback;
+
 			_maxDegreeOfParallelism =
 				_partitionedBulkRequest.MaxDegreeOfParallelism ?? CoordinatedRequestDefaults.BulkAllMaxDegreeOfParallelismDefault;
 			_compositeCancelTokenSource = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken);
@@ -132,6 +135,8 @@ private async Task<IBulkAllResponse> BulkAsync(IList<T> buffer, long page, int b
 
 			_compositeCancelToken.ThrowIfCancellationRequested();
 
+			_bulkResponseCallback?.Invoke(response);
+
 			if (!response.ApiCall.Success)
 				return await HandleBulkRequest(buffer, page, backOffRetries, response);
 
@@ -177,26 +182,36 @@ private void HandleDroppedDocuments(List<Tuple<IBulkResponseItem, T>> droppedDoc
 		private async Task<IBulkAllResponse> HandleBulkRequest(IList<T> buffer, long page, int backOffRetries, IBulkResponse response)
 		{
 			var clientException = response.ApiCall.OriginalException as ElasticsearchClientException;
-			//TODO expose this on IAPiCallDetails as RetryLater in 7.0?
-			var failureReason = clientException?.FailureReason.GetValueOrDefault(PipelineFailure.Unexpected);
+			var failureReason = clientException?.FailureReason;
+			var reason = failureReason?.GetStringValue() ?? nameof(PipelineFailure.BadRequest);
 			switch (failureReason)
 			{
 				case PipelineFailure.MaxRetriesReached:
 					//TODO move this to its own PipelineFailure classification in 7.0
 					if (response.ApiCall.AuditTrail.Last().Event == AuditEvent.FailedOverAllNodes)
 						throw ThrowOnBadBulk(response, $"BulkAll halted after attempted bulk failed over all the active nodes");
 
-					return await RetryDocuments(page, ++backOffRetries, buffer);
+					ThrowOnExhaustedRetries();
+					return await RetryDocuments(page, ++backOffRetries, buffer).ConfigureAwait(false);
 				case PipelineFailure.CouldNotStartSniffOnStartup:
 				case PipelineFailure.BadAuthentication:
 				case PipelineFailure.NoNodesAttempted:
 				case PipelineFailure.SniffFailure:
 				case PipelineFailure.Unexpected:
-					throw ThrowOnBadBulk(response,
-						$"BulkAll halted after {nameof(PipelineFailure)}{failureReason.GetStringValue()} from _bulk");
+					throw ThrowOnBadBulk(response, $"BulkAll halted after {nameof(PipelineFailure)}.{reason} from _bulk");
 				default:
+					ThrowOnExhaustedRetries();
 					return await RetryDocuments(page, ++backOffRetries, buffer).ConfigureAwait(false);
 			}
+
+			void ThrowOnExhaustedRetries()
+			{
+				if (_partitionedBulkRequest.ContinueAfterDroppedDocuments || backOffRetries < _backOffRetries) return;
+
+				throw ThrowOnBadBulk(response,
+					$"BulkAll halted after {nameof(PipelineFailure)}.{reason} from _bulk and exhausting retries ({backOffRetries})"
+				);
+			}
 		}
 
 		private async Task<IBulkAllResponse> RetryDocuments(long page, int backOffRetries, IList<T> retryDocuments)

src/Nest/Document/Multiple/BulkAll/BulkAllRequest.cs

@@ -103,6 +103,12 @@ public interface IBulkAllRequest<T> where T : class
 		/// non-negative value less than or equal to the total number of copies for the shard (number of replicas + 1)
 		/// </summary>
 		int? WaitForActiveShards { get; set; }
+
+		/// <summary>
+		/// Be notified every time a bulk response returns, this includes retries.
+		/// <see cref="IObserver{T}.OnNext"/> is only called for successful batches.
+		/// </summary>
+		Action<IBulkResponse> BulkResponseCallback { get; set; }
 	}
 
 	public class BulkAllRequest<T> : IBulkAllRequest<T>
@@ -171,6 +177,9 @@ public BulkAllRequest(IEnumerable<T> documents)
 
 		/// <inheritdoc />
 		public int? WaitForActiveShards { get; set; }
+
+		/// <inheritdoc />
+		public Action<IBulkResponse> BulkResponseCallback { get; set; }
 	}
 
 	public class BulkAllDescriptor<T> : DescriptorBase<BulkAllDescriptor<T>, IBulkAllRequest<T>>, IBulkAllRequest<T>
@@ -205,6 +214,7 @@ public BulkAllDescriptor(IEnumerable<T> documents)
 		Time IBulkAllRequest<T>.Timeout { get; set; }
 		TypeName IBulkAllRequest<T>.Type { get; set; }
 		int? IBulkAllRequest<T>.WaitForActiveShards { get; set; }
+		Action<IBulkResponse> IBulkAllRequest<T>.BulkResponseCallback { get; set; }
 
 		/// <inheritdoc cref="IBulkAllRequest{T}.MaxDegreeOfParallelism" />
 		public BulkAllDescriptor<T> MaxDegreeOfParallelism(int? parallelism) =>
@@ -285,5 +295,9 @@ public BulkAllDescriptor<T> ContinueAfterDroppedDocuments(bool proceed = true) =
 		/// <inheritdoc cref="IBulkAllRequest{T}.DroppedDocumentCallback" />
 		public BulkAllDescriptor<T> DroppedDocumentCallback(Action<IBulkResponseItem, T> callback) =>
 			Assign(callback, (a, v) => a.DroppedDocumentCallback = v);
+
+		/// <inheritdoc cref="IBulkAllRequest{T}.BulkResponseCallback" />
+		public BulkAllDescriptor<T> BulkResponseCallback(Action<IBulkResponse> callback) =>
+			Assign(callback, (a, v) => a.BulkResponseCallback = v);
 	}
 }

src/Tests/Tests/Document/Multiple/BulkAll/BulkAllExceptionApiTests.cs

@@ -1,6 +1,7 @@
 using System;
 using System.Threading;
 using Elastic.Xunit.XunitPlumbing;
+using Elasticsearch.Net;
 using FluentAssertions;
 using Nest;
 using Tests.Core.ManagedElasticsearch.Clusters;
@@ -49,4 +50,64 @@ [I] public void WaitBulkAllThrowsAndIsCaught()
 			}
 		}
 	}
+
+
+	public class BulkAllBadRetriesApiTests : BulkAllApiTestsBase
+	{
+		public BulkAllBadRetriesApiTests(IntrusiveOperationCluster cluster, EndpointUsage usage) : base(cluster, usage) { }
+
+		[U] public void Completes()
+		{
+			var client = Tests.Framework.Cluster.Nodes(2)
+				.ClientCalls(c => c.FailAlways())
+				.StaticConnectionPool()
+				.AllDefaults()
+				.Client;
+
+			var index = CreateIndexName();
+
+			var size = 1000;
+			var pages = 10;
+			var seenPages = 0;
+			var numberOfDocuments = size * pages;
+			var documents = CreateLazyStreamOfDocuments(numberOfDocuments);
+			var requests = 0;
+
+			Exception ex = null;
+			var tokenSource = new CancellationTokenSource();
+			var observableBulk = client.BulkAll(documents, f => f
+					.MaxDegreeOfParallelism(1)
+					.BulkResponseCallback(r => Interlocked.Increment(ref requests))
+					.BackOffTime(TimeSpan.FromMilliseconds(1))
+					.BackOffRetries(2)
+					.Size(size)
+					.RefreshOnCompleted()
+					.Index(index)
+					.BufferToBulk((r, buffer) => r.IndexMany(buffer))
+				, tokenSource.Token);
+
+			try
+			{
+				observableBulk.Wait(TimeSpan.FromSeconds(30), b =>
+				{
+					Interlocked.Increment(ref seenPages);
+				});
+			}
+			catch (Exception e)
+			{
+				ex = e;
+			}
+			ex.Should().NotBeNull();
+
+			var clientException = ex.Should().BeOfType<ElasticsearchClientException>().Subject;
+
+			clientException.Message.Should()
+				.StartWith("BulkAll halted after")
+				.And.EndWith("from _bulk and exhausting retries (2)");
+
+			requests.Should().Be(3);
+			// OnNext only called for successful batches.
+			seenPages.Should().Be(0);
+		}
+	}
 }

src/Tests/Tests/Framework/VirtualClustering/VirtualizedCluster.cs

@@ -29,7 +29,7 @@ public VirtualizedCluster(TestableDateTimeProvider dateTimeProvider, ConnectionS
 		}
 
 		public IConnectionPool ConnectionPool => Client.ConnectionSettings.ConnectionPool;
-		private ElasticClient Client => _fixedRequestPipeline?.Client;
+		public ElasticClient Client => _fixedRequestPipeline?.Client;
 
 		public VirtualizedCluster ClientProxiesTo(
 			Func<IElasticClient, Func<RequestConfigurationDescriptor, IRequestConfiguration>, IResponse> sync,