org.apache.flink.streaming.api.functions.source

Class MessageAcknowledgingSourceBase<Type,UId>

java.lang.Object

org.apache.flink.api.common.functions.AbstractRichFunction

org.apache.flink.streaming.api.functions.source.RichSourceFunction<Type>

org.apache.flink.streaming.api.functions.source.MessageAcknowledgingSourceBase<Type,UId>

Type Parameters:

Type - The type of the messages created by the source.

UId - The type of unique IDs which may be used to acknowledge elements.

All Implemented Interfaces:

Serializable, Function, RichFunction, CheckpointListener, CheckpointedFunction, SourceFunction<Type>

Direct Known Subclasses:

MultipleIdsMessageAcknowledgingSourceBase

Deprecated.

This class is based on the SourceFunction API, which is due to be removed. Use the new Source API instead.

@Deprecated
 @PublicEvolving
public abstract class MessageAcknowledgingSourceBase<Type,UId>
extends RichSourceFunction<Type>
implements CheckpointedFunction, CheckpointListener

Abstract base class for data sources that receive elements from a message queue and acknowledge them back by IDs.

The mechanism for this source assumes that messages are identified by a unique ID. When messages are taken from the message queue, the message must not be dropped immediately, but must be retained until acknowledged. Messages that are not acknowledged within a certain time interval will be served again (to a different connection, established by the recovered source).

Note that this source can give no guarantees about message order in the case of failures, because messages that were retrieved but not yet acknowledged will be returned later again, after a set of messages that was not retrieved before the failure.

Internally, this source gathers the IDs of elements it emits. Per checkpoint, the IDs are stored and acknowledged when the checkpoint is complete. That way, no message is acknowledged unless it is certain that it has been successfully processed throughout the topology and the updates to any state caused by that message are persistent.

All messages that are emitted and successfully processed by the streaming program will eventually be acknowledged. In corner cases, the source may receive certain IDs multiple times, if a failure occurs while acknowledging. To cope with this situation, an additional Set stores all processed IDs. IDs are only removed after they have been acknowledged.

A typical way to use this base in a source function is by implementing a run() method as follows:

 public void run(SourceContext<Type> ctx) throws Exception {
     while (running) {
         Message msg = queue.retrieve();
         synchronized (ctx.getCheckpointLock()) {
             if (addId(msg.getMessageId())) {
                 ctx.collect(msg.getMessageData());
             }
         }
     }
 }
 

NOTE: This source has a parallelism of 1.

See Also:

Serialized Form

Nested Class Summary

Nested classes/interfaces inherited from interface org.apache.flink.streaming.api.functions.source.SourceFunction

SourceFunction.SourceContext<T>

Field Summary

Fields

Modifier and Type	Field and Description
protected ArrayDeque<Tuple2<Long,Set<UId>>>	pendingCheckpoints Deprecated. The list with IDs from checkpoints that were triggered, but not yet completed or notified of completion.

Constructor Summary

Constructors

Modifier	Constructor and Description
protected	MessageAcknowledgingSourceBase(Class<UId> idClass) Deprecated. Creates a new MessageAcknowledgingSourceBase for IDs of the given type.
protected	MessageAcknowledgingSourceBase(TypeInformation<UId> idTypeInfo) Deprecated. Creates a new MessageAcknowledgingSourceBase for IDs of the given type.

Method Summary

Modifier and Type	Method and Description
protected abstract void	acknowledgeIDs(long checkpointId, Set<UId> uIds) Deprecated. This method must be implemented to acknowledge the given set of IDs back to the message queue.
protected boolean	addId(UId uid) Deprecated. Adds an ID to be stored with the current checkpoint.
void	close() Deprecated. Tear-down method for the user code.
void	initializeState(FunctionInitializationContext context) Deprecated. This method is called when the parallel function instance is created during distributed execution.
void	notifyCheckpointAborted(long checkpointId) Deprecated. This method is called as a notification once a distributed checkpoint has been aborted.
void	notifyCheckpointComplete(long checkpointId) Deprecated. Notifies the listener that the checkpoint with the given checkpointId completed and was committed.
void	snapshotState(FunctionSnapshotContext context) Deprecated. This method is called when a snapshot for a checkpoint is requested.

Methods inherited from class org.apache.flink.api.common.functions.AbstractRichFunction

getIterationRuntimeContext, getRuntimeContext, open, setRuntimeContext

Methods inherited from class java.lang.Object

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

Methods inherited from interface org.apache.flink.streaming.api.functions.source.SourceFunction

cancel, run

Methods inherited from interface org.apache.flink.api.common.functions.RichFunction

open

Field Detail

pendingCheckpoints

protected transient ArrayDeque<Tuple2<Long,Set<UId>>> pendingCheckpoints

Deprecated.

The list with IDs from checkpoints that were triggered, but not yet completed or notified of completion.

Constructor Detail

MessageAcknowledgingSourceBase

protected MessageAcknowledgingSourceBase(Class<UId> idClass)

Deprecated.

Creates a new MessageAcknowledgingSourceBase for IDs of the given type.

Parameters:

idClass - The class of the message ID type, used to create a serializer for the message IDs.

MessageAcknowledgingSourceBase

protected MessageAcknowledgingSourceBase(TypeInformation<UId> idTypeInfo)

Deprecated.

Creates a new MessageAcknowledgingSourceBase for IDs of the given type.

Parameters:

idTypeInfo - The type information of the message ID type, used to create a serializer for the message IDs.

Method Detail

initializeState

public void initializeState(FunctionInitializationContext context)
                     throws Exception

Deprecated.

Description copied from interface: CheckpointedFunction

This method is called when the parallel function instance is created during distributed execution. Functions typically set up their state storing data structures in this method.

Specified by:

initializeState in interface CheckpointedFunction

Parameters:

context - the context for initializing the operator

Throws:

Exception - Thrown, if state could not be created ot restored.

close

public void close()
           throws Exception

Deprecated.

Description copied from interface: RichFunction

Tear-down method for the user code. It is called after the last call to the main working methods (e.g. map or join). For functions that are part of an iteration, this method will be invoked after each iteration superstep.

This method can be used for clean up work.

Specified by:

close in interface RichFunction

Overrides:

close in class AbstractRichFunction

Throws:

Exception - Implementations may forward exceptions, which are caught by the runtime. When the runtime catches an exception, it aborts the task and lets the fail-over logic decide whether to retry the task execution.

acknowledgeIDs

protected abstract void acknowledgeIDs(long checkpointId,
                                       Set<UId> uIds)

Deprecated.

This method must be implemented to acknowledge the given set of IDs back to the message queue.

Parameters:

uIds - The list od IDs to acknowledge.

addId

protected boolean addId(UId uid)

Deprecated.

Adds an ID to be stored with the current checkpoint. In order to achieve exactly-once guarantees, implementing classes should only emit records with IDs for which this method return true.

Parameters:

uid - The ID to add.

Returns:

True if the id has not been processed previously.

snapshotState

public void snapshotState(FunctionSnapshotContext context)
                   throws Exception

Deprecated.

Description copied from interface: CheckpointedFunction

This method is called when a snapshot for a checkpoint is requested. This acts as a hook to the function to ensure that all state is exposed by means previously offered through FunctionInitializationContext when the Function was initialized, or offered now by FunctionSnapshotContext itself.

Specified by:

snapshotState in interface CheckpointedFunction

Parameters:

context - the context for drawing a snapshot of the operator

Throws:

Exception - Thrown, if state could not be created ot restored.

notifyCheckpointComplete

public void notifyCheckpointComplete(long checkpointId)
                              throws Exception

Deprecated.

Description copied from interface: CheckpointListener

Notifies the listener that the checkpoint with the given checkpointId completed and was committed.

These notifications are "best effort", meaning they can sometimes be skipped. To behave properly, implementers need to follow the "Checkpoint Subsuming Contract". Please see the class-level JavaDocs for details.

Please note that checkpoints may generally overlap, so you cannot assume that the notifyCheckpointComplete() call is always for the latest prior checkpoint (or snapshot) that was taken on the function/operator implementing this interface. It might be for a checkpoint that was triggered earlier. Implementing the "Checkpoint Subsuming Contract" (see above) properly handles this situation correctly as well.

Please note that throwing exceptions from this method will not cause the completed checkpoint to be revoked. Throwing exceptions will typically cause task/job failure and trigger recovery.

Specified by:

notifyCheckpointComplete in interface CheckpointListener

Parameters:

checkpointId - The ID of the checkpoint that has been completed.

Throws:

Exception - This method can propagate exceptions, which leads to a failure/recovery for the task. Note that this will NOT lead to the checkpoint being revoked.

notifyCheckpointAborted

public void notifyCheckpointAborted(long checkpointId)

Deprecated.

Description copied from interface: CheckpointListener

This method is called as a notification once a distributed checkpoint has been aborted.

Important: The fact that a checkpoint has been aborted does NOT mean that the data and artifacts produced between the previous checkpoint and the aborted checkpoint are to be discarded. The expected behavior is as if this checkpoint was never triggered in the first place, and the next successful checkpoint simply covers a longer time span. See the "Checkpoint Subsuming Contract" in the class-level JavaDocs for details.

These notifications are "best effort", meaning they can sometimes be skipped.

This method is very rarely necessary to implement. The "best effort" guarantee, together with the fact that this method should not result in discarding any data (per the "Checkpoint Subsuming Contract") means it is mainly useful for earlier cleanups of auxiliary resources. One example is to pro-actively clear a local per-checkpoint state cache upon checkpoint failure.

Specified by:

notifyCheckpointAborted in interface CheckpointListener

Parameters:

checkpointId - The ID of the checkpoint that has been aborted.