org.apache.flink.streaming.connectors.kinesis.internals

Class KinesisDataFetcher<T>

java.lang.Object

org.apache.flink.streaming.connectors.kinesis.internals.KinesisDataFetcher<T>

Direct Known Subclasses:

DynamoDBStreamsDataFetcher

@Internal
public class KinesisDataFetcher<T>
extends Object

A KinesisDataFetcher is responsible for fetching data from multiple Kinesis shards. Each parallel subtask instantiates and runs a single fetcher throughout the subtask's lifetime. The fetcher accomplishes the following:

• 1. continuously poll Kinesis to discover shards that the subtask should subscribe to. The subscribed subset of shards, including future new shards, is non-overlapping across subtasks (no two subtasks will be subscribed to the same shard) and determinate across subtask restores (the subtask will always subscribe to the same subset of shards even after restoring)
• 2. decide where in each discovered shard should the fetcher start subscribing to
• 3. subscribe to shards by creating a single thread for each shard

The fetcher manages two states: 1) last seen shard ids of each subscribed stream (used for continuous shard discovery), and 2) last processed sequence numbers of each subscribed shard. Since operations on the second state will be performed by multiple threads, these operations should only be done using the handler methods provided in this class.

Nested Class Summary

Nested Classes

Modifier and Type	Class and Description
static interface	KinesisDataFetcher.FlinkKinesisProxyFactory Factory to create Kinesis proxy instances used by a fetcher.
static interface	KinesisDataFetcher.FlinkKinesisProxyV2Factory Factory to create Kinesis proxy V@ instances used by a fetcher.

Field Summary

Fields

Modifier and Type	Field and Description
static KinesisShardAssigner	DEFAULT_SHARD_ASSIGNER

Constructor Summary

Constructors

Modifier	Constructor and Description
protected	KinesisDataFetcher(List<String> streams, SourceFunction.SourceContext<T> sourceContext, Object checkpointLock, RuntimeContext runtimeContext, Properties configProps, KinesisDeserializationSchema<T> deserializationSchema, KinesisShardAssigner shardAssigner, AssignerWithPeriodicWatermarks<T> periodicWatermarkAssigner, WatermarkTracker watermarkTracker, AtomicReference<Throwable> error, List<KinesisStreamShardState> subscribedShardsState, HashMap<String,String> subscribedStreamsToLastDiscoveredShardIds, KinesisDataFetcher.FlinkKinesisProxyFactory kinesisProxyFactory, KinesisDataFetcher.FlinkKinesisProxyV2Factory kinesisProxyV2Factory)
	KinesisDataFetcher(List<String> streams, SourceFunction.SourceContext<T> sourceContext, RuntimeContext runtimeContext, Properties configProps, KinesisDeserializationSchema<T> deserializationSchema, KinesisShardAssigner shardAssigner, AssignerWithPeriodicWatermarks<T> periodicWatermarkAssigner, WatermarkTracker watermarkTracker) Creates a Kinesis Data Fetcher.

Method Summary

Modifier and Type	Method and Description
void	advanceLastDiscoveredShardOfStream(String stream, String shardId) Updates the last discovered shard of a subscribed stream; only updates if the update is valid.
void	awaitTermination() After calling shutdownFetcher(), this can be called to await the fetcher shutdown.
protected void	closeRecordPublisherFactory() Closes recordRecordPublisherFactory.
static StreamShardHandle	convertToStreamShardHandle(StreamShardMetadata streamShardMetadata) Utility function to convert StreamShardMetadata into StreamShardHandle.
static StreamShardMetadata	convertToStreamShardMetadata(StreamShardHandle streamShardHandle) Utility function to convert StreamShardHandle into StreamShardMetadata.
protected static HashMap<String,String>	createInitialSubscribedStreamsToLastDiscoveredShardsState(List<String> streams) Utility function to create an initial map of the last discovered shard id of each subscribed stream, set to null; This is called in the constructor; correct values will be set later on by calling advanceLastDiscoveredShardOfStream().
protected RecordPublisher	createRecordPublisher(SequenceNumber sequenceNumber, Properties configProps, MetricGroup metricGroup, StreamShardHandle subscribedShard)
protected RecordPublisherFactory	createRecordPublisherFactory()
protected ShardConsumer<T>	createShardConsumer(Integer subscribedShardStateIndex, StreamShardHandle subscribedShard, SequenceNumber lastSequenceNum, MetricGroup metricGroup, KinesisDeserializationSchema<T> shardDeserializer) Create a new shard consumer.
protected ExecutorService	createShardConsumersThreadPool(String subtaskName)
protected void	deregisterStreamConsumer() Deregisters stream consumers.
List<StreamShardHandle>	discoverNewShardsToSubscribe() A utility function that does the following:
protected void	emitRecordAndUpdateState(T record, long recordTimestamp, int shardStateIndex, SequenceNumber lastSequenceNumber) Prepare a record and hand it over to the RecordEmitter, which may collect it asynchronously.
protected void	emitWatermark() Called periodically to emit a watermark.
protected Properties	getConsumerConfiguration()
protected long	getCurrentTimeMillis() Return the current system time.
List<KinesisStreamShardState>	getSubscribedShardsState()
static boolean	isThisSubtaskShouldSubscribeTo(int shardHash, int totalNumberOfConsumerSubtasks, int indexOfThisConsumerSubtask) Utility function to determine whether a shard should be subscribed by this consumer subtask.
int	registerNewSubscribedShardState(KinesisStreamShardState newSubscribedShardState) Register a new subscribed shard state.
void	runFetcher() Starts the fetcher.
protected boolean	shouldAdvanceLastDiscoveredShardId(String shardId, String lastSeenShardIdOfStream) Given lastSeenShardId, check if last discovered shardId should be advanced.
void	shutdownFetcher() Starts shutting down the fetcher.
HashMap<StreamShardMetadata,SequenceNumber>	snapshotState() Creates a snapshot of the current last processed sequence numbers of each subscribed shard.
protected void	stopWithError(Throwable throwable) Called by created threads to pass on errors.
protected void	updateState(int shardStateIndex, SequenceNumber lastSequenceNumber) Update the shard to last processed sequence number state.

Methods inherited from class java.lang.Object

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

Field Detail

DEFAULT_SHARD_ASSIGNER

public static final KinesisShardAssigner DEFAULT_SHARD_ASSIGNER

Constructor Detail

KinesisDataFetcher

public KinesisDataFetcher(List<String> streams,
                          SourceFunction.SourceContext<T> sourceContext,
                          RuntimeContext runtimeContext,
                          Properties configProps,
                          KinesisDeserializationSchema<T> deserializationSchema,
                          KinesisShardAssigner shardAssigner,
                          AssignerWithPeriodicWatermarks<T> periodicWatermarkAssigner,
                          WatermarkTracker watermarkTracker)

Creates a Kinesis Data Fetcher.

Parameters:

streams - the streams to subscribe to

sourceContext - context of the source function

runtimeContext - this subtask's runtime context

configProps - the consumer configuration properties

deserializationSchema - deserialization schema

KinesisDataFetcher

@VisibleForTesting
protected KinesisDataFetcher(List<String> streams,
                                                SourceFunction.SourceContext<T> sourceContext,
                                                Object checkpointLock,
                                                RuntimeContext runtimeContext,
                                                Properties configProps,
                                                KinesisDeserializationSchema<T> deserializationSchema,
                                                KinesisShardAssigner shardAssigner,
                                                AssignerWithPeriodicWatermarks<T> periodicWatermarkAssigner,
                                                WatermarkTracker watermarkTracker,
                                                AtomicReference<Throwable> error,
                                                List<KinesisStreamShardState> subscribedShardsState,
                                                HashMap<String,String> subscribedStreamsToLastDiscoveredShardIds,
                                                KinesisDataFetcher.FlinkKinesisProxyFactory kinesisProxyFactory,
                                                @Nullable
                                                KinesisDataFetcher.FlinkKinesisProxyV2Factory kinesisProxyV2Factory)

Method Detail

createShardConsumer

protected ShardConsumer<T> createShardConsumer(Integer subscribedShardStateIndex,
                                               StreamShardHandle subscribedShard,
                                               SequenceNumber lastSequenceNum,
                                               MetricGroup metricGroup,
                                               KinesisDeserializationSchema<T> shardDeserializer)
                                        throws InterruptedException

Create a new shard consumer. Override this method to customize shard consumer behavior in subclasses.

Parameters:

subscribedShardStateIndex - the state index of the shard this consumer is subscribed to

subscribedShard - the shard this consumer is subscribed to

lastSequenceNum - the sequence number in the shard to start consuming

metricGroup - the metric group to report metrics to

Returns:

shard consumer

Throws:

InterruptedException

createRecordPublisherFactory

protected RecordPublisherFactory createRecordPublisherFactory()

createRecordPublisher

protected RecordPublisher createRecordPublisher(SequenceNumber sequenceNumber,
                                                Properties configProps,
                                                MetricGroup metricGroup,
                                                StreamShardHandle subscribedShard)
                                         throws InterruptedException

Throws:

InterruptedException

runFetcher

public void runFetcher()
                throws Exception

Starts the fetcher. After starting the fetcher, it can only be stopped by calling shutdownFetcher().

Throws:

Exception - the first error or exception thrown by the fetcher or any of the threads created by the fetcher.

snapshotState

public HashMap<StreamShardMetadata,SequenceNumber> snapshotState()

Creates a snapshot of the current last processed sequence numbers of each subscribed shard.

Returns:

state snapshot

shutdownFetcher

public void shutdownFetcher()

Starts shutting down the fetcher. Must be called to allow runFetcher() to complete. Once called, the shutdown procedure will be executed and all shard consuming threads will be interrupted.

closeRecordPublisherFactory

@VisibleForTesting
protected void closeRecordPublisherFactory()

Closes recordRecordPublisherFactory. Allows test to override this to simulate exception for shutdown logic.

deregisterStreamConsumer

@VisibleForTesting
protected void deregisterStreamConsumer()

Deregisters stream consumers. Allows test to override this to simulate exception for shutdown logic.

awaitTermination

public void awaitTermination()
                      throws InterruptedException

After calling shutdownFetcher(), this can be called to await the fetcher shutdown.

Throws:

InterruptedException

stopWithError

protected void stopWithError(Throwable throwable)

Called by created threads to pass on errors. Only the first thrown error is set. Once set, the shutdown process will be executed and all shard consuming threads will be interrupted.

advanceLastDiscoveredShardOfStream

public void advanceLastDiscoveredShardOfStream(String stream,
                                               String shardId)

Updates the last discovered shard of a subscribed stream; only updates if the update is valid.

shouldAdvanceLastDiscoveredShardId

protected boolean shouldAdvanceLastDiscoveredShardId(String shardId,
                                                     String lastSeenShardIdOfStream)

Given lastSeenShardId, check if last discovered shardId should be advanced.

discoverNewShardsToSubscribe

public List<StreamShardHandle> discoverNewShardsToSubscribe()
                                                     throws InterruptedException

A utility function that does the following:

1. Find new shards for each stream that we haven't seen before 2. For each new shard, determine whether this consumer subtask should subscribe to them; if yes, it is added to the returned list of shards 3. Update the subscribedStreamsToLastDiscoveredShardIds state so that we won't get shards that we have already seen before the next time this function is called

Throws:

InterruptedException

getConsumerConfiguration

protected Properties getConsumerConfiguration()

emitRecordAndUpdateState

protected void emitRecordAndUpdateState(T record,
                                        long recordTimestamp,
                                        int shardStateIndex,
                                        SequenceNumber lastSequenceNumber)

Prepare a record and hand it over to the RecordEmitter, which may collect it asynchronously. This method is called by ShardConsumers.

Parameters:

record - the record to collect

recordTimestamp - timestamp to attach to the collected record

shardStateIndex - index of the shard to update in subscribedShardsState; this index should be the returned value from registerNewSubscribedShardState(KinesisStreamShardState), called when the shard state was registered.

lastSequenceNumber - the last sequence number value to update

updateState

protected final void updateState(int shardStateIndex,
                                 SequenceNumber lastSequenceNumber)

Update the shard to last processed sequence number state. This method is called by ShardConsumers.

Parameters:

shardStateIndex - index of the shard to update in subscribedShardsState; this index should be the returned value from registerNewSubscribedShardState(KinesisStreamShardState), called when the shard state was registered.

lastSequenceNumber - the last sequence number value to update

registerNewSubscribedShardState

public int registerNewSubscribedShardState(KinesisStreamShardState newSubscribedShardState)

Register a new subscribed shard state.

Parameters:

newSubscribedShardState - the new shard state that this fetcher is to be subscribed to

getCurrentTimeMillis

@VisibleForTesting
protected long getCurrentTimeMillis()

Return the current system time. Allow tests to override this to simulate progress for watermark logic.

Returns:

current processing time

emitWatermark

@VisibleForTesting
protected void emitWatermark()

Called periodically to emit a watermark. Checks all shards for the current event time watermark, and possibly emits the next watermark.

Shards that have not received an update for a certain interval are considered inactive so as to not hold back the watermark indefinitely. When all shards are inactive, the subtask will be marked as temporarily idle to not block downstream operators.

isThisSubtaskShouldSubscribeTo

public static boolean isThisSubtaskShouldSubscribeTo(int shardHash,
                                                     int totalNumberOfConsumerSubtasks,
                                                     int indexOfThisConsumerSubtask)

Utility function to determine whether a shard should be subscribed by this consumer subtask.

Parameters:

shardHash - hash code for the shard

totalNumberOfConsumerSubtasks - total number of consumer subtasks

indexOfThisConsumerSubtask - index of this consumer subtask

createShardConsumersThreadPool

@VisibleForTesting
protected ExecutorService createShardConsumersThreadPool(String subtaskName)

getSubscribedShardsState

@VisibleForTesting
public List<KinesisStreamShardState> getSubscribedShardsState()

createInitialSubscribedStreamsToLastDiscoveredShardsState

protected static HashMap<String,String> createInitialSubscribedStreamsToLastDiscoveredShardsState(List<String> streams)

Utility function to create an initial map of the last discovered shard id of each subscribed stream, set to null; This is called in the constructor; correct values will be set later on by calling advanceLastDiscoveredShardOfStream().

Parameters:

streams - the list of subscribed streams

Returns:

the initial map for subscribedStreamsToLastDiscoveredShardIds

convertToStreamShardMetadata

public static StreamShardMetadata convertToStreamShardMetadata(StreamShardHandle streamShardHandle)

Utility function to convert StreamShardHandle into StreamShardMetadata.

Parameters:

streamShardHandle - the StreamShardHandle to be converted

Returns:

a StreamShardMetadata object

convertToStreamShardHandle

public static StreamShardHandle convertToStreamShardHandle(StreamShardMetadata streamShardMetadata)

Utility function to convert StreamShardMetadata into StreamShardHandle.

Parameters:

streamShardMetadata - the StreamShardMetadata to be converted

Returns:

a StreamShardHandle object