Page History

public abstract class AsyncSinkWriter<InputT, RequestEntryT extends Serializable> implements SinkWriter<InputT, Collection<CompletableFuture<?>>Void, Collection<RequestEntryT>> {

    /**
     * This method specifies how to persist buffered request entries into the
     * destination. It is implemented when support for a new destination is
     * added.
     * <p>
     * The method is invoked with a set of request entries according to the
     * buffering hints (and the valid limits of the destination). The logic then
     * needs to create and execute the request against the destination (ideally
     * by batching together multiple request entries to increase efficiency).
     * The logic also needs to identify individual request entries that were not
     * persisted successfully and resubmit them using the {@code
     * requeueFailedRequestEntryrequestResult} method.
     * <p>
     * The method returns a future During checkpointing, the sink needs to ensure that indicates, once completed, that allthere are no
     * outstanding in-flight requests.
     *
     * @param requestEntries a set of request entries that haveshould beenbe passedsent to the method on invocation have
     *  either been successfully persisted in the destination or have been
     * re-queued.
     * <p>destination
     * During checkpointing, the sink@param requestResult  a ResultFuture that needs to ensurebe thatcompleted thereonce are noall
     * outstanding in-flight requests. Ie, that all futures returned by this
     * method are completed.
     *
     * @param requestEntries a set of requestsrequest entries that shouldhave bebeen sentpassed to the APImethod
     *                       endpointon invocation have either been successfully
     * @return      a future that completes when all request entries have been
     * successfully persisted toin the APIdestination or werehave re-queuedbeen
     */
    protected abstract CompletableFuture<?> submitRequestEntries(List<RequestEntryT> requestEntries);


    /**
     * The ElementConverter provides a mapping betweenre-queued
 for the elements of a*/
    protected *abstract streamvoid 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;

    ...
}submitRequestEntries(List<RequestEntryT> requestEntries, ResultFuture<RequestEntryT> requestResult);

    ...
}

    /**
     * BufferThe toElementConverter holdprovides requesta entriesmapping thatbetween shouldfor bethe persistedelements intoof thea
     * stream to request entries that can be sent to the destination.
     * <p>
     * AThe resulting request entry containis allbuffered relevantby detailsthe toAsyncSinkWriter make a call to theand sent
     * destination.to Eg,the fordestination Kinesiswhen Datathe Streams{@code asubmitRequestEntries} requestmethod entryis
 contains the
   * invoked.
 * payload and partition key.*/
     * <p>private final ElementConverter<InputT, RequestEntryT> elementConverter;


     /**
 It seems more natural to* bufferBuffer InputT,to ie,hold therequest eventsentries that should be persisted into the
     * persisted,destination.
 rather than RequestEntryT. However, in practice, the response* <p>
     * ofA a failed request callentry cancontain makeall itrelevant verydetails hard,to ifmake nota impossible,call to the
     * reconstructdestination. theEg, originalfor event.Kinesis ItData isStreams mucha easier,request toentry justcontains construct athe
     * newpayload (retry)and request entry from the response and add that back to thepartition key.
     * <p>
     * queueIt forseems latermore retry.
natural to buffer InputT, ie, */
the events that should privatebe
 final Deque<RequestEntryT> bufferedRequests = new ConcurrentLinkedDeque<>();


    /**
  * persisted, rather than RequestEntryT. However, in practice, the response
     * of Tracksa failed allrequest pendingcall asynccan callsmake thatit havevery beenhard, executedif sincenot theimpossible, lastto
     * checkpoint reconstruct the original event. CallsIt is thatmuch alreadyeasier, completedto (successfullyjust orconstruct unsuccessfully)a
     * arenew automatically removed(retry) request entry from the queue.response Anyand requestadd entrythat thatback wasto notthe
     * successfullyqueue persistedfor needlater toretry.
 be handled and retried by*/
 the logic in
 private final Deque<RequestEntryT> bufferedRequestEntries *= {@code submitRequestsToApi}.new ArrayDeque<>();


     /** <p>
     * ThereTracks isall apending limitasync oncalls thethat numberhave ofbeen concurrentexecuted (async)since requests that can bethe last
     * handled by the client librarycheckpoint. ThisCalls limitthat iscompleted enforced(successfully byor checkingunsuccessfully) theare
     * size of this queue before issuing new requests.automatically decrementing the counter. Any request entry that was not
     * <p>
successfully persisted needs to be *handled Toand completeretried aby checkpoint,the welogic needin
 to make sure that no* requests are in{@code submitRequestsToApi}.
     * flight,<p>
 as they may fail, which* couldThere thenis leada tolimit dataon loss.
the number of concurrent  */
    private Queue<CompletableFuture<?>> inFlightRequests = new ConcurrentLinkedQueue<>();


    /**
     * Signals if enough RequestEntryTs have been buffered according to the user(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.
     * specified<p>
 buffering hints to make a* request against the destination. This
     * functionality will be added to the sink interface by means of an
     * additional FLIPTo 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 * @return a future that will be completed once there is are record available   
int inFlightRequestsCount;


    @Override
    public void write(InputT element, Context context) throws IOException, InterruptedException {
        * to make a request against the destination// blocks if too many elements have been buffered
     */
    public CompletableFuture<Void> isAvailable(while (bufferedRequestEntries.size() >= MAX_BUFFERED_REQUESTS_ENTRIES) {
         ...
    }


mailboxExecutor.yield();
    @Override
    public void write(InputT element, Context context) throws IOException {}

        bufferedRequestsbufferedRequestEntries.offerLastadd(elementConverter.apply(element, context));

    }


    /**
/ blocks if too many *async Therequests entireare requestin mayflight
 fail or single request entries that are part of  flush();
    }


     /**
 the request may not be* persistedPersists successfully,buffered eg,RequestsEntries becauseinto ofthe network
destination by invoking {@code
  * issues or service* sidesubmitRequestEntries} throttling.with Allbatches requestaccording entriesto thatthe faileduser withspecified
     * transientbuffering failureshints.
 need to be re-queued with*
 this method so that aren't
* The method blocks if *too lostmany andasync canrequests beare retriedin laterflight.
     * <p>/
    private *void Requestflush() entriesthrows thatInterruptedException are{
 causing the same error in a reproducible manner,
     * eg, ill-formed request entries, must not be re-queued but the error needs
     * to be handledwhile (bufferedRequestEntries.size() >= MAX_BATCH_SIZE) {

            // create a batch of request entries that should be persisted in the logicdestination
 of {@code submitRequestEntries}. Otherwise
     * these request entries willArrayList<RequestEntryT> bebatch retried= indefinitely, always causing thenew ArrayList<>(MAX_BATCH_SIZE);

     * same error.
     */
    protected void requeueFailedRequestEntry(RequestEntryT requestEntrywhile (batch.size() <= MAX_BATCH_SIZE && !bufferedRequestEntries.isEmpty()) {
        bufferedRequests.offerFirst(requestEntry);
     }


   try /**{
     * In-flight requests may   fail, but they will be retried if the sink is batch.add(bufferedRequestEntries.remove());
     * still healthy.
     * <p>
   } catch *(NoSuchElementException Toe) not{
 lose any requests, there cannot be any outstanding in-flight
     * requests when a commit is initialized.// Toif this end, all in-flight
     * requests need to be completed as part of the pre commit.
there are not enough elements, just create a smaller batch
                */
    @Overridebreak;
    public List<Collection<CompletableFuture<?>>> prepareCommit(boolean flush) throws IOException {

      }
  // reuse current inFlightRequests as commitable and create an empty queue }

        // to avoid  copyResultFuture<RequestEntryT> andrequestResult clearing=
        List<Collection<CompletableFuture<?>>> committable = Collections.singletonList(inFlightRequests);

        // allfailedRequestEntries in-flight requests are handled by the AsyncSinkCommiter and new 
> mailboxExecutor.execute(
               // elements cannot be added to the queue during a commit, so it's save() to -> completeRequest(failedRequestEntries),
        // create a new queue
        inFlightRequests = new ConcurrentLinkedQueue<>();

     "Mark in-flight request returnas committable;
completed and requeue %d }


    /**request entries",
     * 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
 failedRequestEntries.size());

            *while endpoint. These request entries are stored in the snapshot state so that
(inFlightRequestsCount >= MAX_IN_FLIGHT_REQUESTS) {
            * they don't get lost in case of a failure/restart of the application.
 mailboxExecutor.yield();
             */
}

      @Override
    public List<Collection<RequestEntryT>> snapshotState() throws IOException {
 inFlightRequestsCount++;
           return Collections.singletonList(bufferedRequests);submitRequestEntries(batch, requestResult);
        }
    }

    /**
     * Marks an in-flight request as completed and prepends failed requestEntries back to the
     * internal requestEntry buffer for later retry.
     *
     * @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.
        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 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) {
            flush();
        }

        // wait until all in-flight requests completed
        while (inFlightRequestsCount > 0) {
            mailboxExecutor.yield();
        }

        return Collections.emptyList();
    }


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