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, Checkpointed<SerializedCheckpointData[]>, SourceFunction<Type>

Direct Known Subclasses:

MultipleIdsMessageAcknowledgingSourceBase

@PublicEvolving
public abstract class MessageAcknowledgingSourceBase<Type,UId>
extends RichSourceFunction<Type>
implements Checkpointed<SerializedCheckpointData[]>, 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()) {
             ctx.collect(msg.getMessageData());
             addId(msg.getMessageId());
         }
     }
 }
 

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 int	numCheckpointsToKeep

Constructor Summary

Constructors

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

Method Summary

Modifier and Type	Method and Description
protected abstract void	acknowledgeIDs(long checkpointId, List<UId> UIds) This method must be implemented to acknowledge the given set of IDs back to the message queue.
protected boolean	addId(UId uid) Adds an ID to be stored with the current checkpoint.
void	close() Tear-down method for the user code.
void	notifyCheckpointComplete(long checkpointId) This method is called as a notification once a distributed checkpoint has been completed.
void	open(Configuration parameters) Initialization method for the function.
void	restoreState(SerializedCheckpointData[] state) Restores the state of the function or operator to that of a previous checkpoint.
SerializedCheckpointData[]	snapshotState(long checkpointId, long checkpointTimestamp) Gets the current state of the function of operator.

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

getIterationRuntimeContext, getRuntimeContext, 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

Field Detail

numCheckpointsToKeep

protected int numCheckpointsToKeep

Constructor Detail

MessageAcknowledgingSourceBase

protected MessageAcknowledgingSourceBase(Class<UId> idClass)

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)

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

open

public void open(Configuration parameters)
          throws Exception

Description copied from interface: RichFunction

Initialization method for the function. It is called before the actual working methods (like map or join) and thus suitable for one time setup work. For functions that are part of an iteration, this method will be invoked at the beginning of each iteration superstep.

The configuration object passed to the function can be used for configuration and initialization. The configuration contains all parameters that were configured on the function in the program composition.

 public class MyMapper extends FilterFunction<String> {
 
     private String searchString;
     
     public void open(Configuration parameters) {
         this.searchString = parameters.getString("foo");
     }
     
     public boolean filter(String value) {
         return value.equals(searchString);
     }
 }
 

By default, this method does nothing.

Specified by:

open in interface RichFunction

Overrides:

open in class AbstractRichFunction

Parameters:

parameters - The configuration containing the parameters attached to the contract.

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.

See Also:

Configuration

close

public void close()
           throws Exception

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,
                                       List<UId> UIds)

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)

Adds an ID to be stored with the current checkpoint.

Parameters:

uid - The ID to add.

Returns:

True if the id has not been processed previously.

snapshotState

public SerializedCheckpointData[] snapshotState(long checkpointId,
                                                long checkpointTimestamp)
                                         throws Exception

Description copied from interface: Checkpointed

Gets the current state of the function of operator. The state must reflect the result of all prior invocations to this function.

Specified by:

snapshotState in interface Checkpointed<SerializedCheckpointData[]>

Parameters:

checkpointId - The ID of the checkpoint.

checkpointTimestamp - The timestamp of the checkpoint, as derived by System.currentTimeMillis() on the JobManager.

Returns:

A snapshot of the operator state.

Throws:

Exception - Thrown if the creation of the state object failed. This causes the checkpoint to fail. The system may decide to fail the operation (and trigger recovery), or to discard this checkpoint attempt and to continue running and to try again with the next checkpoint attempt.

restoreState

public void restoreState(SerializedCheckpointData[] state)
                  throws Exception

Description copied from interface: Checkpointed

Restores the state of the function or operator to that of a previous checkpoint. This method is invoked when a function is executed as part of a recovery run. Note that restoreState() is called before open().

Specified by:

restoreState in interface Checkpointed<SerializedCheckpointData[]>

Parameters:

state - The state to be restored.

Throws:

Exception

notifyCheckpointComplete

public void notifyCheckpointComplete(long checkpointId)
                              throws Exception

Description copied from interface: CheckpointListener

This method is called as a notification once a distributed checkpoint has been completed. Note that any exception during this method will not cause the checkpoint to fail any more.

Specified by:

notifyCheckpointComplete in interface CheckpointListener

Parameters:

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

Throws:

Exception