BulkObservable should look at item.IsValid not response.IsValid

This also feeds the failing documents through the callback whether we
continue afterwards or not

Author: Mpdreamz

src/Nest/Document/Multiple/Bulk/BulkResponseItem/BulkResponseItemBase.cs

@@ -34,7 +34,7 @@ public bool IsValid
 				switch (this.Operation.ToLowerInvariant())
 				{
 					case "delete": return this.Status == 200 || this.Status == 404;
-					case "update": 
+					case "update":
 					case "index":
 					case "create":
 						return this.Status == 200 || this.Status == 201;
@@ -44,6 +44,7 @@ public bool IsValid
 			}
 		}
 
-		public override string ToString() => $"{Operation} returned {Status} _index: {Index} _type: {Type} _id: {Id} _version: {Version} error: {Error}";
+		public override string ToString() =>
+			$"{Operation} returned {Status} _index: {Index} _type: {Type} _id: {Id} _version: {Version} error: {Error}";
 	}
-}
+}

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

@@ -124,7 +124,7 @@ private async Task<IBulkAllResponse> BulkAsync(IList<T> buffer, long page, int b
 			this.HandleDroppedDocuments(documentsWithResponse, response);
 
 			var retryDocuments = documentsWithResponse
-				.Where(x=> !response.IsValid && this._retryPredicate(x.Item1, x.Item2))
+				.Where(x=> !x.Item1.IsValid && this._retryPredicate(x.Item1, x.Item2))
 				.Select(x => x.Item2)
 				.ToList();
 
@@ -140,14 +140,12 @@ private async Task<IBulkAllResponse> BulkAsync(IList<T> buffer, long page, int b
 		private void HandleDroppedDocuments(List<Tuple<BulkResponseItemBase, T>> documentsWithResponse, IBulkResponse response)
 		{
 			var droppedDocuments = documentsWithResponse
-				.Where(x => !response.IsValid && !this._retryPredicate(x.Item1, x.Item2))
+				.Where(x => !x.Item1.IsValid && !this._retryPredicate(x.Item1, x.Item2))
 				.ToList();
-			if (droppedDocuments.Count > 0 && !this._partionedBulkRequest.ContinueAfterDroppedDocuments)
+			if (droppedDocuments.Count <= 0) return;
+			foreach (var dropped in droppedDocuments) this._droppedDocumentCallBack(dropped.Item1, dropped.Item2);
+			if (!this._partionedBulkRequest.ContinueAfterDroppedDocuments)
 				throw this.ThrowOnBadBulk(response, $"BulkAll halted after receiving failures that can not be retried from _bulk");
-			else if (droppedDocuments.Count > 0 && this._partionedBulkRequest.ContinueAfterDroppedDocuments)
-			{
-				foreach (var dropped in droppedDocuments) this._droppedDocumentCallBack(dropped.Item1, dropped.Item2);
-			}
 		}
 
 		private async Task<IBulkAllResponse> HandleBulkRequest(IList<T> buffer, long page, int backOffRetries, IBulkResponse response)

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

@@ -81,9 +81,9 @@ public interface IBulkAllRequest<T> where T : class
 		bool ContinueAfterDroppedDocuments { get; set; }
 
 		/// <summary>
-		/// If <see cref="ContinueAfterDroppedDocuments"/> is set to true dropped messages will be fed through
-		/// this callback. Use this if you don't expect many failures and want to feed these dropped messages in a dead letter queue
-		/// for instance.
+		/// If a bulk operation fails because it receives documents it can not retry they will be fed to this callback.
+		/// If <see cref="ContinueAfterDroppedDocuments"/> is set to true processing will continue, so this callback can be used
+		/// to feed into a dead letter queue. Otherwise the bulk all indexation will be halted.
 		/// </summary>
 		Action<BulkResponseItemBase, T> DroppedDocumentCallback { get; set; }
 

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

@@ -56,22 +56,26 @@ private void ScrollAll(string index, int size, int numberOfShards, int numberOfD
 		private void BulkAll(string index, IEnumerable<SmallObject> documents, int size,  int pages, int numberOfDocuments)
 		{
 			var seenPages = 0;
+
+			var droppedDocuments = new ConcurrentBag<Tuple<BulkResponseItemBase,SmallObject>>();
 			//first we setup our cold observable
 			var observableBulk = this.Client.BulkAll(documents, f => f
 				.MaxDegreeOfParallelism(8)
 				.BackOffTime(TimeSpan.FromSeconds(10))
 				.BackOffRetries(2)
+				.DroppedDocumentCallback((b, i) => droppedDocuments.Add(Tuple.Create(b, i)))
 				.Size(size)
 				.RefreshOnCompleted()
 				.Index(index)
 			);
 			//we set up an observer
 			var bulkObserver = observableBulk.Wait(TimeSpan.FromMinutes(5), b => Interlocked.Increment(ref seenPages));
 
-			seenPages.Should().Be(pages);
+			droppedDocuments.Take(10).Should().BeEmpty();
+			bulkObserver.TotalNumberOfFailedBuffers.Should().Be(0, "All buffers are expected to be indexed");
+			seenPages.Should().Be(pages, "BulkAll() did not run to completion");
 			var count = this.Client.Count<SmallObject>(f => f.Index(index));
-			count.Count.Should().Be(numberOfDocuments);
-			bulkObserver.TotalNumberOfFailedBuffers.Should().Be(0);
+			count.Count.Should().Be(numberOfDocuments, "Target index should have the same document count as source index");
 		}
 	}
 }