[FLINK-20261] Uncaught exception in ExecutorNotifier due to split assignment broken by failed task

• Type: Bug
• Status: Closed
• Priority: Blocker
• Resolution: Fixed
• Affects Version/s: 1.12.0
• Fix Version/s: 1.12.0
• Component/s: Connectors / Common, Connectors / FileSystem
• Assignee: Stephan Ewen
• Reporter: Andrey Zagrebin
• Created: 20/Nov/20 13:34
• Updated: 22/Nov/20 22:56
• Resolved: 22/Nov/20 22:56

Description

While trying to extend FileSourceTextLinesITCase::testContinuousTextFileSourceWithTaskManagerFailover with recovery test after TM failure (TestingMiniCluster::terminateTaskExecutor, branch) in FLINK-20118, I encountered the following case:

• SourceCoordinatorContext::assignSplits schedules async assignment (all reader tasks alive)
• call TestingMiniCluster::terminateTaskExecutor while doing writeFile in a loop of testContinuousTextFileSource
• causes graceful TaskExecutor::onStop shutdown
• causes TM/RM disconnect and failing slot allocations in JM by RM
• eventually causes SourceCoordinatorContext::unregisterSourceReader
• actual assignment starts (SourceCoordinatorContext::assignSplits: callInCoordinatorThread)
• registeredReaders.containsKey(subtaskId) check fails (due to failed task) with IllegalArgumentException which is uncaught in single thread executor
• forces ThreadPool to recreate the single thread
• calls CoordinatorExecutorThreadFactory::newThread
• fails expected condition of single thread creation with IllegalStateException which is uncaught
• calls FatalExitExceptionHandler and exits JVM abruptly

[SourceCoordinator-Source: file-source] ERROR org.apache.flink.runtime.util.FatalExitExceptionHandler - FATAL: Thread 'SourceCoordinator-Source: file-source' produced an uncaught exception. Stopping the process...
java.lang.IllegalStateException: Should never happen. This factory should only be used by a SingleThreadExecutor.
	at org.apache.flink.runtime.source.coordinator.SourceCoordinatorProvider$CoordinatorExecutorThreadFactory.newThread(SourceCoordinatorProvider.java:94) ~[classes/:?]
	at java.util.concurrent.ThreadPoolExecutor$Worker.<init>(ThreadPoolExecutor.java:619) ~[?:1.8.0_172]
	at java.util.concurrent.ThreadPoolExecutor.addWorker(ThreadPoolExecutor.java:932) ~[?:1.8.0_172]
	at java.util.concurrent.ThreadPoolExecutor.processWorkerExit(ThreadPoolExecutor.java:1025) ~[?:1.8.0_172]
	at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1167) ~[?:1.8.0_172]
	at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624) ~[?:1.8.0_172]
	at java.lang.Thread.run(Thread.java:748) [?:1.8.0_172]

Process finished with exit code 239

Issue Links

• blocks: FLINK-20118 Test New File Source API - Continuous Streaming Execution
• is related to: FLINK-20081 ExecutorNotifier should run handler in the main thread when receive an exception from the callable.
• is related to: FLINK-20157 SourceCoordinatorProvider kills JobManager with IllegalStateException on job submission