Several remain source serializer fixes

* JoinField Deserialize needs to be special case because we can not pass
  a string directly to JsonSerializer.Deserialize, it won't pick up
  Converters in that case.

* Bulk and LazyDocument, did not play nicely together.

* The reindex tests were not setting the routing correctly

* I don't think CreateIndexRequest(name, state) should copy aliases from
  `state`. cc @codebrain @russcam

Author: Mpdreamz

src/Nest/CommonAbstractions/Extensions/Extensions.cs

@@ -244,6 +244,7 @@ internal static async Task ForEachAsync<TSource, TResult>(
 			try
 			{
 				var tasks = new List<Task>();
+				int i = 0;
 				foreach (var item in lazyList)
 				{
 					tasks.Add(ProcessAsync(item, taskSelector, resultProcessor, semaphore, additionalRateLimitter, page++));
@@ -252,6 +253,7 @@ internal static async Task ForEachAsync<TSource, TResult>(
 
 					var task = await Task.WhenAny(tasks);
 					tasks.Remove(task);
+					i++;
 				}
 
 				await Task.WhenAll(tasks);

src/Nest/CommonAbstractions/LazyDocument/LazyDocument.cs

@@ -2,12 +2,11 @@
 using System.IO;
 using System.Text;
 using Elasticsearch.Net;
-using Newtonsoft.Json;
 using Newtonsoft.Json.Linq;
 
 namespace Nest
 {
-	[JsonConverter(typeof(LazyDocumentJsonConverter))]
+	[ContractJsonConverter(typeof(LazyDocumentJsonConverter))]
 	public interface ILazyDocument
 	{
 		/// <summary>

src/Nest/CommonAbstractions/SerializationBehavior/JsonNetSerializer.cs

@@ -1,8 +1,5 @@
 using System;
-using System.Collections.Concurrent;
-using System.Collections.Generic;
 using System.IO;
-using System.Reflection;
 using System.Text;
 using System.Threading;
 using System.Threading.Tasks;

src/Nest/Document/Multiple/Bulk/BulkOperation/BulkIndex.cs

@@ -11,6 +11,7 @@ public interface IBulkIndexOperation<T> : IBulkOperation
 		[JsonProperty("pipeline")]
 		string Pipeline { get; set; }
 
+		[JsonConverter(typeof(SourceConverter))]
 		T Document { get; set; }
 	}
 

src/Nest/Document/Multiple/Bulk/BulkRequestJsonConverter.cs

@@ -30,7 +30,11 @@ public override void WriteJson(JsonWriter writer, object value, JsonSerializer s
 				writer.WriteRaw($"{{\"{op.Operation}\":" + opJson + "}\n");
 				var body = op.GetBody();
 				if (body == null) continue;
-				var bodyJson = (op.Operation == "update" ? requestResponseSerializer : sourceSerializer)
+				var bodyJson = (
+					op.Operation == "update" || body is ILazyDocument
+						? requestResponseSerializer
+						: sourceSerializer
+					)
 					.SerializeToString(body, SerializationFormatting.None);
 
 				writer.WriteRaw(bodyJson + "\n");

src/Nest/Document/Multiple/Reindex/ReindexObservable.cs

@@ -77,12 +77,10 @@ private void Reindex(IObserver<IBulkAllResponse> observer)
 				.SelectMany(r => r.SearchResponse.Hits)
 				.Select(r=> r.Copy(this._reindexRequest.Map));
 
-
 			var observableBulk = this.BulkAll(scrollDocuments, backPressure, toIndex);
 
 			//by casting the observable can potentially store more meta data on the user provided observer
-			var moreInfoObserver = observer as BulkAllObserver;
-			if (moreInfoObserver != null)
+			if (observer is BulkAllObserver moreInfoObserver)
 				observableBulk.Subscribe(moreInfoObserver);
 			else observableBulk.Subscribe(observer);
 		}
@@ -93,6 +91,7 @@ private BulkAllObservable<IHitMetadata<TTarget>> BulkAll(IEnumerable<IHitMetadat
 			if (bulkAllRequest == null)
 				throw new Exception("BulkAll must set on ReindexRequest in order to get the target of a Reindex operation");
 
+			bulkAllRequest.Type = null;
 			bulkAllRequest.BackPressure = backPressure;
 			bulkAllRequest.BufferToBulk = (bulk, hits) =>
 			{
@@ -219,12 +218,6 @@ private int CreateIndex(string toIndex, IScrollAllRequest scrollAll)
 				originalIndexState = getIndexResponse.Indices[resolvedFrom];
 				if (this._reindexRequest.OmitIndexCreation)
 					return originalIndexState.Settings.NumberOfShards;
-
-				// Black list internal settings that cannot be copied over
-				// See https://github.com/elastic/elasticsearch/issues/21096
-				originalIndexState.Settings.Remove("index.provided_name");
-				originalIndexState.Settings.Remove("index.creation_date");
-				originalIndexState.Settings.Remove("index.version.created");
 			}
 
 			var createIndexRequest = this._reindexRequest.CreateIndexRequest ??

src/Nest/Indices/IndexManagement/CreateIndex/CreateIndexRequest.cs

@@ -13,11 +13,10 @@ public partial class CreateIndexRequest
 		//Only here for ReadAsType new() constraint needs to be updated
 		internal CreateIndexRequest() { }
 
-		public CreateIndexRequest(IndexName index, IndexState state) : this(index)
+		public CreateIndexRequest(IndexName index, IIndexState state) : this(index)
 		{
 			this.Settings = state.Settings;
 			this.Mappings = state.Mappings;
-			this.Aliases = state.Aliases;
 			this.Similarity = state.Similarity;
 			CreateIndexRequest.RemoveReadOnlySettings(this.Settings);
 		}
@@ -27,6 +26,7 @@ public CreateIndexRequest(IndexName index, IndexState state) : this(index)
 			"index.creation_date",
 			"index.uuid",
 			"index.version.created",
+			"index.provided_name"
 		};
 
 		internal static void RemoveReadOnlySettings (IIndexSettings settings)

src/Nest/Mapping/Types/Core/Join/JoinFieldJsonConverter.cs

@@ -10,6 +10,7 @@ internal class JoinFieldJsonConverter :JsonConverter
 	{
 		public override bool CanRead => true;
 		public override bool CanWrite => true;
+		public override bool CanConvert(Type objectType) => true;
 
 		public override object ReadJson(JsonReader reader, Type objectType, object existingValue, JsonSerializer serializer)
 		{
@@ -47,6 +48,5 @@ public override void WriteJson(JsonWriter writer, object value, JsonSerializer s
 			);
 		}
 
-		public override bool CanConvert(Type objectType) => true;
 	}
 }

src/Serializers/Nest.JsonNetSerializer/Converters/RevertBackToBuiltinSerializer.cs

@@ -30,6 +30,12 @@ public override void WriteJson(JsonWriter writer, object value, JsonSerializer s
 		public override object ReadJson(JsonReader reader, Type objectType, object existingValue, JsonSerializer serializer)
 		{
 			var token = JToken.ReadFrom(reader);
+			//in place because JsonConverter.Deserialize() only works on full json objects.
+			//even though we pass type JSON.NET won't try the registered converter for that type
+			//even if it can handle string tokens :(
+			if (objectType == typeof(JoinField) && token.Type == JTokenType.String)
+				return JoinField.Root(token.ToString());
+
 			using (var ms = new MemoryStream(Encoding.UTF8.GetBytes(token.ToString())))
 				return _builtInSerializer.Deserialize(objectType, ms);
 		}
@@ -38,7 +44,8 @@ public override object ReadJson(JsonReader reader, Type objectType, object exist
 			typeof(JoinField),
 			typeof(QueryContainer),
 			typeof(CompletionField),
-			typeof(Attachment)
+			typeof(Attachment),
+			typeof(ILazyDocument)
 		};
 
 		public override bool CanConvert(Type objectType) => NestTypesThatCanAppearInSource.Contains(objectType);

src/Tests/Document/Multiple/Reindex/ReindexApiTests.cs

@@ -62,20 +62,25 @@ public ReindexApiTests(ManualReindexCluster cluster, EndpointUsage usage)
 			// create a thousand commits and associate with the projects
 			var commits = CommitActivity.Generator.Generate(5000).ToList();
 			var bb = new BulkDescriptor();
-			for (int i = 0; i < commits.Count; i++)
+			for (var i = 0; i < commits.Count; i++)
 			{
 				var commit = commits[i];
 				var project = i % 2 == 0
 					? projects[0].Name
 					: projects[1].Name;
 
 				bb.Index<CommitActivity>(bi => bi
-					.Document(commit)
+					.Document(UpdateJoin(commit, project))
 					.Index(IndexName)
 					.Id(commit.Id)
 					.Routing(project)
 				);
 			}
+            CommitActivity UpdateJoin(CommitActivity a, string p)
+            {
+                a.ProjectName = p;
+	            return a;
+            }
 
 			var bulkResult = this._client.Bulk(b => bb);
 			bulkResult.ShouldBeValid();
@@ -126,7 +131,7 @@ [I] public void ReturnsExpectedResponse()
 			CountdownEvent observableWait = null;
 			var reindexRoutines = new List<Action>
 			{
-				//() => ReindexMany(GetSignal, Signal),
+				() => ReindexMany(GetSignal, Signal),
 				() => ReindexSingleType(GetSignal, Signal),
 				() => ReindexProjection(GetSignal, Signal)
 			};
@@ -148,7 +153,6 @@ public void ReindexMany(Func<CountdownEvent> getCountDown, Action<Exception> sig
 			);
 
 			this._reindexManyTypesResult.Subscribe(manyTypesObserver);
-			this._reindexManyTypesResult.Wait(TimeSpan.FromMinutes(3), r => { });
 		}
 
 		private void ReindexManyTypesCompleted(CountdownEvent handle)