Page History

 The ElementConverter provides a mapping between for the elements of a
     * stream to request entries that can be sent to the destination.
     * <p>
     * The resulting request entry is buffered by the AsyncSinkWriter and sent
     * to the destination when the {@code submitRequestEntries} method is
     * invoked.
     */
    private final ElementConverter<InputT, RequestEntryT> elementConverter;


    /**
     * Buffer to hold request entries that should be persisted into the
     * destination.
     * <p>
     * A request entry contain all relevant details to make a call to the
     * destination. Eg, for Kinesis Data Streams a request entry contains the
     * payload and partition key.
     * <p>
     * It seems more natural to buffer InputT, ie, the events that should be
     * persisted, rather than RequestEntryT. However, in practice, the response
     * of a failed request call can make it very hard, if not impossible, to
     * reconstruct the original event. It is much easier, to just construct a
     * new (retry) request entry from the response and add that back to the
     * queue for later retry.
     */
    private final Deque<RequestEntryT> bufferedRequestsbufferedRequestEntries = new ConcurrentLinkedDeque<>ArrayDeque<>();


    /**
     * Tracks all pending async calls that have been executed since the last
     * checkpoint. Calls that already completed (successfully or unsuccessfully) are
     * are automatically removed fromdecrementing the queuecounter. Any request entry that was not
     * successfully persisted needneeds to be handled and retried by the logic in
     * {@code submitRequestsToApi}.
     * <p>
     * There is a limit on the number of concurrent (async) requests that can be
     * handled by the client library. This limit is enforced by checking the
     * size of this queue before issuing new requests.
     * <p>
     * To complete a checkpoint, we need to make sure that no requests are in
     * flight, as they may fail, which could then lead to data loss.
     */
    private Queue<CompletableFuture<?>> inFlightRequests = new ConcurrentLinkedQueue<>()int inFlightRequestsCount;


    /**@Override
    public *void Signals if enough RequestEntryTs have been buffered according to the userwrite(InputT element, Context context) throws IOException, InterruptedException {
     * specified buffering hints// toblocks makeif atoo requestmany againstelements thehave destination.been Thisbuffered
     * functionality will be added to the sink interface by means of an
while (bufferedRequestEntries.size() >= MAX_BUFFERED_REQUESTS_ENTRIES) {
          * additional FLIPmailboxExecutor.yield();
     *
   }

  * @return a future that will be completed once there is are record available   
     * to make a request against the destination
 bufferedRequestEntries.add(elementConverter.apply(element, context));

        // blocks if too many async requests are in flight
        */flush();
    public}


  CompletableFuture<Void> isAvailable() {
 /**
     * Persists buffered RequestsEntries into ...
the destination by invoking }

{@code
    @Override
 * submitRequestEntries} with publicbatches voidaccording write(InputT element, Context context) throws IOException {to the user specified
     *   bufferedRequests.offerLast(elementConverter.apply(element, context));buffering hints.
    }


    /**
     * The entiremethod requestblocks mayif failtoo ormany single request entries thatasync requests are partin offlight.
     */
 the request may notprivate bevoid persisted successfully, eg, because of networkflush() throws InterruptedException {
     * issues or service side throttling. All request entries that failed with
while (bufferedRequestEntries.size() >= MAX_BATCH_SIZE) {

        * transient failures need to// becreate re-queueda withbatch thisof methodrequest soentries that aren't
should be persisted in the *destination
 lost and can be retried later.
     * <p>
ArrayList<RequestEntryT> batch = new  * Request entries that are causing the same error in a reproducible manner,
     * eg, ill-formed request entries, must not be re-queued but the error needs
ArrayList<>(MAX_BATCH_SIZE);

            while (batch.size() <= MAX_BATCH_SIZE && !bufferedRequestEntries.isEmpty()) {
                *try to{
 be handled in the logic of {@code submitRequestEntries}. Otherwise
     * these request entries will be retried indefinitely, always causing the
 batch.add(bufferedRequestEntries.remove());
         * same error.
     */
}    protected void requeueFailedRequestEntry(RequestEntryT requestEntrycatch (NoSuchElementException e) {
        bufferedRequests.offerFirst(requestEntry);
     }


    /**
   // if *there In-flightare requestsnot mayenough failelements, just butcreate theya willsmaller bebatch
 retried if the sink is
     * still healthy.
       * <p>break;
     * To not lose any requests, there cannot be any outstanding in-flight}
     * requests when a commit is initialized. To}

 this end, all in-flight
     * requests need toResultFuture<RequestEntryT> berequestResult completed=
 as part of the pre commit.
     */
    @Override
    public List<Collection<CompletableFuture<?>>> prepareCommit(boolean flush) throws IOException {

failedRequestEntries -> mailboxExecutor.execute(
              // reuse current inFlightRequests as commitable and create an empty queue 
   () -> completeRequest(failedRequestEntries),
   // to avoid copy and clearing
        List<Collection<CompletableFuture<?>>> committable = Collections.singletonList(inFlightRequests);

        // all"Mark in-flight requestsrequest areas handled by the AsyncSinkCommiter completed and newrequeue 
%d        // elements cannot be added to the queue during a commit, so it's save to 
request entries",
             // create a new queue
        inFlightRequests = new ConcurrentLinkedQueue<>failedRequestEntries.size());

        return committable;
    }


    /**while (inFlightRequestsCount >= MAX_IN_FLIGHT_REQUESTS) {
     * All in-flight requests have been completed, but there may still be mailboxExecutor.yield();
     * request entries in the internal buffer that}

 are yet to be sent to the
     * endpoint. These request entries are stored in the snapshot state so that
 inFlightRequestsCount++;
            submitRequestEntries(batch, requestResult);
    * they don't get lost}
 in case of a failure/restart of the application.}


    /**
     */
 Marks an in-flight @Override
request as completed and publicprepends List<Collection<RequestEntryT>>failed snapshotState()requestEntries throwsback IOExceptionto {the
     * internal requestEntry return Collections.singletonList(bufferedRequests);buffer for later retry.
    }

new ElementConverter<InputT, PutRecordsRequestEntry>() {
    @Override
    public PutRecordsRequestEntry apply(InputT element, SinkWriter.Context context) {
        return PutRecordsRequestEntry
        *
     * @param failedRequestEntries requestEntries that need to be retried
     */
    private void completeRequest(Collection<RequestEntryT> failedRequestEntries) {
        inFlightRequestsCount--;

        // By just iterating through failedRequestEntries, it reverses the order of the
        // failedRequestEntries. It doesn't make a difference for kinesis:putRecords, as the api
        // does not make any order guarantees, but may cause avoidable reorderings for other
        // destinations.builder()
        failedRequestEntries.forEach(bufferedRequestEntries::addFirst);
    }


    .data(SdkBytes.fromUtf8String(element.toString()))/**
     * In flight requests will be retried if the sink is still .partitionKey(String.valueOf(element.hashCode()))healthy. But if in-flight requests
     * fail after a checkpoint has been triggered and Flink  .build();
    }
}

public class AmazonKinesisDataStreamSink<InputT> extends AsyncSink<InputT, PutRecordsRequestEntry> {

    @Override
    protected CompletableFuture<?> submitRequestEntries(List<PutRecordsRequestEntry> requestEntries) {

        // create a batch request
        PutRecordsRequest batchRequest = PutRecordsRequest
                .builder()needs to recover from the checkpoint,
     * the (failed) in-flight requests are gone and cannot be retried. Hence, there cannot be any
     * outstanding in-flight requests when a commit is initialized.
     *
     * <p>To this end, all in-flight requests need to completed before proceeding with the commit.
     */
    @Override
    public List<Void> prepareCommit(boolean flush) throws IOException, InterruptedException {
        if (flush)  {
     .records(requestEntries)
       flush();
         .streamName(streamName)}

        // wait until all in-flight requests completed
  .build();

      while (inFlightRequestsCount //> call0) api{
 with batch request
        CompletableFuture<PutRecordsResponse> future = client.putRecords(batchRequestmailboxExecutor.yield();

        // re-queue elements of failed requests
   }

        return Collections.emptyList();
    }


 CompletableFuture<PutRecordsResponse> handleResponse = future/**
     * All in-flight requests have   .whenComplete((responsebeen completed, err)but ->there {
may still be
     * request entries in the internal buffer that are yet if (response.failedRecordCount() > 0) {to be sent to the
     * endpoint. These request entries are stored in the snapshot state so that
   List<PutRecordsResultEntry> records = response.records();

           * they don't get lost in case of a failure/restart of the application.
      */
   for (int@Override
 i = 0; ipublic <List<Collection<RequestEntryT>> records.sizesnapshotState(); i++) throws IOException {
                        if (records.get(i).errorCode() != null) {
                            requeueFailedRequest(requestEntries.get(i));
                        }
                    }
                }

                //TODO: handle errors of the entire request...
            });

        // return future to track completion of async request
        return handleResponse;
    }

    ...
}return Collections.singletonList(bufferedRequestEntries);
    }