You are viewing a plain text version of this content. The canonical link for it is here.

Posted to issues@flink.apache.org by "yuxiqian (via GitHub)" <gi...@apache.org> on 2024/08/23 03:50:07 UTC

[PR] [hotfix] Fix schema registry hanging in multiple parallelism [flink-cdc]

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: issues-unsubscribe@flink.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] [hotfix] Fix schema registry hanging in multiple parallelism [flink-cdc]

Posted by "yuxiqian (via GitHub)" <gi...@apache.org>.

yuxiqian commented on PR #3567:
URL: https://github.com/apache/flink-cdc/pull/3567#issuecomment-2306127604

   @leonardBang @ruanhang1993 PTAL


-- 
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: issues-unsubscribe@flink.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] [hotfix] Fix schema registry flush state switch logic with multiple parallelism [flink-cdc]

Posted by "leonardBang (via GitHub)" <gi...@apache.org>.

leonardBang merged PR #3567:
URL: https://github.com/apache/flink-cdc/pull/3567


-- 
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: issues-unsubscribe@flink.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org