Skip to content

Commit

Permalink
all basic key points implemented
Browse files Browse the repository at this point in the history
  • Loading branch information
@l-trotta
l-trotta committed Jan 22, 2025
1 parent 9a6f784 commit 3858711
Show file tree
Hide file tree
Showing 3 changed files with 127 additions and 57 deletions.
93 changes: 74 additions & 19 deletions java-client/src/main/java/co/elastic/clients/elasticsearch/_helpers/bulk/BulkIngester.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand All @@ -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;
Expand Down Expand Up @@ -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;

Expand All @@ -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;
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -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
Expand All @@ -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
Expand All @@ -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();
Expand All @@ -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
Expand Down Expand Up @@ -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);
}
Expand Down
25 changes: 15 additions & 10 deletions ...src/main/java/co/elastic/clients/elasticsearch/_helpers/bulk/BulkOperationRepeatable.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -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() {
Expand All @@ -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());
}
}
Loading

0 comments on commit 3858711

Please sign in to comment.