Added the ability to retry a bulk request

Author: niemyjski

src/Foundatio.Repositories.Elasticsearch/Extensions/ElasticIndexExtensions.cs

@@ -103,35 +103,35 @@ public static IDictionary<string, AggregationResult> ToAggregationResult<T>(this
             return res.Aggregations.ToAggregationResult();
         }
 
-        public static Task<IBulkResponse> IndexManyAsync<T>(this IElasticClient client, IEnumerable<T> objects, Func<T, string> getParent, Func<T, string> getIndex = null, string type = null) where T : class {
+        public static Task<IBulkResponse> IndexManyAsync<T>(this IElasticClient client, IEnumerable<T> objects, Func<T, string> getParent, Func<T, string> getIndex = null, string type = null, bool isCreateOperation = false) where T : class {
             if (objects == null)
                 throw new ArgumentNullException(nameof(objects));
 
-            if (getParent == null && getIndex == null)
-                return client.IndexManyAsync(objects, null, type);
-
-            var indexBulkRequest = CreateIndexBulkRequest(objects, getIndex, type, getParent);
+            var indexBulkRequest = CreateIndexBulkRequest(objects, getIndex, type, getParent, isCreateOperation);
             return client.BulkAsync(indexBulkRequest);
         }
 
-        private static BulkRequest CreateIndexBulkRequest<T>(IEnumerable<T> objects, Func<T, string> getIndex, string type, Func<T, string> getParent) where T : class {
+        private static BulkRequest CreateIndexBulkRequest<T>(IEnumerable<T> objects, Func<T, string> getIndex, string type, Func<T, string> getParent, bool isCreateOperation) where T : class {
             var bulkRequest = new BulkRequest();
             var list = objects.Select(o => {
-                var doc = new BulkIndexOperation<T>(o) { Type = type };
+                IBulkOperation doc = isCreateOperation ? (IBulkOperation)new BulkCreateOperation<T>(o) : new BulkIndexOperation<T>(o);
+                doc.Type = type;
                 if (getParent != null)
                     doc.Parent = getParent(o);
 
                 if (getIndex != null)
                     doc.Index = getIndex(o);
 
-                var versionedDoc = o as IVersioned;
-                if (versionedDoc != null && versionedDoc.Version > 0)
-                    doc.Version = versionedDoc.Version;
+                if (!isCreateOperation) {
+                    var versionedDoc = o as IVersioned;
+                    if (versionedDoc != null)
+                        doc.Version = versionedDoc.Version;
+                }
 
                 return doc;
-            }).Cast<IBulkOperation>().ToList();
+            }).ToList();
             bulkRequest.Operations = list;
-            
+
             return bulkRequest;
         }
 

src/Foundatio.Repositories.Elasticsearch/Repositories/ElasticRepositoryBase.cs

@@ -58,7 +58,7 @@ public async Task AddAsync(IEnumerable<T> documents, bool addToCache = false, Ti
                 foreach (var doc in docs)
                     await _validator.ValidateAndThrowAsync(doc).AnyContext();
 
-            await IndexDocumentsAsync(docs).AnyContext();
+            await IndexDocumentsAsync(docs, isCreateOperation: true).AnyContext();
 
             if (addToCache)
                 await AddToCacheAsync(docs, expiresIn).AnyContext();
@@ -668,7 +668,7 @@ private async Task OnDocumentsChangedAsync(ChangeType changeType, IReadOnlyColle
 
         #endregion
 
-        private async Task IndexDocumentsAsync(IReadOnlyCollection<T> documents) {
+        private async Task IndexDocumentsAsync(IReadOnlyCollection<T> documents, bool isCreateOperation = false) {
             if (HasMultipleIndexes) {
                 foreach (var documentGroup in documents.GroupBy(TimeSeriesType.GetDocumentIndex))
                     await TimeSeriesType.EnsureIndexAsync(documentGroup.First()).AnyContext();
@@ -677,6 +677,7 @@ private async Task IndexDocumentsAsync(IReadOnlyCollection<T> documents) {
             if (documents.Count == 1) {
                 var document = documents.Single();
                 var response = await _client.IndexAsync(document, i => {
+                    i.OpType(isCreateOperation ? OpType.Create : OpType.Index);
                     i.Type(ElasticType.Name);
 
                     if (GetParentIdFunc != null)
@@ -685,10 +686,9 @@ private async Task IndexDocumentsAsync(IReadOnlyCollection<T> documents) {
                     if (GetDocumentIndexFunc != null)
                         i.Index(GetDocumentIndexFunc(document));
 
-                    if (HasVersion) {
+                    if (HasVersion && isCreateOperation) {
                         var versionDoc = (IVersioned)document;
-                        if (versionDoc.Version > 0)
-                            i.Version(versionDoc.Version);
+                        i.Version(versionDoc.Version);
                     }
 
                     return i;
@@ -706,7 +706,7 @@ private async Task IndexDocumentsAsync(IReadOnlyCollection<T> documents) {
                     versionDoc.Version = response.Version;
                 }
             } else {
-                var response = await _client.IndexManyAsync(documents, GetParentIdFunc, GetDocumentIndexFunc, ElasticType.Name).AnyContext();
+                var response = await _client.IndexManyAsync(documents, GetParentIdFunc, GetDocumentIndexFunc, ElasticType.Name, isCreateOperation).AnyContext();
                 _logger.Trace(() => response.GetRequest());
 
                 if (HasVersion) {
@@ -723,12 +723,26 @@ private async Task IndexDocumentsAsync(IReadOnlyCollection<T> documents) {
                     }
                 }
 
+                var allErrors = response.ItemsWithErrors.ToList();
+                if (allErrors.Count > 0) {
+                    var retryableIds = allErrors.Where(e => e.Status == 429 || e.Status == 503).Select(e => e.Id).ToList();
+                    if (retryableIds.Count > 0) {
+                        var docs = documents.Where(d => retryableIds.Contains(d.Id)).ToList();
+                        await IndexDocumentsAsync(docs, isCreateOperation).AnyContext();
+
+                        // return as all recoverable items were retried.
+                        if (allErrors.Count == retryableIds.Count)
+                            return;
+                    }
+                }
+
                 if (!response.IsValid) {
                     string message = response.GetErrorMessage();
                     _logger.Error().Exception(response.OriginalException).Message(message).Property("request", response.GetRequest()).Write();
                     throw new ApplicationException(message, response.OriginalException);
                 }
             }
+            // 429 // 503
         }
 
         protected virtual async Task AddToCacheAsync(ICollection<T> documents, TimeSpan? expiresIn = null) {