[flink] Bypass operator should never block checkpoint

[JingsongLi]

Purpose

The previous AppendBypassCoordinateOperator spent a lot of time generating compact tasks when there were many files, leading to significant blocking of normal threads and even resulting in checkpoint delays.

Modify it:

1. Scanning in a separate thread.
2. Make sure there is no message in mail box, and send compact tasks to downstream.

Tests

API and Format

Documentation

[wwj6591812]

Should we first check if tasks.isEmpty()? If it's not empty, then compactTasks.addAll(tasks);

[JingsongLi]

Why?

[wwj6591812]

Because if tasks.isEmpty(), then there is no need to execute compactTasks.addAll (tasks);

[JingsongLi]

So what is the benefit?

[wwj6591812]

private

[wwj6591812]

Hi, thx for this @JingsongLi

I have a doubt that you say generating compact tasks blocking normal threads and even resulting in checkpoint delays.
But In my test, our checkpoint is normal. But the problem is that the first operator parallelism is 1024, and the Compact Coordinator parallelism is 1. This leads to the phenomenon of back pressure.

So, should we support both this new topology and old union topology simultaneously?

[JingsongLi]

@wwj6591812

What is your point? You mean you use new code and the job still back pressure?
If not, please valid this first.

[wwj6591812]

@wwj6591812

What is your point? You mean you use new code and the job still back pressure? If not, please valid this first.

ok，after merge，I will test soon.

[leaves12138]

+1