THIS IS A TEST INSTANCE. ALL YOUR CHANGES WILL BE LOST!!!!
...
- We would like to introduce a new event to notify all the user records are finished, and call finish() immediately after the new event is aligned for all the input channels. Since it shares the same semantics with the event EndOfUserRecordsEvent we introduced to solve the unaligned checkpoint issues in previous sections, we could reuse this event.
- We would adjust the process of savepoint to be first finish the task and then trigger a savepoint so that they could be unified with the final checkpoint process and avoid confused records after savepoint.
The detail life cycle of for the StreamTask source stream tasks and the operators would become
...
Event | Stream Task Status | Operator Status | Final Checkpoint | Stop with Savepoint with Drain | Stop with Savepoint |
---|---|---|---|---|---|
RUNNING | RUNNING | - | - | - | |
No more records or Received Savepoint Trigger | - | - | - | ||
- | finish task (cancel source thread for legacy source and suspend the mailbox for the new source) | finish task (cancel source thread for legacy source and suspend the mailbox for the new source) | |||
Advanced To MAX_WATERMARK and trigger all the event timers | Advanced To MAX_WATERMARK and trigger all the event timers | - | |||
Emit MAX_WATERMARK | Emit MAX_WATERMARK | - | |||
WAITING_FOR_FINAL_CP | FINISHED | call operator.endInput() & operator.finish() | call operator.endInput() & operator.finish() | - | |
Emit EndOfUserRecordsEvent[finished = true] | Emit EndOfUserRecordsEvent[finished = true] | Emit EndOfUserRecordsEvent[finished = false] | |||
when checkpoint triggered, emit Checkpoint Barrier | Emit Checkpoint Barrier | Emit Checkpoint Barrier | |||
Wait for Checkpoint / Savepoint Completed | Wait for Checkpoint / Savepoint Completed | Wait for Checkpoint / Savepoint Completed | |||
Checkpoint Completed | Wait for downstream tasks acknowledge EndOfUserRecordsEvent | Wait for downstream tasks acknowledge EndOfUserRecordsEvent | Wait for downstream tasks acknowledge EndOfUserRecordsEvent | ||
Downstream Tasks acknowledge EndOfUserRecordsEvent | CLOSED | CLOSED | - | - | |
Call operator.close() | Call operator.close() | Call operator.close() | |||
Emit EndOfPartitionEvent | Emit EndOfPartitionEvent | Emit EndOfPartitionEvent |
Similarly, the status for the non-source tasks would become
Event | Stream Task Status | Operator Status | Final Checkpoint | Stop with Savepoint with Drain | Stop with Savepoint |
---|---|---|---|---|---|
RUNNING | RUNNING | - | - | - | |
- | - | - | |||
- | - | - | |||
Aligned on MAX_WATERMARK | Advanced To MAX_WATERMARK and trigger all the event timers | Advanced To MAX_WATERMARK and trigger all the event timers | - | ||
Emit MAX_WATERMARK | Emit MAX_WATERMARK | - | |||
Aligned On EndOfUserRecordsEvent | WAITING_FOR_FINAL_CP | FINISHED | call operator.endInput() & operator.finish() | call operator.endInput() & operator.finish() | - |
Emit EndOfUserRecordsEvent[finished = true] | Emit EndOfUserRecordsEvent[finished = true] | Emit EndOfUserRecordsEvent[finished = false] | |||
Aligned on Checkpoint Barrier | Emit CheckpointBarrier | Emit CheckpointBarrier | Emit CheckpointBarrier | ||
Wait for Checkpoint / Savepoint Completed | Wait for Checkpoint / Savepoint Completed | Wait for Checkpoint / Savepoint Completed | |||
Checkpoint Completed | Wait for downstream tasks acknowledge EndOfUserRecordsEvent | Wait for downstream tasks acknowledge EndOfUserRecordsEvent | Wait for downstream tasks acknowledge EndOfUserRecordsEvent | ||
Downstream Tasks acknowledge EndOfUserRecordsEvent | CLOSED | CLOSED | - | - | |
Call operator.close() | Call operator.close() | Call operator.close() | |||
Emit EndOfPartitionEvent | Emit EndOfPartitionEvent | Emit EndOfPartitionEvent |
Rejected Alternatives
CheckpointCoordinator Maintains the Finished Status of Tasks Separately
...