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> bufferedRequestEntries = new ConcurrentLinkedDeque<>(...);


    /**
     * Tracks all pending async calls that have been executed since the last
     * checkpoint. Calls that already completed (successfully or unsuccessfully)
     * are automatically removed from the queue. Any request entry that was not
     * successfully persisted need 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<>(...);




    /**@Override
    public *void Signals if enough RequestEntryTs have been buffered according to the user
write(InputT element, Context context) throws IOException, InterruptedException {        *
 specified buffering hints to make a request against the destination. This
  bufferedRequestEntries.putLast(elementConverter.apply(element, context));

		
    }


    /**
 functionality will be added to* theThe sinkentire interfacerequest bymay meansfail ofor an
single request entries that are * additional FLIP.part of
     *
 the request may not *be @returnpersisted asuccessfully, futureeg, thatbecause willof benetwork
 completed once a  request* againstissues the
or service side throttling. All *request destinationentries canthat befailed madewith
     */
 transient failures need publicto CompletableFuture<Void> isAvailable() {
        ...
    }


    @Overridebe re-queued with this method so that aren't
     * lost and can be retried later.
    public void write(InputT element, Context context) throws IOException {* <p>
     * Request  bufferedRequestEntries.offerLast(elementConverter.apply(element, context));
    }


    /**entries that are causing the same error in a reproducible manner,
     * The entireeg, ill-formed request mayentries, failmust ornot singlebe requestre-queued entriesbut thatthe areerror part ofneeds
     * theto requestbe mayhandled notin bethe persistedlogic successfully, eg, because of networkof {@code submitRequestEntries}. Otherwise
     * issuesthese orrequest serviceentries sidewill throttling.be Allretried requestindefinitely, entriesalways thatcausing failed withthe
     * transientsame failureserror.
 need to be re-queued with this method so that aren't
     * lost and can be retried later. */
    protected void requeueFailedRequestEntry(RequestEntryT requestEntry) {
        bufferedRequestEntries.offerFirst(requestEntry);
    }


    /** <p>
     * RequestIn-flight entriesrequests thatmay arefail, causingbut thethey samewill be errorretried inif athe reproduciblesink manner,is
     * eg, ill-formed request entries, must not be re-queued but the error needs
     * to be handled in the logic of {@code submitRequestEntries}. Otherwise
     * these request entries will be retried indefinitely, always causing the
     * same error still healthy.
     * <p>
     * To not lose any requests, there cannot be any outstanding in-flight
     * requests when a commit is initialized. To this end, all in-flight
     * requests need to be completed as part of the pre commit.
     */
    protected void requeueFailedRequestEntry(RequestEntryT requestEntry) {@Override
    public List<Collection<CompletableFuture<?>>>   bufferedRequestEntries.offerFirst(requestEntry);
    }


prepareCommit(boolean flush) throws IOException {

    /**
    // *reuse In-flightcurrent requestsinFlightRequests mayas fail,commitable butand theycreate willan beempty retriedqueue if
 the sink is
     *// stillto healthy.
avoid copy and clearing
  * <p>
     * To not lose any requests, there cannot be any outstandingList<Collection<CompletableFuture<?>>> committable = Collections.singletonList(inFlightRequests);

        // all in-flight
 requests are handled by *the requestsAsyncSinkCommiter whenand anew commit
 is initialized. To this end, all in-flight
 // elements cannot be *added requeststo needthe toqueue beduring completeda ascommit, partso ofit's thesave preto commit.
     *   //
 create a new @Overridequeue
     public List<Collection<CompletableFuture<?>>> prepareCommit(boolean flush)inFlightRequests throws= IOException {new ConcurrentLinkedQueue<>();

        // reuse current inFlightRequests as commitable and create an empty queue 
        // to avoid copy and clearingreturn committable;
    }


    /**
     * All in-flight requests have been completed, but there may still be
     * request entries List<Collection<CompletableFuture<?>>>in committablethe = Collections.singletonList(inFlightRequests);

        // all in-flight requests are handled by the AsyncSinkCommiter and new 
        // elements cannot be added to the queue during a commit, so it's save to 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.
     */
   // create@Override
 a new queue
 public List<Collection<RequestEntryT>> snapshotState() throws    inFlightRequests = new ConcurrentLinkedQueue<>();

IOException {
        return committableCollections.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.
     */

            .builder()
                .data(SdkBytes.fromUtf8String(element.toString()))
                .partitionKey(String.valueOf(element.hashCode()))
                .build();
    }
}

private class AmazonKinesisDataStreamWriter extends AsyncSinkWriter<InputT, PutRecordsRequestEntry> {

    @Override
    protected CompletableFuture<?> submitRequestEntries(List<PutRecordsRequestEntry> requestEntries
new ElementConverter<InputT, PutRecordsRequestEntry>() {

    @Override
    public// PutRecordsRequestEntrycreate apply(InputT element, SinkWriter.Context context) {
a batch request
        PutRecordsRequest batchRequest return= PutRecordsRequestEntryPutRecordsRequest
                .builder()
                .data(SdkBytes.fromUtf8String(element.toString())records(requestEntries)
                .partitionKey(String.valueOf(element.hashCode()))
  streamName(streamName)
              .build();
    }
}