[FLINK-11431] Upgrade Akka to 2.5

• Type: Sub-task
• Status: Closed
• Priority: Minor
• Resolution: Fixed
• Affects Version/s: 1.7.1
• Fix Version/s: 1.9.0
• Component/s: Build System, Runtime / Coordination
• Labels: pull-request-available
• Assignee: Chesnay Schepler
• Reporter: Matthieu Bonneviot
• Created: 24/Jan/19 15:38
• Updated: 18/Apr/19 14:57
• Resolved: 18/Apr/19 14:57

Description

2019-01-24 14:43:52,059 ERROR akka.remote.Remoting                                          - class [B cannot be cast to class [C ([B and [C are in module java.base of loader 'bootstrap')
java.lang.ClassCastException: class [B cannot be cast to class [C ([B and [C are in module java.base of loader 'bootstrap')
        at akka.remote.artery.FastHash$.ofString(LruBoundedCache.scala:18)
        at akka.remote.serialization.ActorRefResolveCache.hash(ActorRefResolveCache.scala:61)
        at akka.remote.serialization.ActorRefResolveCache.hash(ActorRefResolveCache.scala:55)
        at akka.remote.artery.LruBoundedCache.getOrCompute(LruBoundedCache.scala:110)
        at akka.remote.RemoteActorRefProvider.resolveActorRef(RemoteActorRefProvider.scala:403)
        at akka.actor.SerializedActorRef.readResolve(ActorRef.scala:433)
        at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
        at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
        at java.base/jdk.internal.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
        at java.base/java.lang.reflect.Method.invoke(Method.java:566)
        at java.base/java.io.ObjectStreamClass.invokeReadResolve(ObjectStreamClass.java:1250)
        at java.base/java.io.ObjectInputStream.readOrdinaryObject(ObjectInputStream.java:2096)
        at java.base/java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1594)Running a jobmanager with java 11 fail with the following call stack:

Flink master is using akka 2.4.20.

After some investigation, the error in akka comes from the following line:

def ofString(s: String): Int = {
val chars = Unsafe.instance.getObject(s, EnvelopeBuffer.StringValueFieldOffset).asInstanceOf[Array[Char]]

from java 9 it is now an array of byte. The akka code in the newer version is:

    public static int fastHash(String str) {
          ...
        if (isJavaVersion9Plus) {
            final byte[] chars = (byte[]) instance.getObject(str, stringValueFieldOffset);
            ...
        } else {
            final char[] chars = (char[]) instance.getObject(str, stringValueFieldOffset);
 

Issue Links

• is blocked by: FLINK-10392 Remove legacy mode
• links to: GitHub Pull Request #8036