yuxiqian opened a new pull request, #3567:
URL: https://github.com/apache/flink-cdc/pull/3567
Currently, SchemaRegistryRequestHandler regards Flush success when it has
collected all FlushSuccessEvent from all registered subTasks.
```java
if (flushedSinkWriters.equals(activeSinkWriters)) {
// ...
}
```
However, some subTasks might not be registered early enough, which means
flush success condition might be falsely triggered:
```c
// 6 early came sink tasks have registered & flushed successfully
SchemaRegistry - Sink subtask 5 succeed flushing for table
default_namespace.default_schema.table1. Attempt number: 0
SchemaRegistry - Sink subtask 0 succeed flushing for table
default_namespace.default_schema.table1. Attempt number: 0
SchemaRegistry - Sink subtask 1 succeed flushing for table
default_namespace.default_schema.table1. Attempt number: 0
SchemaRegistry - Sink subtask 2 succeed flushing for table
default_namespace.default_schema.table1. Attempt number: 0
SchemaRegistry - Sink subtask 4 succeed flushing for table
default_namespace.default_schema.table1. Attempt number: 0
SchemaRegistry - Sink subtask 7 succeed flushing for table
default_namespace.default_schema.table1. Attempt number: 0
// Mistake here: Schema registry thought it has collected enough flush events
schemaChangeStatus WAITING_FOR_FLUSH -> APPLYING for request
[CreateTableEvent{tableId=default_namespace.default_schema.table1,
schema=columns={`col1` STRING,`col2` STRING}, primaryKeys=col1, options=()}].
Collected enough flush writers [0, 1, 2, 4, 5, 7] of [0, 1, 2, 4, 5, 7]
// This is not correct: some subtasks have not been registered yet
SchemaRegistryRequestHandler - All sink subtask have flushed for table
default_namespace.default_schema.table1. Start to apply schema change.
// Late-coming flush events will wrongly trigger transition again
SchemaRegistryRequestHandler - Register sink subtask 3.
SchemaRegistryRequestHandler - Register sink subtask 6.
SchemaRegistry - Sink subtask 3 succeed flushing for table
default_namespace.default_schema.table1. Attempt number: 0
SchemaRegistry - Sink subtask 6 succeed flushing for table
default_namespace.default_schema.table1. Attempt number: 0
// Causing schema registry crash
java.lang.IllegalStateException: Illegal schemaChangeStatus state: should be
WAITING_FOR_FLUSH before collecting enough FlushEvents, not FINISHED. When
receiving sink success from 6 with table
default_namespace.default_schema.table1. Current flushedSinkWriters: [0, 1, 2,
3, 4, 5, 6, 7]; activeSinkWriters: [0, 1, 2, 3, 4, 5, 6, 7]
```
After this change, the `WAITING_FOR_FLUSH -> APPLYING` transition also
verifies if all subTasks has been registered (should be >= parallelism).
```
SchemaRegistry - Sink subtask 4 succeed flushing for table
default_namespace.default_schema.table1. Attempt number: 0
SchemaRegistryRequestHandler - Not all active sink writers are ready,
waiting...
SchemaRegistry - Sink subtask 2 succeed flushing for table
default_namespace.default_schema.table1. Attempt number: 0
SchemaRegistryRequestHandler - Not all active sink writers are ready,
waiting...
SchemaRegistry - Sink subtask 6 succeed flushing for table
default_namespace.default_schema.table1. Attempt number: 0
SchemaRegistryRequestHandler - Not all active sink writers are ready,
waiting...
SchemaRegistry - Sink subtask 3 succeed flushing for table
default_namespace.default_schema.table1. Attempt number: 0
SchemaRegistryRequestHandler - Not all active sink writers are ready,
waiting...
SchemaRegistry - Sink subtask 0 succeed flushing for table
default_namespace.default_schema.table1. Attempt number: 0
SchemaRegistryRequestHandler - Not all active sink writers are ready,
waiting...
SchemaRegistry - Sink subtask 1 succeed flushing for table
default_namespace.default_schema.table1. Attempt number: 0
SchemaRegistryRequestHandler - Not all active sink writers are ready,
waiting...
SchemaRegistry - Sink subtask 5 succeed flushing for table
default_namespace.default_schema.table1. Attempt number: 0
SchemaRegistryRequestHandler - Not all active sink writers are ready,
waiting...
// Will not start schema change until all sinks registered and flushed
SchemaRegistryRequestHandler - Register sink subtask 7.
SchemaRegistry - Sink subtask 7 succeed flushing for table
default_namespace.default_schema.table1. Attempt number: 0
// OK now
SchemaRegistryRequestHandler - SchemaChangeStatus WAITING_FOR_FLUSH ->
APPLYING
SchemaRegistryRequestHandler - All sink subtask have flushed for table
default_namespace.default_schema.table1. Start to apply schema change.
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
