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 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 must be checked before issuninis 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 int inFlightRequestsCount;


    Queue<CompletableFuture<?>> inFlightRequests = new ConcurrentLinkedQueue<>();


    /**@Override
    public void write(InputT element, Context context) throws IOException, InterruptedException {
     * Signals   // blocks if too enoughmany RequestEntryTselements have been buffered
 according to the user
     * specified buffering hints to make a request against the destination. This
     * functionality will be added to the sink interface by means of an
     * additional FLIP.
     *
     * @return a future that will be completed once there is are record available   
     * to make a request against the destinationwhile (bufferedRequestEntries.size() >= MAX_BUFFERED_REQUESTS_ENTRIES) {
            mailboxExecutor.yield();
        }

        bufferedRequestEntries.add(elementConverter.apply(element, context));

        // blocks if too many async requests are in flight
        flush();
    }


    /**
     * Persists buffered RequestsEntries into the destination by invoking {@code
     */
 submitRequestEntries} with batches according publicto CompletableFuture<Void>the isAvailable()user {specified
     * buffering  hints...
    }

 *
    @Override
 * The method publicblocks voidif write(InputT element, Context context) throws IOException {too many async requests are in flight.
     */
   bufferedRequests.offerLast(elementConverter.apply(element, context));
    }


 private void flush() throws InterruptedException {
    /**
     * The entire request may fail or single request entries that are part of
     * the request may notwhile (bufferedRequestEntries.size() >= MAX_BATCH_SIZE) {

            // create a batch of request entries that should be persisted successfully,in eg, because of networkthe destination
     * issues or service side throttling. All requestArrayList<RequestEntryT> entriesbatch that= failed withnew ArrayList<>(MAX_BATCH_SIZE);

     * transient failures need to be re-queued with this method so that aren'twhile (batch.size() <= MAX_BATCH_SIZE && !bufferedRequestEntries.isEmpty()) {
     * lost and can be retried later.
     *try <p>{
     *    Request entries that are causing the same error in a reproducible manner, batch.add(bufferedRequestEntries.remove());
     *  eg, ill-formed request entries, must not be re-queued but the error needs
   } catch (NoSuchElementException e) {
      * to be handled in the logic of {@code submitRequestEntries}. Otherwise
    // *if thesethere requestare entriesnot willenough beelements, retriedjust indefinitely,create alwaysa causingsmaller thebatch
     *  same error.
     */
    protected void requeueFailedRequestEntry(RequestEntryT requestEntry) { break;
        bufferedRequests.offerFirst(requestEntry);
    }


    /**}
     *  In-flight requests may fail, but they}

 will be retried if the sink is
     *ResultFuture<RequestEntryT> stillrequestResult healthy.=
     * <p>
     * To not lose any requests, there cannot be any outstandingfailedRequestEntries in-flight
 > mailboxExecutor.execute(
    * requests when a commit is initialized. To this end, all in-flight
     * requests need to be completed as part of() the pre commit.-> completeRequest(failedRequestEntries),
     */
    @Override
    public List<Collection<CompletableFuture<?>>> prepareCommit(boolean flush) throws IOException {

        // reuse current inFlightRequests"Mark in-flight request as commitablecompleted and create an empty queue 
 requeue %d request entries",
            // to avoid copy and clearing
        List<Collection<CompletableFuture<?>>> committable = CollectionsfailedRequestEntries.singletonListsize(inFlightRequests));

        // all in-flight requests arewhile handled(inFlightRequestsCount by the AsyncSinkCommiter and new 
>= MAX_IN_FLIGHT_REQUESTS) {
           // elements cannot be added to the queue during a commit, so it's save to    mailboxExecutor.yield();
            }

        // create a new queueinFlightRequestsCount++;
        inFlightRequests  = new ConcurrentLinkedQueue<>submitRequestEntries(batch, requestResult);

        return committable;}
    }


    /**
     * Marks Allan in-flight requestsrequest have beenas completed, butand thereprepends mayfailed still berequestEntries back to the
     * requestinternal entries in the internal requestEntry buffer thatfor arelater yetretry.
 to be sent to the*
     * endpoint. These request entries are stored in the snapshot state so that@param failedRequestEntries requestEntries that need to be retried
     */
    private *void they don't get lost in case of a failure/restart of the application.
completeRequest(Collection<RequestEntryT> failedRequestEntries) {
        inFlightRequestsCount--;

      */
  // By @Override
just iterating through failedRequestEntries, publicit List<Collection<RequestEntryT>>reverses snapshotState()the throwsorder IOExceptionof {the
        return// Collections.singletonList(bufferedRequests);
    }

new ElementConverter<InputT, PutRecordsRequestEntry>() {
    @Override
    public PutRecordsRequestEntry apply(InputT element, SinkWriter.Context context) {
        return PutRecordsRequestEntry
           // re-queue elements of failed .builder()requests
                .data(SdkBytes.fromUtf8String(element.toString()))
future.whenComplete((response, err) -> {
                if .partitionKey(String.valueOf(element.hashCoderesponse.failedRecordCount()))
 > 0) {
                  .build();
    }
}

public class AmazonKinesisDataStreamSink<InputT> extends AsyncSink<InputT, PutRecordsRequestEntry> {

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

  ArrayList<PutRecordsRequestEntry> failedRequestEntries = new ArrayList<>(response.failedRecordCount());
                    List<PutRecordsResultEntry> records = response.records();
    
         // create a batch request
       for PutRecordsRequest(int batchRequesti = PutRecordsRequest
0; i < records.size();    i++) {
         .builder()
               if .records(requestEntries)
   (records.get(i).errorCode() != null) {
             .streamName(streamName)
                failedRequestEntries.buildadd(requestEntries.get(i));

        // call api with batch request
        CompletableFuture<PutRecordsResponse> future = client.putRecords(batchRequest);

 }
        // re-queue elements of failed requests
       }
 CompletableFuture<PutRecordsResponse> handleResponse = future
            .whenComplete((response, err) -> {
         requestResult.complete(failedRequestEntries);
       if (response.failedRecordCount() > 0) {
       } else {
           List<PutRecordsResultEntry> records = response.records();

      requestResult.complete(Collections.emptyList());
              for (int i}

 = 0; i < records.size(); i++) {
         //TODO: handle errors of the entire request...
         if (records.get(i).errorCode() != null});
 {
                            requeueFailedRequest(requestEntries.get(i));}

    ...
}