Fixed the problem of using the same executor to process schema change requests and flush event requests, resulting in blocking timeout. #3858
Conversation
There was a problem hiding this comment.
Thanks for @linjianchang's quick fix! Just left some trivial comments.
| @@ -96,6 +97,9 @@ public abstract class SchemaRegistry implements OperatorCoordinator, Coordinatio | |||
| protected transient SchemaManager schemaManager; | |||
| protected transient TableIdRouter router; | |||
|
|
|||
| /** Executor service to execute handle event from operator. */ | |||
| private final ExecutorService runInEventFromOperatorExecutor; | |||
There was a problem hiding this comment.
Keep its name consistent with regular schema coordinator
| @@ -253,6 +260,7 @@ public final CompletableFuture<CoordinationResponse> handleCoordinationRequest( | |||
| public final void handleEventFromOperator( | |||
| int subTaskId, int attemptNumber, OperatorEvent event) { | |||
| runInEventLoop( | |||
| runInEventFromOperatorExecutor, | |||
There was a problem hiding this comment.
This seems incorrect. handleEventFromOperator should be submitted to the same single threaded executor like other methods (so they won't be scheduled simultaneously with other critical methods). Only the SchemaCoordinator#startSchemaChange needs to be wrapped in another executor.
There was a problem hiding this comment.
IIUC this problem will only occur if:
- Distributed schema evolution topology is created
- Flush success event takes a while to finish
- The last schema operator initiates request before any flush succeeds
Thus, The handler of FlushSuccessEvent will wait for schema evolution to finish, but the busy-loop is still waiting for collecting all flush success events.
Could you please add a test case to verify this change?
There was a problem hiding this comment.
The test case org.apache.flink.cdc.runtime.operators.schema.distributed.SchemaEvolveTest#testLenientSchemaEvolution() has been covered this case.
There was a problem hiding this comment.
I'm afraid it's not sufficient because runtime unit tests uses ValuesDataSink only. It takes no time to flush and might not be able to verify this.
| @@ -325,6 +337,7 @@ public final void resetToCheckpoint(long checkpointId, @Nullable byte[] checkpoi | |||
| * directly, make sure you're running heavy logics inside, or the entire job might hang! | |||
| */ | |||
| protected void runInEventLoop( | |||
| final ExecutorService coordinatorExecutor, | |||
There was a problem hiding this comment.
It's nice to allow specifying ExecutorService when
calling runInEventLoop. Maybe regular/SchemaCoordinator could also invoke this instead of this:
schemaChangeThreadPool.submit(
() -> {
try {
applySchemaChange(originalEvent, deducedSchemaChangeEvents);
} catch (Throwable t) {
failJob(
"Schema change applying task",
new FlinkRuntimeException(
"Failed to apply schema change event.", t));
throw t;
}
});…or to process schema change requests and flush event requests, resulting in blocking timeout.
linjianchang
force-pushed
the
master-37110
branch
from
032e560 to
0eaf9ef
Compare
|
Already modified according to comment @yuxiqian |
|
Seems there's something wrong with internal state switching logic. Would @linjianchang like to take a further look? 
|
When source generate metadata events in parallel for each table,There is a problem of timeout waiting for flush to complete. Exception be like:
