[FLINK-36690][runtime] Fix schema operator hanging under extreme parallelized pressure

[yuxiqian]

This closes FLINK-36690 by fixing schema operator hanging glitch under extreme parallelized pressure.

---

After FLINK-36114, SchemaOperators will ask for schema evolve permission first, before sending FlushEvents to downstream sinks.

However, Flink regards FlushEvents as normal data records and might block it to align checkpoint barriers. It might cause the following deadlock situation:

• SchemaOperator A has obtained schema evolution permission
• SchemaOperator B does not get the permission and hangs
• SchemaOperator A sends a FlushEvent, but after a checkpoint barrier
• SchemaOperator B received a checkpoint barrier after the schema change event (which is blocked)

Now, neither A nor B can post any event records to downstream, and the entire job blocks with the following iconic error message (in TM):

0> Schema Registry is busy now, waiting for next request...
1> Schema Registry is busy now, waiting for next request...
2> Schema Registry is busy now, waiting for next request...
[Repeated lines]

---

This PR changes the schema evolution permission requesting workflow by:

• SchemaOperators emit FlushEvent immediately when they received a SchemaChangeEvent.
• A schema change request could only be permitted when a) SchemaRegistry is IDLE and b) the requesting SchemaOperator has finished data flushing already.

Since FlushEvent might be emitted from multiple SchemaOperators simultaneously, a nonce value that is uniquely bound to a schema change event is added into FlushEvent payload. WAITING_FOR_FLUSH stage is no longer necessary since this state will not block the SchemaRegistry but one single SchemaOperator now.

It should be noted that current schema evolution design implicitly assumes that for each table, it won't be schema evolving in subTask A and emits normal data change events without blocking in subTask B at the same time.

So, after a SchemaOperator successfully triggered a flush event, there can't be any more uncommitted dirty data got written down since 1) any following data from this subTask is still being blocked and 2) other subTask can't carry any data belonging to this tableId (according to our previous guarantee).

[yuxiqian]

Need more eyes on this PR since this PR tweaks schema evolution communication process. cc @leonardBang @ruanhang1993

[yuxiqian]

Based on previous discussions, I've made the following changes:

• Adjusted flush event nonce generating algorithm. Now the higher 32 bits are timestamp and lower 32 bits are Java hash code (like snowflake ID).
• added parallelized schema change cases in MySQL integrated test & route e2e test.

As we're not going to merge it into 3.2.1, I would propose downgrading parallelized E2e cases to single-parallelism ones (#3718) to avoid breaking our CI pipeline.

[leonardBang]

Thanks @yuxiqian for the contribution, it's great that you've added the flow picture to make the PR easy to catch.

[yuxiqian]

@leonardBang Will this PR be reviewed soon? I'm planning to implement FLINK-36763 based on this.

[leonardBang]

@leonardBang Will this PR be reviewed soon? I'm planning to implement FLINK-36763 based on this.

Added to my todo list

[Shawn-Hx]

Thanks for the great work! Left some minor comments.

[yuxiqian]

Thanks for @Shawn-Hx's review! Addressed your comments.

Sorry to make your review more painful, but considering FLINK-36763 is expected to modify current codebase anyway, I've baked changes in this PR into #3801. Looking forward to your comments on that, too.