Motivation

Currently, there will be a global committer node in Flink Paimon job which is used to ensure the consistency of written data in Paimon. This committer node will connect all the tasks in Flink job together, and all the tasks are within one region. As a result, when any task fails, it will trigger a global failover of the Flink job. We use HDFS as the remote storage, and we often encounter situations where the global failover of jobs is triggered due to write timeouts or errors when writing to HDFS, which are quite a few stability issues.

In order to improve the stability of streaming write to Paimon in Flink, we would like to perform the committer logic within the Flink JobManager (JM), so that when Flink writes to an append unaware bucket table, it can support region failover.

Architecture

Overall Processing

The specific interaction logic between the PaimonWriter and the Committer node is shown in the above figure. There are two main logics for the Committer node:

1. Snapshot state stage
   1. According to the Checkpoint, summarize the file information sent by the upstream Writer
   2. Save the file information in the state. When the two-phase commit fails, the Snapshot information can be restored from the state and the Checkpoint Complete operation can be re-executed to ensure the successful generation of the Snapshot.
2. Checkpoint Complete stage: The Commit operation of the two-phase commit, calling the Commit step of PaimonTable.

We'd like to implement a PaimonWriterCoordinator for Flink, inheriting the OperatorCoordinator interface. The interaction logic between the writer node of Paimon and the PaimonWriterCoordinator component is shown in the following figure.

1. Snapshot State Stage
   1. When each Writer Task executes a checkpoint, it is necessary to ensure that the Paimon File Info is successfully sent to the Paimon Writer Coordinator.
   2. After the last Writer Task executes checkpoint, it is necessary to ensure that the Paimon Writer Coordinator collects all the File Info and flushes the state in the task->checkpoint->file info format to HDFS.
   3. Through the above two steps, ensure the consistency of the File Info of the Writer Task in the state storage of the Paimon Writer Coordinator: either both are saved successfully; or trigger failover and replay from the previous checkpoint.
2. Checkpoint Complete Stage, Paimon Writer Coordinator performs the following actions
   1. Read the File Info from the state based on the Checkpoint Id and create the Meta Info.
   2. Complete the Table Committer operation and Flush the new state to HDFS

Failover Processing

During the process of Writer Task sending File Info to PaimonWriterCoordinator and complete the checkpoint, various abnormal situations such as network issues, task failures, or PaimonWriterCoordinator failover may occur. 

1) Task failed to send File Info to PaimonWriterCoordinator

Writer Task sending File Info to PaimonWriterCoordinator is a synchronous operation, which requires ensuring that PaimonWriterCoordinator receives the message. If PaimonWriterCoordinator does not receive the File Info of the specified Task, the Task will sense the error and actively trigger failover.

2) The PaimonWriterCoordinator receives the File Info of the failed task

Due to timing issues, the PaimonWriterCoordinator received multiple File Info messages for the specified Task and Checkpoint, and the File Info before and after the failed task was out of order. The OperatorCoordinator interface has the executionAttemptFailed and executionAttemptReady methods, after receiving the specified subtask and attemptNumber, PaimonWriterCoordinator can be saved in memory and filter the received File Info.

3) PaimonWriterCoordinator failover

PaimonWriterCoordinator fails over, and the uncompleted checkpoint will be retried after recovery. Since the last checkpoint may not be completed, there will be dirty data in the state. Therefore, when performing the checkpoint complete operation after recovery, the correct state data can be filtered according to the checkpoint id, and at the same time, the dirty data can be overwritten by the new data reported by the task and checkpoint.

Proposed Changes

1) Introduce a new option: sink.committer-coordinator-operator.enabled in FlinkConnectorOptions and the default value is false. When users set it true, it will use PaimonWriterCoordinator instead of committer operator in flink job.

2) Add new paimon writer coordinator class as follows

public class PaimonWriterCoordinator implements OperatorCoordinator {
    /**
     * Save subtask index and attempt number in coordinator to filter file info from writer task.
     */
    @Override
    public void executionAttemptReady(int subtask, int attemptNumber, SubtaskGateway gateway);

    /**
     * Failed specify subtask in coordinator.
     */
    @Override 
    public void executionAttemptFailed(int subtask, int attemptNumber, @Nullable Throwable throwable);
 
    /**
     * Receive file info from writer, check whether all tasks have been reported and then flush state to HDFS.
     */
    @Override 
    public void handleEventFromOperator(int subtask, int attemptNumber, OperatorEvent event)
            throws Exception;

    /**
     * Notify coordinator to perform table commit for given checkpoint id.
     */
    @Override
    public void notifyCheckpointComplete(long checkpointId);
}

Compatibility, Deprecation, and Migration Plan

no