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 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 BlockingDeque<CompletableFuture<?>> inFlightRequests = new LinkedBlockingDeque<>(...)int inFlightRequestsCount;


    @Override
    public void write(InputT element, Context context) throws IOException, InterruptedException {
        // blocks if too many eventselements have been buffered
        while (bufferedRequestEntries.putLast(elementConverter.apply(element, context));
size() >= MAX_BUFFERED_REQUESTS_ENTRIES) {
        // blocks if too many async requests are in flight
     mailboxExecutor.yield();
        }

    if (/*buffering hints are met*/) {
			flush( bufferedRequestEntries.add(elementConverter.apply(element, context));
		}
    }


    /**
/ blocks if too many async * The entire request may fail or single request entries that are part ofrequests are in flight
        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 entriesArrayList<RequestEntryT> willbatch be= retried indefinitely, always causing thenew ArrayList<>(MAX_BATCH_SIZE);

     * same error.
     */
    protected void requeueFailedRequestEntry(RequestEntryT requestEntry) throws InterruptedException while (batch.size() <= MAX_BATCH_SIZE && !bufferedRequestEntries.isEmpty()) {
        bufferedRequestEntries.putFirst(requestEntry);
     }


   try /**{
     * Persists buffered RequestsEntries into the destination by invoking {@code
     * submitRequestEntries} with batches according to the user specified
 batch.add(bufferedRequestEntries.remove());
            * buffering hints.
  } catch  *(NoSuchElementException e) {
     * The method blocks if too many async requests are in flight.
     *//
 if there  private void flush() throws InterruptedException {
are not enough elements, just create a smaller batch
         while (bufferedRequestEntries.size() >= MAX_BATCH_SIZE) {
       break;
     // create a batch of request entries that should be persisted in}
 the destination
          }

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

      ResultFuture<RequestEntryT> requestResult =
    for (int i = 0; i < MAX_BATCH_SIZE; i++) {
       failedRequestEntries -> mailboxExecutor.execute(
       try {
                    batch.add(bufferedRequestEntries.remove());() -> completeRequest(failedRequestEntries),
                } catch (NoSuchElementException e) {
        "Mark in-flight request as completed and requeue %d request entries",
   // if there are not enough elements, just create a smaller batch
                    break;failedRequestEntries.size());

            while (inFlightRequestsCount   }>= MAX_IN_FLIGHT_REQUESTS) {
            }

    mailboxExecutor.yield();
        // call the destination specific}

 code that actually persists the request entries
     inFlightRequestsCount++;
       CompletableFuture<?>   future = submitRequestEntries(batch, requestResult);

        }
    // keep track of in flight request}


    /**
     * Marks an     inFlightRequests.put(future);

      in-flight request as completed and prepends failed requestEntries back to the
     * //internal removerequestEntry thebuffer requestfor fromlater theretry.
 tracking queue once it competed*
     * @param failedRequestEntries requestEntries that need  future.whenComplete((response, err) -> {to be retried
     */
    private void  completeRequest(Collection<RequestEntryT>    inFlightRequests.remove(future);failedRequestEntries) {
        inFlightRequestsCount--;

    });
    // By just iterating }
through failedRequestEntries, it reverses }


the order of  /**the
     * In-flight requests may fail, but they will be retried if the sink is
// failedRequestEntries. It doesn't make a difference for kinesis:putRecords, as the api
      * still healthy.
// does not make any *order <p>
guarantees, but may cause avoidable *reorderings Tofor notother
 lose any requests, there cannot be any outstanding in-flight// destinations.
     * requests when a commit is initialized. To this end, all in-flight failedRequestEntries.forEach(bufferedRequestEntries::addFirst);
    }


    /**
     * In flight requests needwill to be completedretried asif partthe ofsink theis prestill commithealthy.
 But if   */in-flight requests
    @Override
 * fail after publica List<Collection<CompletableFuture<?>>> prepareCommit(boolean flush) throws IOException {

        // reuse current inFlightRequests as commitable and create an empty queue 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 towhen avoida copycommit andis clearinginitialized.
     *
    List<Collection<CompletableFuture<?>>> committable* = Collections.singletonList(inFlightRequests);

        //<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 LinkedBlockingDeque<>flush();

        return committable;}

    }


    /**
/ wait until  all * All in-flight requests have been completed,
 but there may still be
   while (inFlightRequestsCount *> request0) 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* lostAll in case of a failure/restart of the application.-flight requests have been completed, but there may still be
     */
 request entries in @Override
the internal buffer that publicare yet List<Collection<RequestEntryT>>to snapshotState()be throwssent IOExceptionto {the
     * endpoint.  return Collections.singletonList(bufferedRequestEntries);
    }

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
new ElementConverter<InputT, PutRecordsRequestEntry>() {
    @Override
    public PutRecordsRequestEntry apply(InputT element, SinkWriter.Context context) {
        return PutRecordsRequestEntry
    protected void submitRequestEntries(List<PutRecordsRequestEntry> requestEntries, ResultFuture<PutRecordsRequestEntry>        .builder()
                .data(SdkBytes.fromUtf8String(element.toString()))requestResult) {

         // create a batch  request
  .partitionKey(String.valueOf(element.hashCode()))
      PutRecordsRequest batchRequest = PutRecordsRequest
       .build();
    }
}