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

Class ReadChangeStreamPartitionDoFn

java.lang.Object

org.apache.beam.sdk.transforms.DoFn<PartitionMetadata,DataChangeRecord>

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

All Implemented Interfaces:

java.io.Serializable, HasDisplayData

@DoFn.UnboundedPerElement
public class ReadChangeStreamPartitionDoFn
extends DoFn<PartitionMetadata,DataChangeRecord>
implements java.io.Serializable

A SDF (Splittable DoFn) class which is responsible for performing a change stream query for a given partition. A different action will be taken depending on the type of record received from the query. This component will also reflect the partition state in the partition metadata tables.

The processing of a partition is delegated to the QueryChangeStreamAction.

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
ReadChangeStreamPartitionDoFn(DaoFactory daoFactory, MapperFactory mapperFactory, ActionFactory actionFactory, ChangeStreamMetrics metrics) This class needs a DaoFactory to build DAOs to access the partition metadata tables and to perform the change streams query.

Method Summary

Modifier and Type	Method and Description
Instant	getInitialWatermarkEstimatorState(PartitionMetadata partition)
OffsetRange	initialRestriction(PartitionMetadata partition) The restriction for a partition will be defined from the start and end timestamp to query the partition for.
ReadChangeStreamPartitionRangeTracker	newTracker(PartitionMetadata partition, OffsetRange offsetRange)
ManualWatermarkEstimator<Instant>	newWatermarkEstimator(Instant watermarkEstimatorState)
DoFn.ProcessContinuation	processElement(PartitionMetadata partition, RestrictionTracker<OffsetRange,java.lang.Long> tracker, DoFn.OutputReceiver<DataChangeRecord> receiver, ManualWatermarkEstimator<Instant> watermarkEstimator, DoFn.BundleFinalizer bundleFinalizer) Performs a change stream query for a given partition.
void	setup() Constructs instances for the PartitionMetadataDao, ChangeStreamDao, ChangeStreamRecordMapper, PartitionMetadataMapper, DataChangeRecordAction, HeartbeatRecordAction, ChildPartitionsRecordAction and QueryChangeStreamAction.

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

ReadChangeStreamPartitionDoFn

public ReadChangeStreamPartitionDoFn(DaoFactory daoFactory,
                                     MapperFactory mapperFactory,
                                     ActionFactory actionFactory,
                                     ChangeStreamMetrics metrics)

This class needs a DaoFactory to build DAOs to access the partition metadata tables and to perform the change streams query. It uses mappers to transform database rows into the ChangeStreamRecord model. It uses the ActionFactory to construct the action dispatchers, which will perform the change stream query and process each type of record received. It emits metrics for the partition using the ChangeStreamMetrics.

Parameters:

daoFactory - the DaoFactory to construct PartitionMetadataDaos and ChangeStreamDaos

mapperFactory - the MapperFactory to construct ChangeStreamRecordMappers

actionFactory - the ActionFactory to construct actions

metrics - the ChangeStreamMetrics to emit partition related metrics

Method Detail

getInitialWatermarkEstimatorState

@DoFn.GetInitialWatermarkEstimatorState
public Instant getInitialWatermarkEstimatorState(@DoFn.Element
                                                                                         PartitionMetadata partition)

newWatermarkEstimator

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

initialRestriction

@DoFn.GetInitialRestriction
public OffsetRange initialRestriction(@DoFn.Element
                                                                  PartitionMetadata partition)

The restriction for a partition will be defined from the start and end timestamp to query the partition for. These timestamps are converted to microseconds. The OffsetRange restriction represents a closed-open interval, while the start / end timestamps represent a closed-closed interval, so we add 1 microsecond to the end timestamp to convert it to closed-open.

In this function we also update the partition state to PartitionMetadata.State#RUNNING.

Parameters:

partition - the partition to be queried

Returns:

the offset range from the partition start timestamp to the partition end timestamp + 1 microsecond

newTracker

@DoFn.NewTracker
public ReadChangeStreamPartitionRangeTracker newTracker(@DoFn.Element
                                                                         PartitionMetadata partition,
                                                                         @DoFn.Restriction
                                                                         OffsetRange offsetRange)

setup

@DoFn.Setup
public void setup()

Constructs instances for the PartitionMetadataDao, ChangeStreamDao, ChangeStreamRecordMapper, PartitionMetadataMapper, DataChangeRecordAction, HeartbeatRecordAction, ChildPartitionsRecordAction and QueryChangeStreamAction.

processElement

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

Performs a change stream query for a given partition. A different action will be taken depending on the type of record received from the query. This component will also reflect the partition state in the partition metadata tables.

The processing of a partition is delegated to the QueryChangeStreamAction.

Parameters:

partition - the partition to be queried

tracker - an instance of ReadChangeStreamPartitionRangeTracker

receiver - a DataChangeRecord OutputReceiver

watermarkEstimator - a ManualWatermarkEstimator of Instant

bundleFinalizer - the bundle finalizer

Returns:

a ProcessContinuation#stop() if a record timestamp could not be claimed or if the partition processing has finished