Package org.apache.beam.sdk.io.aws2.common

Class AsyncBatchWriteHandler<RecT,ResT>

java.lang.Object

org.apache.beam.sdk.io.aws2.common.AsyncBatchWriteHandler<RecT,ResT>

Type Parameters:

RecT - Record type in batch

ResT - Potentially erroneous result that needs to be correlated to a record using failedRecords(List, List)

@NotThreadSafe
@Internal
public abstract class AsyncBatchWriteHandler<RecT,ResT>
extends Object

Async handler that automatically retries unprocessed records in case of a partial success.

The handler enforces the provided upper limit of concurrent requests. Once that limit is reached any further call to batchWrite(String, List) will block until another request completed.

The handler is fail fast and won't submit any further request after a failure. Async failures can be polled using checkForAsyncFailure().

Nested Class Summary

Nested Classes

Modifier and Type	Class	Description
static interface	AsyncBatchWriteHandler.Stats	Statistics on the batch request.

Field Summary

Fields

Modifier and Type	Field	Description
protected final Function<ResT,String>	errorCodeFn
protected final BiFunction<String,List<RecT>,CompletableFuture<List<ResT>>>	submitFn

Constructor Summary

Constructors

Modifier	Constructor	Description
protected	AsyncBatchWriteHandler(int concurrency, org.apache.beam.sdk.util.FluentBackoff backoff, AsyncBatchWriteHandler.Stats stats, Function<ResT,String> errorCodeFn, BiFunction<String,List<RecT>,CompletableFuture<List<ResT>>> submitFn)

Method Summary

Modifier and Type	Method	Description
final void	batchWrite(String destination, List<RecT> records)	Asynchronously trigger a batch write request (unless already in error state).
final void	batchWrite(String destination, List<RecT> records, boolean throwAsyncFailures)	Asynchronously trigger a batch write request (unless already in error state).
static <RecT, ErrT> AsyncBatchWriteHandler<RecT,ErrT>	byId(int concurrency, int partialRetries, RetryConfiguration retry, AsyncBatchWriteHandler.Stats stats, BiFunction<String,List<RecT>,CompletableFuture<List<ErrT>>> submitFn, Function<ErrT,String> errorCodeFn, Function<RecT,String> recordIdFn, Function<ErrT,String> errorIdFn)	AsyncBatchWriteHandler that correlates records and results by id, all results are erroneous.
static <RecT, ErrT> AsyncBatchWriteHandler<RecT,ErrT>	byId(int concurrency, org.apache.beam.sdk.util.FluentBackoff backoff, AsyncBatchWriteHandler.Stats stats, BiFunction<String,List<RecT>,CompletableFuture<List<ErrT>>> submitFn, Function<ErrT,String> errorCodeFn, Function<RecT,String> recordIdFn, Function<ErrT,String> errorIdFn)	AsyncBatchWriteHandler that correlates records and results by id, all results are erroneous.
static <RecT, ResT> AsyncBatchWriteHandler<RecT,ResT>	byPosition(int concurrency, int partialRetries, RetryConfiguration retry, AsyncBatchWriteHandler.Stats stats, BiFunction<String,List<RecT>,CompletableFuture<List<ResT>>> submitFn, Function<ResT,String> errorCodeFn)	AsyncBatchWriteHandler that correlates records and results by position in the respective list.
static <RecT, ResT> AsyncBatchWriteHandler<RecT,ResT>	byPosition(int concurrency, org.apache.beam.sdk.util.FluentBackoff backoff, AsyncBatchWriteHandler.Stats stats, BiFunction<String,List<RecT>,CompletableFuture<List<ResT>>> submitFn, Function<ResT,String> errorCodeFn)	AsyncBatchWriteHandler that correlates records and results by position in the respective list.
final void	checkForAsyncFailure()	Check if any failure happened async.
protected abstract List<RecT>	failedRecords(List<RecT> records, List<ResT> results)
final boolean	hasErrored()	If this handler has errored since it was last reset.
protected abstract boolean	hasFailedRecords(List<ResT> results)
final int	requestsInProgress()
final void	reset()
final void	waitForCompletion()	Wait for all pending requests to complete and check for failures.

Methods inherited from class java.lang.Object

clone, equals, finalize, getClass, hashCode, notify, notifyAll, toString, wait, wait, wait

Field Details

submitFn

protected final BiFunction<String,List<RecT>,CompletableFuture<List<ResT>>> submitFn

errorCodeFn

protected final Function<ResT,String> errorCodeFn

Constructor Details

AsyncBatchWriteHandler

protected AsyncBatchWriteHandler(int concurrency,
 org.apache.beam.sdk.util.FluentBackoff backoff,
 AsyncBatchWriteHandler.Stats stats,
 Function<ResT,String> errorCodeFn,
 BiFunction<String,List<RecT>,CompletableFuture<List<ResT>>> submitFn)

Method Details

requestsInProgress

public final int requestsInProgress()

reset

public final void reset()

hasErrored

public final boolean hasErrored()

If this handler has errored since it was last reset.

checkForAsyncFailure

public final void checkForAsyncFailure()
                                throws Throwable

Check if any failure happened async.

Throws:

Throwable - The last async failure, afterwards reset it.

waitForCompletion

public final void waitForCompletion()
                             throws Throwable

Wait for all pending requests to complete and check for failures.

Throws:

Throwable - The last async failure if present using checkForAsyncFailure()

batchWrite

public final void batchWrite(String destination,
 List<RecT> records)
                      throws Throwable

Asynchronously trigger a batch write request (unless already in error state).

This will respect the concurrency limit of the handler and first wait for a permit.

Throws:

Throwable - The last async failure if present using checkForAsyncFailure()

batchWrite

public final void batchWrite(String destination,
 List<RecT> records,
 boolean throwAsyncFailures)
                      throws Throwable

Asynchronously trigger a batch write request (unless already in error state).

This will respect the concurrency limit of the handler and first wait for a permit.

Parameters:

throwAsyncFailures - If to check and throw pending async failures

Throws:

Throwable - The last async failure if present using checkForAsyncFailure()

failedRecords

protected abstract List<RecT> failedRecords(List<RecT> records,
 List<ResT> results)

hasFailedRecords

protected abstract boolean hasFailedRecords(List<ResT> results)

byPosition

public static <RecT,
ResT>
AsyncBatchWriteHandler<RecT,ResT> byPosition(int concurrency,
 int partialRetries,
 @Nullable
 RetryConfiguration retry,
 AsyncBatchWriteHandler.Stats stats,
 BiFunction<String,List<RecT>,CompletableFuture<List<ResT>>> submitFn,
 Function<ResT,String> errorCodeFn)

AsyncBatchWriteHandler that correlates records and results by position in the respective list.

byPosition

public static <RecT,
ResT>
AsyncBatchWriteHandler<RecT,ResT> byPosition(int concurrency,
 org.apache.beam.sdk.util.FluentBackoff backoff,
 AsyncBatchWriteHandler.Stats stats,
 BiFunction<String,List<RecT>,CompletableFuture<List<ResT>>> submitFn,
 Function<ResT,String> errorCodeFn)

AsyncBatchWriteHandler that correlates records and results by position in the respective list.

byId

public static <RecT,
ErrT>
AsyncBatchWriteHandler<RecT,ErrT> byId(int concurrency,
 int partialRetries,
 @Nullable
 RetryConfiguration retry,
 AsyncBatchWriteHandler.Stats stats,
 BiFunction<String,List<RecT>,CompletableFuture<List<ErrT>>> submitFn,
 Function<ErrT,String> errorCodeFn,
 Function<RecT,String> recordIdFn,
 Function<ErrT,String> errorIdFn)

AsyncBatchWriteHandler that correlates records and results by id, all results are erroneous.

byId

public static <RecT,
ErrT>
AsyncBatchWriteHandler<RecT,ErrT> byId(int concurrency,
 org.apache.beam.sdk.util.FluentBackoff backoff,
 AsyncBatchWriteHandler.Stats stats,
 BiFunction<String,List<RecT>,CompletableFuture<List<ErrT>>> submitFn,
 Function<ErrT,String> errorCodeFn,
 Function<RecT,String> recordIdFn,
 Function<ErrT,String> errorIdFn)

AsyncBatchWriteHandler that correlates records and results by id, all results are erroneous.