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 BlockingDeque<RequestEntryT>Deque<RequestEntryT> bufferedRequestEntries = new LinkedBlockingDeque<>(...)ArrayDeque<>();


    /**
     * Tracks all pending async calls that have been executed since the last
     * checkpoint. Calls that already completed (successfully or unsuccessfully) are
     * are automatically removeddecrementing from 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 BlockingDeque<CompletableFuture<?>> inFlightRequests = new LinkedBlockingDeque<>(...)int inFlightRequestsCount;


    @Override
    public void write(InputT element, Context context) throws IOException, InterruptedException {
        
// blocks if too many elements have  bufferedRequestEntries.putLast(elementConverter.apply(element, context));

		been buffered
    }


    /**
     * The entire request may fail or single request entries that are part of
while (bufferedRequestEntries.size() >= MAX_BUFFERED_REQUESTS_ENTRIES) {
            mailboxExecutor.yield();
        *}

 the request may not be persisted successfully, eg, because of network
 bufferedRequestEntries.add(elementConverter.apply(element, context));

       * issues// orblocks serviceif sidetoo throttling.many Allasync requestrequests entriesare thatin failed withflight
     * transient failures needflush();
 to be re-queued with this method so that aren't }


    /**
     * lost and can be retried later.Persists buffered RequestsEntries into the destination by invoking {@code
     * <p>
     * Request entries that are causing the same error in a reproducible manner,submitRequestEntries} with batches according to the user specified
     * buffering hints.
     *
     * eg, ill-formed request entries, must not be re-queued but the error needsThe method blocks if too many async requests are in flight.
     */
 to be handled inprivate thevoid logic of {@code submitRequestEntries}. Otherwiseflush() throws InterruptedException {
     * these request entries will be retried indefinitely, always causing thewhile (bufferedRequestEntries.size() >= MAX_BATCH_SIZE) {

     * same error.
     *//
 create a batch protectedof voidrequest requeueFailedRequestEntry(RequestEntryT requestEntry) throws InterruptedException {
   entries that should be persisted in the destination
         bufferedRequestEntries.putFirst(requestEntry);
    }


   ArrayList<RequestEntryT> batch = new ArrayList<>(MAX_BATCH_SIZE);

    /**
     * Persists buffered RequestsEntries into the destination by invoking {@codewhile (batch.size() <= MAX_BATCH_SIZE && !bufferedRequestEntries.isEmpty()) {
     * submitRequestEntries} with batches according to the user specified
     * buffering hints.try {
     */
    public void flush() throws InterruptedException {
        while batch.add(bufferedRequestEntries.sizeremove() >= MAX_BATCH_SIZE) {);
            // create a batch of} requestcatch entries(NoSuchElementException thate) should{
 be persisted in the destination
            ArrayList<RequestEntryT> batch = new ArrayList<>(MAX_BATCH_SIZE);

            for (int i = 0; i < MAX_BATCH_SIZE; i++) {
// if there are not enough elements, just create a smaller batch
                    break;
       try {
        }
            batch.add(bufferedRequestEntries.remove());}

            ResultFuture<RequestEntryT> requestResult =
   } catch (NoSuchElementException e) {
             failedRequestEntries -> mailboxExecutor.execute(
     // if there are not enough elements, just create a smaller batch
            () -> completeRequest(failedRequestEntries),
      break;
                }
      "Mark in-flight request as completed  }

and requeue %d request entries",
             logger.info("submit requests for {} elements", batch            failedRequestEntries.size());

            //while call(inFlightRequestsCount the destination specific code that actually persists the request entries
>= MAX_IN_FLIGHT_REQUESTS) {
                  CompletableFuture<?> future = submitRequestEntries(batchmailboxExecutor.yield();

            //}

 keep track of in flight request
            inFlightRequests.put(future)inFlightRequestsCount++;

            // remove the request from the tracking queue once it competedsubmitRequestEntries(batch, requestResult);
        }
    }


    /**
    future.whenComplete((response, err) -> {
            * Marks an in-flight request as completed and prepends failed requestEntries back to the
     inFlightRequests.remove(future);
      * internal requestEntry buffer for later retry.
     *
 });
    * @param failedRequestEntries requestEntries }
that need to be }retried


     */**
    private * In-flight requests may fail, but they will be retried if the sink is
void completeRequest(Collection<RequestEntryT> failedRequestEntries) {
        inFlightRequestsCount--;

       * still healthy.
     * <p>
     * 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 to be completed as part of the pre commit.
     */
    @Override
    public List<Collection<CompletableFuture<?>>> prepareCommit(boolean flush) throws IOException {

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

        // all// 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.
        failedRequestEntries.forEach(bufferedRequestEntries::addFirst);
    }


    /**
     * In flight requests will be retried if the sink is still healthy. But if in-flight requests
     * fail after a checkpoint has been triggered and Flink needs to recover from the checkpoint,
     * the (failed) in-flight requests are handled by the AsyncSinkCommiter gone and newcannot 
be retried. Hence, there cannot be any
  // elements cannot be* addedoutstanding toin-flight therequests queuewhen during a commit, so it's save to 
  is initialized.
     *
     * //<p>To createthis aend, new queue
        inFlightRequests = new ConcurrentLinkedQueue<>();

all in-flight requests need to completed before proceeding with the commit.
     */
   return committable;@Override
    }


public List<Void> prepareCommit(boolean  /**
     * All in-flight requests have been completed, but there may still be
flush) throws IOException, InterruptedException {
        if (flush) {
        * request entries in the internal buffer that are yet to be sent to the
flush();
        }

        // *wait endpoint.until Theseall requestin-flight entriesrequests arecompleted
 stored in the snapshot state so that
 while (inFlightRequestsCount > 0) *{
 they don't get lost in case of a failure/restart of the applicationmailboxExecutor.yield();
     */
   }

   @Override
    public List<Collection<RequestEntryT>>return snapshotStateCollections.emptyList() throws IOException {;
    }


    /**
     * All  return Collections.singletonList(bufferedRequestEntries);in-flight requests have been completed, but there may still be
    }

  * 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
     * they don't get lost in case of a failure/restart of the application.
     */
    @Override
    public List<Collection<RequestEntryT>> snapshotState() throws IOException {
        return Collections.singletonList(bufferedRequestEntries);
    }

private class AmazonKinesisDataStreamWriter extends AsyncSinkWriter<InputT, PutRecordsRequestEntry> {

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

private class AmazonKinesisDataStreamWriter extends AsyncSinkWriter<InputT, PutRecordsRequestEntry> {

    @Override
    protected void submitRequestEntries(List<PutRecordsRequestEntry> requestEntries, ResultFuture<PutRecordsRequestEntry> requestResult) {

        // create a batch request
        PutRecordsRequest batchRequest = PutRecordsRequest
                .builder()
                .records(requestEntries)
        // create a batch request
    .streamName(streamName)
    PutRecordsRequest batchRequest = PutRecordsRequest
         .build();

       .builder()
      // call api with batch request
        CompletableFuture<PutRecordsResponse> future = client.recordsputRecords(requestEntriesbatchRequest);

        // re-queue elements of failed requests
   .streamName(streamName)     future.whenComplete((response, err) -> {
                if (response.buildfailedRecordCount();
 > 0) {
        // call api with batch request
       ArrayList<PutRecordsRequestEntry> CompletableFuture<PutRecordsResponse>failedRequestEntries future= =new clientArrayList<>(response.putRecordsfailedRecordCount(batchRequest));

        // re-queue elements of failed requests
        CompletableFuture<PutRecordsResponse>List<PutRecordsResultEntry> handleResponserecords = futureresponse.records();
    
        .whenComplete((response, err) -> {
         for (int i = 0; i < if (response.failedRecordCountrecords.size() > 0; i++) {
                    List<PutRecordsResultEntry>    if (records.get(i).errorCode() != response.records();

null) {
                     for (int i = 0; i < records.size(); i++) {
failedRequestEntries.add(requestEntries.get(i));
                        }
        if (records.get(i).errorCode() != null) {
        }
    
                requeueFailedRequest(requestEntries.get(i));
    requestResult.complete(failedRequestEntries);
                } else   }{
                    }requestResult.complete(Collections.emptyList());
                }

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

    }

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

    ...
}

...
}