org.apache.beam.sdk.io.gcp.spanner.changestreams.restriction

Class TimestampRangeTracker

java.lang.Object

org.apache.beam.sdk.transforms.splittabledofn.RestrictionTracker<TimestampRange,com.google.cloud.Timestamp>

org.apache.beam.sdk.io.gcp.spanner.changestreams.restriction.TimestampRangeTracker

All Implemented Interfaces:

RestrictionTracker.HasProgress

Direct Known Subclasses:

DetectNewPartitionsRangeTracker, ReadChangeStreamPartitionRangeTracker

public class TimestampRangeTracker
extends RestrictionTracker<TimestampRange,com.google.cloud.Timestamp>
implements RestrictionTracker.HasProgress

A RestrictionTracker for claiming positions in a TimestampRange in a monotonically increasing fashion.

The smallest position is Timestamp.MIN_VALUE and the largest position is Timestamp.MAX_VALUE - 1 nanosecond.

Nested Class Summary

Nested classes/interfaces inherited from class org.apache.beam.sdk.transforms.splittabledofn.RestrictionTracker

RestrictionTracker.HasProgress, RestrictionTracker.IsBounded, RestrictionTracker.Progress, RestrictionTracker.TruncateResult<RestrictionT>

Field Summary

Fields

Modifier and Type	Field and Description
protected @Nullable com.google.cloud.Timestamp	lastAttemptedPosition
protected @Nullable com.google.cloud.Timestamp	lastClaimedPosition
protected TimestampRange	range
protected java.util.function.Supplier<com.google.cloud.Timestamp>	timeSupplier

Constructor Summary

Constructors

Constructor and Description
TimestampRangeTracker(TimestampRange range)

Method Summary

Modifier and Type	Method and Description
void	checkDone() Checks if the restriction has been processed successfully.
TimestampRange	currentRestriction() Returns a restriction accurately describing the full range of work the current DoFn.ProcessElement call will do, including already completed work.
RestrictionTracker.Progress	getProgress() Returns the progress made within the restriction so far.
RestrictionTracker.IsBounded	isBounded() Return the boundedness of the current restriction.
void	setTimeSupplier(java.util.function.Supplier<com.google.cloud.Timestamp> timeSupplier)
boolean	tryClaim(com.google.cloud.Timestamp position) Attempts to claim the given position.
@Nullable SplitResult<TimestampRange>	trySplit(double fractionOfRemainder) Splits the restriction through the following algorithm:

Methods inherited from class java.lang.Object

clone, equals, finalize, getClass, hashCode, notify, notifyAll, toString, wait, wait, wait

Field Detail

range

protected TimestampRange range

lastAttemptedPosition

protected @Nullable com.google.cloud.Timestamp lastAttemptedPosition

lastClaimedPosition

protected @Nullable com.google.cloud.Timestamp lastClaimedPosition

timeSupplier

protected java.util.function.Supplier<com.google.cloud.Timestamp> timeSupplier

Constructor Detail

TimestampRangeTracker

public TimestampRangeTracker(TimestampRange range)

Method Detail

setTimeSupplier

public void setTimeSupplier(java.util.function.Supplier<com.google.cloud.Timestamp> timeSupplier)

tryClaim

public boolean tryClaim(com.google.cloud.Timestamp position)

Attempts to claim the given position. Depending on the position following outcomes are possible:

• If the position is less than or equal to a previous attempted one, an IllegalArgumentException will be thrown
• If the position is less than the restriction range start, an IllegalArgumentException will be thrown
• If the position is greater than or equal to the range end, the position will not be claimed
• If the position is greater or equal to the range start and less than the range end, the position will be claimed

If an error is not thrown, this function will register the position as the lastAttemptedPosition.

Specified by:

tryClaim in class RestrictionTracker<TimestampRange,com.google.cloud.Timestamp>

Returns:

true if the position was successfully claimed, false otherwise

trySplit

public @Nullable SplitResult<TimestampRange> trySplit(double fractionOfRemainder)

Splits the restriction through the following algorithm:

    currentPosition = lastAttemptedPosition == null ? (from - 1ns) : lastAttemptedPosition
    splitPosition = currentPosition + max(1, (range.to - currentPosition) * fractionOfRemainder)
    primary = [range.from, splitPosition)
    residual = [splitPosition, range.to)
    this.range = primary
 

If the splitPosition is greater than the range.to, null will be returned. For checkpoints the fractionOfRemainder will always be zero.

Specified by:

trySplit in class RestrictionTracker<TimestampRange,com.google.cloud.Timestamp>

Parameters:

fractionOfRemainder - A hint as to the fraction of work the primary restriction should represent based upon the current known remaining amount of work.

Returns:

a SplitResult if a split was possible or null if the splitPosition is beyond the end of the range.

checkDone

public void checkDone()
               throws java.lang.IllegalStateException

Checks if the restriction has been processed successfully. If not, throws an IllegalStateException.

The restriction is considered processed successfully if:

• The range is empty (range.from == range.to)
• The lastAttemptedPosition + 1ns >= range.to

The restriction is considered not processed successfully if:

• No position claim was attempted for a non-empty range
• The lastAttemptedPosition + 1ns < range.to

Specified by:

checkDone in class RestrictionTracker<TimestampRange,com.google.cloud.Timestamp>

Throws:

java.lang.IllegalStateException

getProgress

public RestrictionTracker.Progress getProgress()

Returns the progress made within the restriction so far. If lastAttemptedPosition is null, the start of the range is used as the completed work; otherwise, lastAttemptedPosition will be used. The time gap between lastAttemptedPosition and now is used as the remaining work. In this way, when the time gap becomes large, we will have more backlog to process and we should add more resources.

Specified by:

getProgress in interface RestrictionTracker.HasProgress

Returns:

work completed and work remaining in seconds.

currentRestriction

public TimestampRange currentRestriction()

Description copied from class: RestrictionTracker

Returns a restriction accurately describing the full range of work the current DoFn.ProcessElement call will do, including already completed work.

The current restriction returned by method may be updated dynamically due to due to concurrent invocation of other methods of the RestrictionTracker, For example, RestrictionTracker.trySplit(double).

This method is required to be implemented.

Specified by:

currentRestriction in class RestrictionTracker<TimestampRange,com.google.cloud.Timestamp>

isBounded

public RestrictionTracker.IsBounded isBounded()

Description copied from class: RestrictionTracker

Return the boundedness of the current restriction. If the current restriction represents a finite amount of work, it should return RestrictionTracker.IsBounded.BOUNDED. Otherwise, it should return RestrictionTracker.IsBounded.UNBOUNDED.

It is valid to return RestrictionTracker.IsBounded.BOUNDED after returning RestrictionTracker.IsBounded.UNBOUNDED once the end of a restriction is discovered. It is not valid to return RestrictionTracker.IsBounded.UNBOUNDED after returning RestrictionTracker.IsBounded.BOUNDED.

This method is required to be implemented.

Specified by:

isBounded in class RestrictionTracker<TimestampRange,com.google.cloud.Timestamp>