Package org.apache.beam.sdk.io.gcp.bigtable.changestreams.restriction

Class ReadChangeStreamPartitionProgressTracker

java.lang.Object
org.apache.beam.sdk.transforms.splittabledofn.RestrictionTracker<StreamProgress,StreamProgress>
org.apache.beam.sdk.io.gcp.bigtable.changestreams.restriction.ReadChangeStreamPartitionProgressTracker
@Internal public class ReadChangeStreamPartitionProgressTracker extends RestrictionTracker<StreamProgress,StreamProgress>
RestrictionTracker used by ReadChangeStreamPartitionDoFn to keep track of the progress of the stream and to split the restriction for runner initiated checkpoints.

StreamProgress usually is a continuation token which represents a position in time of the stream of a specific partition. The token is used to resume streaming a partition.

On ChangeStreamMutation or Heartbeat response, the tracker will try to claim the continuation token from the response. The tracker stores that continuation token (wrapped in a StreamProgress) so that if the DoFn checkpoints or restarts, the token can be used to resume the stream.

The tracker will fail to claim a token if runner has initiated a checkpoint (by calling trySplit(0)). This signals to the DoFn to stop.

When runner initiates a checkpoint, the tracker returns null for the primary split and the residual split includes the entire token. The next time the DoFn try to claim a new StreamProgress, it will fail, and stop. The residual will be scheduled on a new DoFn to resume the work from the previous StreamProgress

  • Constructor Details

    • ReadChangeStreamPartitionProgressTracker

      public ReadChangeStreamPartitionProgressTracker(StreamProgress streamProgress)
      Constructs a restriction tracker with the streamProgress.
      Parameters:
      streamProgress - represents a position in time of the stream.

  • Method Details

    • isBounded

      public RestrictionTracker.IsBounded isBounded()
      This restriction tracker is for unbounded streams.
      Specified by:
      isBounded in class RestrictionTracker<StreamProgress,StreamProgress>
      Returns:
      invalid reference 
      IsBounded.UNBOUNDED

    • checkDone

      public void checkDone() throws IllegalStateException
      This is to signal to the runner that this restriction has completed. Throw an exception if there is more work to be done, and it should not stop. A restriction tracker stops after a runner initiated checkpoint or the streamProgress contains a closeStream response and not a token.
      Specified by:
      checkDone in class RestrictionTracker<StreamProgress,StreamProgress>
      Throws:
      IllegalStateException - when the restriction is not done and there is more work to be done.

    • tryClaim

      public boolean tryClaim(StreamProgress streamProgress)
      Claims a new StreamProgress to be processed. StreamProgress can either be a ContinuationToken or a CloseStream.

      The claim fails if the runner has previously initiated a checkpoint. The restriction tracker respects the runner initiated checkpoint and fails to claim this streamProgress. The new split will start from the previously successfully claimed streamProgress.

      Specified by:
      tryClaim in class RestrictionTracker<StreamProgress,StreamProgress>
      Parameters:
      streamProgress - position in time of the stream that is being claimed.
      Returns:
      true if claim was successful, otherwise false.

    • currentRestriction

      public StreamProgress currentRestriction()
      Returns the streamProgress that was successfully claimed.
      Specified by:
      currentRestriction in class RestrictionTracker<StreamProgress,StreamProgress>
      Returns:
      the streamProgress that was successfully claimed.

    • trySplit

      public @Nullable SplitResult<StreamProgress> trySplit(double fractionOfRemainder)
      Splits the work that's left. Since the position in the stream isn't a contiguous value, we cannot estimate how much work is left or breakdown the work into smaller chunks. Therefore, there's no way to split the work. To conform to the API, we return null for the primary split and then continue the work on the residual split.

      Also note that, we only accept checkpoints (fractionOfRemainder = 0). Any other value, we reject (by returning null) the request to split since StreamProgress cannot be broken down into fractions.

      Specified by:
      trySplit in class RestrictionTracker<StreamProgress,StreamProgress>
      Parameters:
      fractionOfRemainder - the fraction of work remaining, where 0 is a request to checkpoint current work.
      Returns:
      split result when fractionOfRemainder = 0, otherwise null.

    • toString

      public String toString()
      Overrides:
      toString in class Object