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

Class 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, org.apache.flink.api.common.functions.Function, org.apache.flink.api.common.functions.RichFunction, org.apache.flink.api.common.state.CheckpointListener, CheckpointedFunction, SinkFunction<IN>
    @PublicEvolving
public abstract class TwoPhaseCommitSinkFunction<IN,​TXN,​CONTEXT>
extends RichSinkFunction<IN>
implements CheckpointedFunction, org.apache.flink.api.common.state.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

    • Constructor Detail

      • TwoPhaseCommitSinkFunction

        public TwoPhaseCommitSinkFunction​(org.apache.flink.api.common.typeutils.TypeSerializer<TXN> transactionSerializer,
                                  org.apache.flink.api.common.typeutils.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()

      • currentTransaction

        @Nullable
protected TXN currentTransaction()

      • 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​(Collection<TXN> handledTransactions)
        Callback for subclasses which is called after restoring (each) user context.
        Parameters:
        handledTransactions - transactions which were already committed or aborted and do not need further handling

      • finishProcessing

        protected void finishProcessing​(@Nullable
                                TXN transaction)
        This method is called at the end of data processing.

        The method is expected to flush all remaining buffered data. Exceptions will cause the pipeline to be recognized as failed, because the last data items are not processed properly. You may use this method to flush remaining buffered elements in the state into the current transaction which will be committed in the last checkpoint.

      • 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.

      • finish

        public final void finish()
                  throws Exception
        Description copied from interface: SinkFunction
        This method is called at the end of data processing.

        The method is expected to flush all remaining buffered data. Exceptions will cause the pipeline to be recognized as failed, because the last data items are not processed properly. You may use this method to flush remaining buffered elements in the state into transactions which you can commit in the last checkpoint.

        NOTE:This method does not need to close any resources. You should release external resources in the AbstractRichFunction.close() method.

        Specified by:
        finish in interface SinkFunction<IN>
        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
        Specified by:
        notifyCheckpointComplete in interface org.apache.flink.api.common.state.CheckpointListener
        Throws:
        Exception

      • notifyCheckpointAborted

        public void notifyCheckpointAborted​(long checkpointId)
        Specified by:
        notifyCheckpointAborted in interface org.apache.flink.api.common.state.CheckpointListener

      • snapshotState

        public void snapshotState​(org.apache.flink.runtime.state.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 - Thrown, if state could not be created ot restored.

      • initializeState

        public void initializeState​(org.apache.flink.runtime.state.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 - Thrown, if state could not be created ot restored.

      • close

        public void close()
           throws Exception
        Specified by:
        close in interface org.apache.flink.api.common.functions.RichFunction
        Overrides:
        close in class org.apache.flink.api.common.functions.AbstractRichFunction
        Throws:
        Exception

      • 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: