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

Class DetectNewPartitionsDoFn

java.lang.Object

org.apache.beam.sdk.transforms.DoFn<byte[],PartitionMetadata>

org.apache.beam.sdk.io.gcp.spanner.changestreams.dofn.DetectNewPartitionsDoFn

All Implemented Interfaces:

java.io.Serializable, HasDisplayData

@DoFn.UnboundedPerElement
public class DetectNewPartitionsDoFn
extends DoFn<byte[],PartitionMetadata>

A SplittableDoFn (SDF) that is responsible for scheduling partitions to be queried. This component will periodically scan the partition metadata table looking for partitions in the PartitionMetadata.State.CREATED, update their state to PartitionMetadata.State.SCHEDULED and output them to the next stage in the pipeline.

See Also:

Serialized Form

Nested Class Summary

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

DoFn.AlwaysFetched, DoFn.BoundedPerElement, DoFn.BundleFinalizer, DoFn.Element, DoFn.FieldAccess, DoFn.FinishBundle, DoFn.FinishBundleContext, DoFn.GetInitialRestriction, DoFn.GetInitialWatermarkEstimatorState, DoFn.GetRestrictionCoder, DoFn.GetSize, DoFn.GetWatermarkEstimatorStateCoder, DoFn.Key, DoFn.MultiOutputReceiver, DoFn.NewTracker, DoFn.NewWatermarkEstimator, DoFn.OnTimer, DoFn.OnTimerContext, DoFn.OnTimerFamily, DoFn.OnWindowExpiration, DoFn.OnWindowExpirationContext, DoFn.OutputReceiver<T>, DoFn.ProcessContext, DoFn.ProcessContinuation, DoFn.ProcessElement, DoFn.RequiresStableInput, DoFn.RequiresTimeSortedInput, DoFn.Restriction, DoFn.Setup, DoFn.SideInput, DoFn.SplitRestriction, DoFn.StartBundle, DoFn.StartBundleContext, DoFn.StateId, DoFn.Teardown, DoFn.TimerFamily, DoFn.TimerId, DoFn.Timestamp, DoFn.TruncateRestriction, DoFn.UnboundedPerElement, DoFn.WatermarkEstimatorState, DoFn.WindowedContext

Constructor Summary

Constructors

Constructor and Description
DetectNewPartitionsDoFn(DaoFactory daoFactory, MapperFactory mapperFactory, ChangeStreamMetrics metrics) This class needs a DaoFactory to build DAOs to access the partition metadata tables.
DetectNewPartitionsDoFn(DaoFactory daoFactory, MapperFactory mapperFactory, ChangeStreamMetrics metrics, Duration resumeDuration) This class needs a DaoFactory to build DAOs to access the partition metadata tables.

Method Summary

Modifier and Type	Method and Description
Instant	getInitialWatermarkEstimatorState(Instant currentElementTimestamp)
OffsetRange	initialRestriction() Uses an OffsetRange with a max range.
ManualWatermarkEstimator<Instant>	newWatermarkEstimator(Instant watermarkEstimatorState)
DoFn.ProcessContinuation	processElement(RestrictionTracker<OffsetRange,java.lang.Long> tracker, DoFn.OutputReceiver<PartitionMetadata> receiver, ManualWatermarkEstimator<Instant> watermarkEstimator) Main processing function for the DetectNewPartitionsDoFn function.
OffsetRangeTracker	restrictionTracker(OffsetRange restriction)
void	setup() Obtains the instances of PartitionMetadataDao and PartitionMetadataMapper from their respective factories.

Methods inherited from class org.apache.beam.sdk.transforms.DoFn

getAllowedTimestampSkew, getInputTypeDescriptor, getOutputTypeDescriptor, populateDisplayData, prepareForProcessing

Methods inherited from class java.lang.Object

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

Constructor Detail

DetectNewPartitionsDoFn

public DetectNewPartitionsDoFn(DaoFactory daoFactory,
                               MapperFactory mapperFactory,
                               ChangeStreamMetrics metrics)

This class needs a DaoFactory to build DAOs to access the partition metadata tables. It uses mappers to transform database rows into the PartitionMetadata model. It emits metrics for the partitions read using the ChangeStreamMetrics. This constructors sets the the periodic re-execution of the component to be scheduled using the DEFAULT_RESUME_DURATION duration (best effort).

Parameters:

daoFactory - the DaoFactory to construct PartitionMetadataDaos

mapperFactory - the MapperFactory to construct PartitionMetadataMappers

metrics - the ChangeStreamMetrics to emit partition related metrics

DetectNewPartitionsDoFn

public DetectNewPartitionsDoFn(DaoFactory daoFactory,
                               MapperFactory mapperFactory,
                               ChangeStreamMetrics metrics,
                               Duration resumeDuration)

This class needs a DaoFactory to build DAOs to access the partition metadata tables. It uses mappers to transform database rows into the PartitionMetadata model. It emits metrics for the partitions read using the ChangeStreamMetrics. It re-schedules the process element function to be executed according to the specified duration (best effort).

Parameters:

daoFactory - the DaoFactory to construct PartitionMetadataDaos

mapperFactory - the MapperFactory to construct PartitionMetadataMappers

metrics - the ChangeStreamMetrics to emit partition related metrics

resumeDuration - specifies the periodic schedule to re-execute this component

Method Detail

getInitialWatermarkEstimatorState

@DoFn.GetInitialWatermarkEstimatorState
public Instant getInitialWatermarkEstimatorState(@DoFn.Timestamp
                                                                                         Instant currentElementTimestamp)

newWatermarkEstimator

@DoFn.NewWatermarkEstimator
public ManualWatermarkEstimator<Instant> newWatermarkEstimator(@DoFn.WatermarkEstimatorState
                                                                                           Instant watermarkEstimatorState)

initialRestriction

@DoFn.GetInitialRestriction
public OffsetRange initialRestriction()

Uses an OffsetRange with a max range. This is because it does not know before hand how many partitions it will schedule.

In order to circumvent a bug in Apache Beam (https://issues.apache.org/jira/browse/BEAM-12756) we don't use Long.MAX_VALUE, but rather a value slightly smaller.

Returns:

the offset range for the component

restrictionTracker

@DoFn.NewTracker
public OffsetRangeTracker restrictionTracker(@DoFn.Restriction
                                                              OffsetRange restriction)

setup

@DoFn.Setup
public void setup()

Obtains the instances of PartitionMetadataDao and PartitionMetadataMapper from their respective factories.

processElement

@DoFn.ProcessElement
public DoFn.ProcessContinuation processElement(RestrictionTracker<OffsetRange,java.lang.Long> tracker,
                                                                    DoFn.OutputReceiver<PartitionMetadata> receiver,
                                                                    ManualWatermarkEstimator<Instant> watermarkEstimator)

Main processing function for the DetectNewPartitionsDoFn function. It follows this procedure periodically:

1. Fetches the min watermark from all the unfinished partitions in the metadata tables.
2. Updates the component's watermark to the min fetched.
3. Fetches all the partitions that are in PartitionMetadata.State.CREATED.
4. Updates the state of the partitions to PartitionMetadata.State.SCHEDULED.
5. Outputs the partitions to the next stage in the pipeline.
6. Schedule the function to resume after the configuration resume duration.

In the beginning of this function if there are no more unfinished partitions, it indicates that the work is complete. Thus, this function will not be re-scheduled.

Parameters:

tracker - an instance of OffsetRangeTracker

receiver - a PartitionMetadata OutputReceiver

watermarkEstimator - a ManualWatermarkEstimator of Instant

Returns:

a ProcessContinuation#stop() if there are no more partitions to process or ProcessContinuation#resume() to re-schedule the function after the configured interval.