org.apache.flink.streaming.api.functions.sink

Class TwoPhaseCommitSinkFunction<IN,TXN,CONTEXT>

java.lang.Object

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

org.apache.flink.streaming.api.functions.sink.RichSinkFunction<IN>

org.apache.flink.streaming.api.functions.sink.TwoPhaseCommitSinkFunction<IN,TXN,CONTEXT>

Type Parameters:

IN - Input type for SinkFunction.

TXN - Transaction to store all of the information required to handle a transaction.

CONTEXT - Context that will be shared across all invocations for the given TwoPhaseCommitSinkFunction instance. Context is created once

All Implemented Interfaces:

Serializable, Function, RichFunction, CheckpointListener, CheckpointedFunction, SinkFunction<IN>

Direct Known Subclasses:

FlinkKafkaProducer, FlinkKafkaProducer011

@PublicEvolving
public abstract class TwoPhaseCommitSinkFunction<IN,TXN,CONTEXT>
extends RichSinkFunction<IN>
implements CheckpointedFunction, CheckpointListener

This is a recommended base class for all of the SinkFunction that intend to implement exactly-once semantic. It does that by implementing two phase commit algorithm on top of the CheckpointedFunction and CheckpointListener. User should provide custom TXN (transaction handle) and implement abstract methods handling this transaction handle.

See Also:

Serialized Form

Nested Class Summary

Nested Classes

Modifier and Type	Class and Description
static class	TwoPhaseCommitSinkFunction.State<TXN,CONTEXT> State POJO class coupling pendingTransaction, context and pendingCommitTransactions.
static class	TwoPhaseCommitSinkFunction.StateSerializer<TXN,CONTEXT> Custom TypeSerializer for the sink state.
static class	TwoPhaseCommitSinkFunction.StateSerializerConfigSnapshot<TXN,CONTEXT> Deprecated. this snapshot class is no longer in use, and is maintained only for backwards compatibility purposes. It is fully replaced by TwoPhaseCommitSinkFunction.StateSerializerSnapshot.
static class	TwoPhaseCommitSinkFunction.StateSerializerSnapshot<TXN,CONTEXT> Snapshot for the TwoPhaseCommitSinkFunction.StateSerializer.
static class	TwoPhaseCommitSinkFunction.TransactionHolder<TXN> Adds metadata (currently only the start time of the transaction) to the transaction object.

Nested classes/interfaces inherited from interface org.apache.flink.streaming.api.functions.sink.SinkFunction

SinkFunction.Context<T>

Field Summary

Fields

Modifier and Type	Field and Description
protected LinkedHashMap<Long,TwoPhaseCommitSinkFunction.TransactionHolder<TXN>>	pendingCommitTransactions
protected ListState<TwoPhaseCommitSinkFunction.State<TXN,CONTEXT>>	state
protected Optional<CONTEXT>	userContext

Constructor Summary

Constructors

Constructor and Description
TwoPhaseCommitSinkFunction(TypeSerializer<TXN> transactionSerializer, TypeSerializer<CONTEXT> contextSerializer) Use default ListStateDescriptor for internal state serialization.

Method Summary

Modifier and Type	Method and Description
protected abstract void	abort(TXN transaction) Abort a transaction.
protected abstract TXN	beginTransaction() Method that starts a new transaction.
void	close() Tear-down method for the user code.
protected abstract void	commit(TXN transaction) Commit a pre-committed transaction.
protected TXN	currentTransaction()
protected TwoPhaseCommitSinkFunction<IN,TXN,CONTEXT>	enableTransactionTimeoutWarnings(double warningRatio) Enables logging of warnings if a transaction's elapsed time reaches a specified ratio of the transactionTimeout.
protected void	finishRecoveringContext()
protected Optional<CONTEXT>	getUserContext()
protected TwoPhaseCommitSinkFunction<IN,TXN,CONTEXT>	ignoreFailuresAfterTransactionTimeout() If called, the sink will only log but not propagate exceptions thrown in recoverAndCommit(Object) if the transaction is older than a specified transaction timeout.
void	initializeState(FunctionInitializationContext context) This method is called when the parallel function instance is created during distributed execution.
protected Optional<CONTEXT>	initializeUserContext()
void	invoke(IN value) This should not be implemented by subclasses.
void	invoke(IN value, SinkFunction.Context context) Writes the given value to the sink.
protected abstract void	invoke(TXN transaction, IN value, SinkFunction.Context context) Write value within a transaction.
void	notifyCheckpointComplete(long checkpointId) This method is called as a notification once a distributed checkpoint has been completed.
protected java.util.stream.Stream<Map.Entry<Long,TXN>>	pendingTransactions()
protected abstract void	preCommit(TXN transaction) Pre commit previously created transaction.
protected void	recoverAndAbort(TXN transaction) Abort a transaction that was rejected by a coordinator after a failure.
protected void	recoverAndCommit(TXN transaction) Invoked on recovered transactions after a failure.
protected TwoPhaseCommitSinkFunction<IN,TXN,CONTEXT>	setTransactionTimeout(long transactionTimeout) Sets the transaction timeout.
void	snapshotState(FunctionSnapshotContext context) 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

Field Detail

pendingCommitTransactions

protected final LinkedHashMap<Long,TwoPhaseCommitSinkFunction.TransactionHolder<TXN>> pendingCommitTransactions

userContext

protected transient Optional<CONTEXT> userContext

state

protected transient ListState<TwoPhaseCommitSinkFunction.State<TXN,CONTEXT>> state

Constructor Detail

TwoPhaseCommitSinkFunction

public TwoPhaseCommitSinkFunction(TypeSerializer<TXN> transactionSerializer,
                                  TypeSerializer<CONTEXT> contextSerializer)

Use default ListStateDescriptor for internal state serialization. Helpful utilities for using this constructor are TypeInformation.of(Class), TypeHint and TypeInformation.of(TypeHint). Example:

 
 TwoPhaseCommitSinkFunction(TypeInformation.of(new TypeHint<State<TXN, CONTEXT>>() {}));
 
 

Parameters:

transactionSerializer - TypeSerializer for the transaction type of this sink

contextSerializer - TypeSerializer for the context type of this sink

Method Detail

initializeUserContext

protected Optional<CONTEXT> initializeUserContext()

getUserContext

protected Optional<CONTEXT> getUserContext()

currentTransaction

@Nullable
protected TXN currentTransaction()

pendingTransactions

@Nonnull
protected java.util.stream.Stream<Map.Entry<Long,TXN>> pendingTransactions()

invoke

protected abstract void invoke(TXN transaction,
                               IN value,
                               SinkFunction.Context context)
                        throws Exception

Write value within a transaction.

Throws:

Exception

beginTransaction

protected abstract TXN beginTransaction()
                                 throws Exception

Method that starts a new transaction.

Returns:

newly created transaction.

Throws:

Exception

preCommit

protected abstract void preCommit(TXN transaction)
                           throws Exception

Pre commit previously created transaction. Pre commit must make all of the necessary steps to prepare the transaction for a commit that might happen in the future. After this point the transaction might still be aborted, but underlying implementation must ensure that commit calls on already pre committed transactions will always succeed.

Usually implementation involves flushing the data.

Throws:

Exception

commit

protected abstract void commit(TXN transaction)

Commit a pre-committed transaction. If this method fail, Flink application will be restarted and recoverAndCommit(Object) will be called again for the same transaction.

recoverAndCommit

protected void recoverAndCommit(TXN transaction)

Invoked on recovered transactions after a failure. User implementation must ensure that this call will eventually succeed. If it fails, Flink application will be restarted and it will be invoked again. If it does not succeed eventually, a data loss will occur. Transactions will be recovered in an order in which they were created.

abort

protected abstract void abort(TXN transaction)

Abort a transaction.

recoverAndAbort

protected void recoverAndAbort(TXN transaction)

Abort a transaction that was rejected by a coordinator after a failure.

finishRecoveringContext

protected void finishRecoveringContext()

invoke

public final void invoke(IN value)
                  throws Exception

This should not be implemented by subclasses.

Specified by:

invoke in interface SinkFunction<IN>

Throws:

Exception

invoke

public final void invoke(IN value,
                         SinkFunction.Context context)
                  throws Exception

Description copied from interface: SinkFunction

Writes the given value to the sink. This function is called for every record.

You have to override this method when implementing a SinkFunction, this is a default method for backward compatibility with the old-style method only.

Specified by:

invoke in interface SinkFunction<IN>

Parameters:

value - The input record.

context - Additional context about the input record.

Throws:

Exception - This method may throw exceptions. Throwing an exception will cause the operation to fail and may trigger recovery.

notifyCheckpointComplete

public final 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

snapshotState

public void snapshotState(FunctionSnapshotContext context)
                   throws Exception

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

initializeState

public void initializeState(FunctionInitializationContext context)
                     throws Exception

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

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.

setTransactionTimeout

protected TwoPhaseCommitSinkFunction<IN,TXN,CONTEXT> setTransactionTimeout(long transactionTimeout)

Sets the transaction timeout. Setting only the transaction timeout has no effect in itself.

Parameters:

transactionTimeout - The transaction timeout in ms.

See Also:

ignoreFailuresAfterTransactionTimeout(), enableTransactionTimeoutWarnings(double)

ignoreFailuresAfterTransactionTimeout

protected TwoPhaseCommitSinkFunction<IN,TXN,CONTEXT> ignoreFailuresAfterTransactionTimeout()

If called, the sink will only log but not propagate exceptions thrown in recoverAndCommit(Object) if the transaction is older than a specified transaction timeout. The start time of an transaction is determined by System.currentTimeMillis(). By default, failures are propagated.

enableTransactionTimeoutWarnings

protected TwoPhaseCommitSinkFunction<IN,TXN,CONTEXT> enableTransactionTimeoutWarnings(double warningRatio)

Enables logging of warnings if a transaction's elapsed time reaches a specified ratio of the transactionTimeout. If warningRatio is 0, a warning will be always logged when committing the transaction.

Parameters:

warningRatio - A value in the range [0,1].

Returns: