[FLINK-18352] org.apache.flink.core.execution.DefaultExecutorServiceLoader not thread safe

• Type: Bug
• Status: Closed
• Priority: Major
• Resolution: Fixed
• Affects Version/s: 1.10.0
• Fix Version/s: 1.10.2, 1.11.0, 1.12.0
• Component/s: Client / Job Submission
• Labels: pull-request-available
• Assignee: Kostas Kloudas
• Reporter: Marcos Klein
• Created: 17/Jun/20 23:30
• Updated: 23/Jun/20 06:30
• Resolved: 22/Jun/20 14:26

Description

The singleton nature of the  org.apache.flink.core.execution.DefaultExecutorServiceLoader class is not thread-safe due to the fact that java.util.ServiceLoader class is not thread-safe.

https://docs.oracle.com/en/java/javase/11/docs/api/java.base/java/util/ServiceLoader.html#Concurrency

 

This can result in ServiceLoader class entering into an inconsistent state for processes which attempt to self-heal. This then requires bouncing the process/container in the hopes the race condition does not re-occur.

https://stackoverflow.com/questions/60391499/apache-flink-cannot-find-compatible-factory-for-specified-execution-target-lo

 

Additionally the following stack traces have been seen when using a org.apache.flink.streaming.api.environment.RemoteStreamEnvironment instances.

java.lang.ArrayIndexOutOfBoundsException: 2
    at sun.misc.CompoundEnumeration.nextElement(CompoundEnumeration.java:61)
    at java.util.ServiceLoader$LazyIterator.hasNextService(ServiceLoader.java:357)
    at java.util.ServiceLoader$LazyIterator.hasNext(ServiceLoader.java:393)
    at java.util.ServiceLoader$1.hasNext(ServiceLoader.java:474)
    at org.apache.flink.core.execution.DefaultExecutorServiceLoader.getExecutorFactory(DefaultExecutorServiceLoader.java:60)
    at org.apache.flink.streaming.api.environment.StreamExecutionEnvironment.executeAsync(StreamExecutionEnvironment.java:1724)
    at org.apache.flink.streaming.api.environment.StreamExecutionEnvironment.executeAsync(StreamExecutionEnvironment.java:1706)

 

java.util.NoSuchElementException: null
    at sun.misc.CompoundEnumeration.nextElement(CompoundEnumeration.java:59)
    at java.util.ServiceLoader$LazyIterator.hasNextService(ServiceLoader.java:357)
    at java.util.ServiceLoader$LazyIterator.hasNext(ServiceLoader.java:393)
    at java.util.ServiceLoader$1.hasNext(ServiceLoader.java:474)
    at org.apache.flink.core.execution.DefaultExecutorServiceLoader.getExecutorFactory(DefaultExecutorServiceLoader.java:60)
    at org.apache.flink.streaming.api.environment.StreamExecutionEnvironment.executeAsync(StreamExecutionEnvironment.java:1724)
    at org.apache.flink.streaming.api.environment.StreamExecutionEnvironment.executeAsync(StreamExecutionEnvironment.java:1706)

The workaround for using the **StreamExecutionEnvironment implementations is to write a custom implementation of DefaultExecutorServiceLoader which is thread-safe and pass that to the StreamExecutionEnvironment constructors.

Issue Links

• causes: FLINK-18411 CollectionExecutorTest failed to compiled in release-1.11
• links to: GitHub Pull Request #12719