Flink - StreamTask With Mailbox

A couple of days ago, I saw a StreamTask refactoring proposal on Flink’s development mailing list. After studying it today, I was struck by the simplicity of the developers’ approach. Read the proposal here.

Backgroud

Anyone who has read Flink’s source has seen checkpointLock, which isolates state operations across threads. But using an object lock means the main thread must pass that object to every thread that needs to access state. Before long, the code fills up with synchronized(lock) blocks, making development, debugging, and reading the source extremely inconvenient. The checkpoint lock is mainly used in three places:

• Event Processing: During initialization, operators use the lock to isolate TimerService callbacks; during element processing, they use it to prevent asynchronous snapshot threads from interfering with state. In short, the checkpoint lock appears in many places.
• Checkpoint: Unsurprisingly, performCheckpoint uses the lock.
• Processing Time Timers: Callbacks triggered by this thread often manipulate state, so they must acquire the lock too.

Without a clear understanding of these components and their relationships, checkpointLock is a black box to most people, making failures difficult to diagnose.

Refactor

The community proposes replacing checkpointLock with a mailbox and a single-threaded execution model. The mailbox becomes StreamTask’s source of work, replacing StreamTask#run() in most cases:

1
BlockingQueue<Runnable> mailbox = ...
2

3
void runMailboxProcessing() {
4
    //TODO: can become a cancel-event through mailbox eventually
5
    Runnable letter;
6
    while (isRunning()) {
7
        while ((letter = mailbox.poll()) != null) {
8
            letter.run();
9
        }
10

11
        defaultAction();
12
    }
13
}
14

15
void defaultAction() {
16
    // e.g. event-processing from an input
17
}

For the three cases above, asynchronous checkpoints and processing-time timers turn the logic previously protected by checkpointLock into Runnables and put them in the mailbox. Concurrency becomes a mailbox-based single-threaded model, making the entire StreamTask lighter.

The checkpoint lock interface is also exposed to user-defined SourceFunctions, so the proposal includes backward-compatibility measures that I will not cover here.

Thoughts

CheckpointLock is frequently needed when writing custom sources. I vaguely remember being caught out by it while writing a unit test for a patch… Reading this proposal was therefore enlightening. Although I had not thought deeply about the problem before, it helped me form a clear understanding of the tangled locking mechanisms. The idea behind the change is simple, but the way it simplifies the problem is worth learning from. It solves the immediate issue and simplifies the overall structure, making future performance optimization directions clearer and benefiting later work.