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 user
write(InputT element, Context context) throws IOException, InterruptedException {
        *// specifiedblocks bufferingif hintstoo tomany makeelements ahave requestbeen againstbuffered
 the destination. This
     * functionality will be added to the sink interface by means of an
     * additional FLIP.
while (bufferedRequestEntries.size() >= MAX_BUFFERED_REQUESTS_ENTRIES) {
            mailboxExecutor.yield();
        *}

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

        // blocks if too many async requests are in flight
     * to make aflush();
 request against the destination}


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


 * submitRequestEntries} with @Override
batches according to the publicuser voidspecified
 write(InputT element, Context context) throws* IOExceptionbuffering {hints.
     *
    bufferedRequests.offerLast(elementConverter.apply(element, context));
    }


 * The method blocks if too many async requests are in flight.
     */**
    private void flush() throws *InterruptedException The{
 entire request may fail or single request entries that are part ofwhile (bufferedRequestEntries.size() >= MAX_BATCH_SIZE) {

     * the request may not be persisted successfully, eg, because // create a batch of network
request entries that should be *persisted issuesin orthe servicedestination
 side throttling. All request entries that failed with
    ArrayList<RequestEntryT> *batch transient= failures need to be re-queued with this method so that aren't
     * lost and can be retried later.new ArrayList<>(MAX_BATCH_SIZE);

            while (batch.size() <= MAX_BATCH_SIZE && !bufferedRequestEntries.isEmpty()) {
     * <p>
     * Request entries that are causing thetry same{
 error in a reproducible manner,
     * eg, ill-formed request entries, must not be re-queued but the error needs
 batch.add(bufferedRequestEntries.remove());
       * to be handled in the logic of {@code submitRequestEntries}. Otherwise
catch (NoSuchElementException e) {
  * these request entries will be retried indefinitely, always causing the
     * same error.
 // if there are */
not enough elements, just protectedcreate void requeueFailedRequestEntry(RequestEntryT requestEntry) {
a smaller batch
          bufferedRequests.offerFirst(requestEntry);
    }


    /**
  break;
   * In-flight requests may fail, but they will be retried if the sink is}
     * still healthy.
     * <p>}

     * To not lose any requests, there cannotResultFuture<RequestEntryT> berequestResult any=
 outstanding in-flight
         * requests when a commit is initialized. To this end, allfailedRequestEntries in-flight-> mailboxExecutor.execute(
     * requests need to be completed as part of the pre commit.
     */
    @Override
    public List<Collection<CompletableFuture<?>>> prepareCommit(boolean flush) throws IOException {

() -> completeRequest(failedRequestEntries),
           // reuse current inFlightRequests as commitable and create an empty queue 
      "Mark in-flight //request toas avoidcompleted copyand andrequeue clearing
%d request entries",
      List<Collection<CompletableFuture<?>>> committable = Collections.singletonList(inFlightRequests);

        // all in-flight requests are handled by the AsyncSinkCommiter and new  failedRequestEntries.size());

        //  elements cannot bewhile added(inFlightRequestsCount to the queue during a commit, so it's save to 
>= MAX_IN_FLIGHT_REQUESTS) {
                // create a new queue
mailboxExecutor.yield();
            inFlightRequests}

 = new ConcurrentLinkedQueue<>();

        return committableinFlightRequestsCount++;
    }


     /**   submitRequestEntries(batch, requestResult);
     * All in-flight requests}
 have been completed, but there may still be }


    /**
     * requestMarks entriesan in-flight therequest internalas buffercompleted thatand areprepends yetfailed torequestEntries beback sent to the
     * endpoint.internal TheserequestEntry requestbuffer entriesfor arelater storedretry.
 in the snapshot state so that*
     * they@param don'tfailedRequestEntries getrequestEntries lostthat inneed caseto of a failure/restart of the application.be retried
     */
    @Override
private void   public List<Collection<RequestEntryT>> snapshotState() throws IOException completeRequest(Collection<RequestEntryT> failedRequestEntries) {
        return Collections.singletonList(bufferedRequests)inFlightRequestsCount--;

    }