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 {        
        bufferedRequestEntries.putLast(elementConverter.apply(element, context));

		// forblocks now,if flushtoo willmany check if enough events elements have been buffered
		flush();
    }


    /**
     * 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 }


    /**
     * lostPersists andbuffered canRequestsEntries beinto retriedthe later.
destination by    * <p>invoking {@code
     * RequestsubmitRequestEntries} entrieswith thatbatches areaccording causingto the user specified
 same  error in a* reproduciblebuffering manner,hints.
     * eg, ill-formed request entries, must not be re-queued but the error needs
     * The method blocks if too many async requests are in flight.
     */
 to be handled in the logic of {@code submitRequestEntries}. Otherwiseprivate void flush() 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 {@code
while (batch.size() <= MAX_BATCH_SIZE && !bufferedRequestEntries.isEmpty()) {
      * submitRequestEntries}    with batches according to the usertry specified{
     *   buffering hints.
     */
    public void flush batch.add(bufferedRequestEntries.remove() throws InterruptedException {
);
                } whilecatch (bufferedRequestEntries.size() >= MAX_BATCH_SIZENoSuchElementException e) {
            // create a batch of request entries that should// beif persistedthere inare thenot destination
enough elements, just create a smaller batch
      ArrayList<RequestEntryT> batch = new ArrayList<>(MAX_BATCH_SIZE);

          break;
  for (int i = 0; i < MAX_BATCH_SIZE; i++) {
     }
           try {}

            ResultFuture<RequestEntryT> requestResult =
      batch.add(bufferedRequestEntries.remove());
              failedRequestEntries -> mailboxExecutor.execute(
   } catch (NoSuchElementException e) {
                    // 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",
       }
            }

         failedRequestEntries.size());

   //   call the destination specific code that actuallywhile persists(inFlightRequestsCount the request entries>= MAX_IN_FLIGHT_REQUESTS) {
            CompletableFuture<?>  future = submitRequestEntriesmailboxExecutor.yield(batch);

            }

    //   keep track of in flight requestinFlightRequestsCount++;
            inFlightRequests.put(futuresubmitRequestEntries(batch, requestResult);

        }
    // remove the request from the tracking queue once it competed
            future.whenComplete((response, err) -> {
    }


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


    /**inFlightRequestsCount--;

     * In-flight requests may// fail,By butjust theyiterating willthrough befailedRequestEntries, retriedit ifreverses the order sinkof isthe
      * still healthy.
     * <p>// failedRequestEntries. It doesn't make a difference for kinesis:putRecords, as the api
     * To   // does not losemake any requestsorder guarantees, but theremay cannotcause beavoidable anyreorderings outstandingfor in-flightother
     * requests when a commit is initialized. To this end, all in-flight
 // destinations.
        failedRequestEntries.forEach(bufferedRequestEntries::addFirst);
    }


    /**
  requests need to be* completedIn asflight partrequests ofwill thebe preretried commit.
if the sink is still */
healthy. But if in-flight @Overriderequests
    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);
* fail after a checkpoint has been triggered and Flink 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 areneed handledto bycompleted thebefore AsyncSinkCommiterproceeding andwith newthe commit.
     */
   // elements@Override
 cannot be added topublic theList<Void> queue during a commit, so it's save to 
prepareCommit(boolean flush) throws IOException, InterruptedException {
        if (flush) {
 // create a new queue
        inFlightRequests = new ConcurrentLinkedQueue<>flush();

        return committable;}

    }


    /**
/ wait until   * Allall in-flight requests havecompleted
 been completed, but there may still be
 while (inFlightRequestsCount > 0) *{
 request entries in the internal buffer that are yet to be sent to the mailboxExecutor.yield();
     * endpoint. These request}

 entries are stored in the snapshot state so thatreturn Collections.emptyList();
    }


 * 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);
    }

 /**
     * All 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);
    }