all basic key points implemented

Author: l-trotta

java-client/src/main/java/co/elastic/clients/elasticsearch/_helpers/bulk/BulkIngester.java

@@ -24,6 +24,7 @@
 import co.elastic.clients.elasticsearch.core.BulkRequest;
 import co.elastic.clients.elasticsearch.core.BulkResponse;
 import co.elastic.clients.elasticsearch.core.bulk.BulkOperation;
+import co.elastic.clients.elasticsearch.core.bulk.BulkResponseItem;
 import co.elastic.clients.transport.BackoffPolicy;
 import co.elastic.clients.transport.TransportOptions;
 import co.elastic.clients.util.ApiTypeHelper;
@@ -33,9 +34,9 @@
 
 import javax.annotation.Nullable;
 import java.time.Duration;
-import java.time.Instant;
 import java.util.ArrayList;
 import java.util.Collections;
+import java.util.Iterator;
 import java.util.List;
 import java.util.Optional;
 import java.util.concurrent.CompletionStage;
@@ -66,6 +67,7 @@ public class BulkIngester<Context> implements AutoCloseable {
 
     private @Nullable ScheduledFuture<?> flushTask;
     private @Nullable ScheduledExecutorService scheduler;
+    private @Nullable ScheduledExecutorService retryScheduler;
     private boolean isExternalScheduler = false;
     private BackoffPolicy backoffPolicy;
 
@@ -81,7 +83,7 @@ public class BulkIngester<Context> implements AutoCloseable {
     private final FnCondition addCondition = new FnCondition(lock, this::canAddOperation);
     private final FnCondition sendRequestCondition = new FnCondition(lock, this::canSendRequest);
     private final FnCondition closeCondition = new FnCondition(lock, this::closedAndFlushed);
-    private AtomicInteger listenerInProgressCount = new AtomicInteger();
+    private final AtomicInteger listenerInProgressCount = new AtomicInteger();
 
     private static class RequestExecution<Context> {
         public final long id;
@@ -138,6 +140,21 @@ private BulkIngester(Builder<Context> builder) {
         if (backoffPolicy == null) {
             backoffPolicy = BackoffPolicy.noBackoff();
         }
+        // preparing a scheduler that will trigger flushes when it finds enqueued requests ready to be retried
+        // TODO should we just keep a single scheduler?
+        else {
+            retryScheduler = Executors.newScheduledThreadPool(maxRequests + 1, (r) -> {
+                Thread t = Executors.defaultThreadFactory().newThread(r);
+                t.setName("bulk-ingester-retry#" + ingesterId + "#" + t.getId());
+                t.setDaemon(true);
+                return t;
+            });
+            retryScheduler.scheduleWithFixedDelay(
+                this::retryFlush,
+                1000,1000, // TODO should we hardcode this?
+                TimeUnit.MILLISECONDS
+            );
+        }
     }
 
     //----- Getters
@@ -283,9 +300,20 @@ private void failsafeFlush() {
         }
     }
 
+    // triggers a flush if it finds queued retries
+    private void retryFlush() {
+        try {
+            if (operations.stream().anyMatch(op -> op.getRetries() != null && op.isSendable())) {
+                flush();
+            }
+        } catch (Throwable thr) {
+            // Log the error and continue
+            logger.error("Error in background flush", thr);
+        }
+    }
+
     public void flush() {
-        // Keeping sent operations for possible retries
-        List<BulkOperationRepeatable<Context>> requestsSent = new ArrayList<>();
+        List<BulkOperationRepeatable<Context>> sentRequests = new ArrayList<>();
         RequestExecution<Context> exec = sendRequestCondition.whenReadyIf(
             () -> {
                 // May happen on manual and periodic flushes
@@ -294,7 +322,7 @@ public void flush() {
             () -> {
                 // Selecting operations that can be sent immediately
                 List<BulkOperationRepeatable<Context>> immediateOpsRep = operations.stream()
-                    .filter(BulkOperationRepeatable::canRetry)
+                    .filter(BulkOperationRepeatable::isSendable)
                     .collect(Collectors.toList());
 
                 // Dividing actual operations from contexts
@@ -309,11 +337,12 @@ public void flush() {
                 // Build the request
                 BulkRequest request = newRequest().operations(immediateOps).build();
 
-                List<Context> requestContexts = contexts.isEmpty() ? Collections.nCopies(immediateOpsRep.size(),
-                    null) : contexts;
+                List<Context> requestContexts = contexts.isEmpty() ?
+                    Collections.nCopies(immediateOpsRep.size(),
+                        null) : contexts;
 
                 // Prepare for next round
-                requestsSent.addAll(immediateOpsRep);
+                sentRequests.addAll(immediateOpsRep);
                 operations.removeAll(immediateOpsRep);
                 currentSize = operations.size();
                 addCondition.signalIfReady();
@@ -340,18 +369,43 @@ public void flush() {
             // A request was actually sent
             exec.futureResponse.handle((resp, thr) -> {
                 if (resp != null) {
-                    // Success
-                    if (listener != null) {
-                        listenerInProgressCount.incrementAndGet();
-                        scheduler.submit(() -> {
-                            try {
-                                listener.afterBulk(exec.id, exec.request, exec.contexts, resp);
-                            } finally {
-                                if (listenerInProgressCount.decrementAndGet() == 0) {
-                                    closeCondition.signalIfReady();
+                    // Success? Checking if total or partial
+                    List<BulkResponseItem> failedRequestsCanRetry = resp.items().stream()
+                        .filter(i -> i.error() != null && i.status() == 429)
+                        .collect(Collectors.toList());
+
+                    if (failedRequestsCanRetry.isEmpty() || !backoffPolicy.equals(BackoffPolicy.noBackoff())) {
+                        // Total success! ...or there's no retry policy implemented. Either way, can call
+                        // listener after bulk
+                        if (listener != null) {
+                            listenerInProgressCount.incrementAndGet();
+                            scheduler.submit(() -> {
+                                try {
+                                    listener.afterBulk(exec.id, exec.request, exec.contexts, resp);
+                                } finally {
+                                    if (listenerInProgressCount.decrementAndGet() == 0) {
+                                        closeCondition.signalIfReady();
+                                    }
                                 }
+                            });
+                        }
+                    } else {
+                        // Partial success, retrying failed requests if policy allows it
+                        // Getting original requests
+                        for (BulkResponseItem bulkItemResponse : failedRequestsCanRetry) {
+                            int index = resp.items().indexOf(bulkItemResponse);
+                            BulkOperationRepeatable<Context> original = sentRequests.get(index);
+                            if (original.getRetries().hasNext()) {
+                                Iterator<Long> retries =
+                                    Optional.ofNullable(original.getRetries()).orElse(backoffPolicy.iterator());
+                                addRetry(new BulkOperationRepeatable<>(original.getOperation(),
+                                    original.getContext(), retries));
+                                // TODO remove after checking
+                                assert (bulkItemResponse.operationType().toString().equals(sentRequests.get(index).getOperation()._kind().toString()));
                             }
-                        });
+                            // TODO should print some message?
+
+                        }
                     }
                 } else {
                     // Failure
@@ -383,7 +437,8 @@ public void add(BulkOperation operation, Context context) {
             throw new IllegalStateException("Ingester has been closed");
         }
 
-        BulkOperationRepeatable<Context> repeatableOp = new BulkOperationRepeatable<>(operation, context, Optional.empty());
+        BulkOperationRepeatable<Context> repeatableOp = new BulkOperationRepeatable<>(operation, context,
+            null);
 
         innerAdd(repeatableOp);
     }

java-client/src/main/java/co/elastic/clients/elasticsearch/_helpers/bulk/BulkOperationRepeatable.java

@@ -10,14 +10,15 @@
 public class BulkOperationRepeatable<Context> {
     private final BulkOperation operation;
     private final Context context;
-    private final Optional<Iterator<Long>> retries;
-    private Instant nextRetry;
+    private final Iterator<Long> retries;
+    private Instant retryTime;
 
-    public BulkOperationRepeatable(BulkOperation request, Context context, Optional<Iterator<Long>> retries) {
+    public BulkOperationRepeatable(BulkOperation request, Context context, Iterator<Long> retries) {
         this.operation = request;
         this.context = context;
         this.retries = retries;
-        this.nextRetry = Instant.now().plus(retries.map(Iterator::next).orElse(0L), ChronoUnit.MILLIS);
+        // if the retries iterator is null it means that it's not a retry, otherwise calculating retry time
+        this.retryTime = Optional.ofNullable(retries).map(r -> Instant.now().plus(r.next(), ChronoUnit.MILLIS)).orElse(Instant.now());
     }
 
     public BulkOperation getOperation() {
@@ -28,15 +29,19 @@ public Context getContext() {
         return context;
     }
 
-    public Optional<Iterator<Long>> getRetries() {
+    public Iterator<Long> getRetries() {
         return retries;
     }
 
-    public Instant getNextRetry() {
-        return this.nextRetry;
-    }
+//    public Instant getCurrentRetryTime() {
+//        return this.retryTime;
+//    }
+//
+//    public Instant getNextRetryTime() {
+//        return Instant.now().plus(retries.next(), ChronoUnit.MILLIS);
+//    }
 
-    public boolean canRetry() {
-        return this.retries.map(Iterator::hasNext).orElse(true);
+    public boolean isSendable() {
+        return retryTime.isBefore(Instant.now());
     }
 }