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

Class MultipleIdsMessageAcknowledgingSourceBase<Type,UId,SessionId>

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>

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

Type Parameters:

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

UId - The type of the unique IDs which are consistent across sessions.

SessionId - The type of the IDs that are used for acknowledging elements (ids valid during session).

All Implemented Interfaces:

Serializable, Function, RichFunction, CheckpointListener, Checkpointed<SerializedCheckpointData[]>, SourceFunction<Type>

Direct Known Subclasses:

RMQSource

@PublicEvolving
public abstract class MultipleIdsMessageAcknowledgingSourceBase<Type,UId,SessionId>
extends MessageAcknowledgingSourceBase<Type,UId>

Abstract base class for data sources that receive elements from a message queue and acknowledge them back by IDs. In contrast to MessageAcknowledgingSourceBase, this source handles two types of ids: 1) Session ids 2) Unique message ids Session ids are used to acknowledge messages in a session. When a checkpoint is restored, unacknowledged messages are redelivered. Duplicates are detected using the unique message ids which are checkpointed.

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 List<SessionId>	sessionIds
protected Deque<Tuple2<Long,List<SessionId>>>	sessionIdsPerSnapshot

Fields inherited from class org.apache.flink.streaming.api.functions.source.MessageAcknowledgingSourceBase

numCheckpointsToKeep

Constructor Summary

Constructors

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

Method Summary

Modifier and Type	Method and Description
protected void	acknowledgeIDs(long checkpointId, List<UId> uniqueIds) Acknowledges the session ids.
protected abstract void	acknowledgeSessionIDs(List<SessionId> sessionIds) Acknowledges the session ids.
void	close() Tear-down method for the user code.
void	open(Configuration parameters) Initialization method for the function.
SerializedCheckpointData[]	snapshotState(long checkpointId, long checkpointTimestamp) Gets the current state of the function of operator.

Methods inherited from class org.apache.flink.streaming.api.functions.source.MessageAcknowledgingSourceBase

addId, notifyCheckpointComplete, restoreState

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

sessionIdsPerSnapshot

protected transient Deque<Tuple2<Long,List<SessionId>>> sessionIdsPerSnapshot

sessionIds

protected transient List<SessionId> sessionIds

Constructor Detail

MultipleIdsMessageAcknowledgingSourceBase

protected MultipleIdsMessageAcknowledgingSourceBase(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.

MultipleIdsMessageAcknowledgingSourceBase

protected MultipleIdsMessageAcknowledgingSourceBase(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 MessageAcknowledgingSourceBase<Type,UId>

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 MessageAcknowledgingSourceBase<Type,UId>

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 final void acknowledgeIDs(long checkpointId,
                                    List<UId> uniqueIds)

Acknowledges the session ids.

Specified by:

acknowledgeIDs in class MessageAcknowledgingSourceBase<Type,UId>

Parameters:

checkpointId - The id of the current checkout to acknowledge ids for.

uniqueIds - The checkpointed unique ids which are ignored here. They only serve as a means of de-duplicating messages when the acknowledgment after a checkpoint fails.

acknowledgeSessionIDs

protected abstract void acknowledgeSessionIDs(List<SessionId> sessionIds)

Acknowledges the session ids.

Parameters:

sessionIds - The message ids for this session.

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[]>

Overrides:

snapshotState in class MessageAcknowledgingSourceBase<Type,UId>

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.