Hi Lian,
I suspect that it is caused by an incompatible Akka version. Flink uses
Akka 2.5.21 instead of 2.5.12. Moreover, you are mixing Flink jars which
use Scala 2.11 with Akka dependencies which are built against Scala 2.12.
I am not an Gradle expert but can't Gradle simply pull in the transitive
dependencies of flink-runtime?
Cheers,
Till
On Wed, Sep 30, 2020 at 2:22 AM Lian Jiang <jiangok2...@gmail.com> wrote:
> Hi,
>
> I use Flink source master to build a snapshot and use the jars in my
> project. The goal is to avoid hacky deserialization code caused by avro 1.8
> in old Flink versions since Flink 1.12 uses avro 1.10. Unfortunately, the
> code throws below ClassNotFoundException. I have verified that the
> akka-actor jar 2.5.12 is available and specified in -classpath. I can even
> create an object using akka/serialization/NullSerializer class in my
> application, indicating there is no problem for this app to use any class
> under namespace akka/serialization.
>
> Caused by: java.lang.NoClassDefFoundError:
> akka/serialization/BaseSerializer$class
> at
> akka.remote.serialization.MiscMessageSerializer.<init>(MiscMessageSerializer.scala:25)
> at sun.reflect.NativeConstructorAccessorImpl.newInstance0(Native Method)
> at
> sun.reflect.NativeConstructorAccessorImpl.newInstance(NativeConstructorAccessorImpl.java:62)
> at
> sun.reflect.DelegatingConstructorAccessorImpl.newInstance(DelegatingConstructorAccessorImpl.java:45)
> at java.lang.reflect.Constructor.newInstance(Constructor.java:423)
> at
> akka.actor.ReflectiveDynamicAccess.$anonfun$createInstanceFor$1(ReflectiveDynamicAccess.scala:33)
> at scala.util.Try$.apply(Try.scala:213)
> at
> akka.actor.ReflectiveDynamicAccess.createInstanceFor(ReflectiveDynamicAccess.scala:28)
> at
> akka.actor.ReflectiveDynamicAccess.$anonfun$createInstanceFor$4(ReflectiveDynamicAccess.scala:39)
> at scala.util.Success.flatMap(Try.scala:251)
> at
> akka.actor.ReflectiveDynamicAccess.createInstanceFor(ReflectiveDynamicAccess.scala:39)
> at akka.serialization.Serialization.serializerOf(Serialization.scala:320)
> at
> akka.serialization.Serialization.$anonfun$serializers$2(Serialization.scala:346)
> at
> scala.collection.TraversableLike$WithFilter.$anonfun$map$2(TraversableLike.scala:874)
> at scala.collection.immutable.HashMap$HashMap1.foreach(HashMap.scala:394)
> at
> scala.collection.immutable.HashMap$HashTrieMap.foreach(HashMap.scala:721)
> at
> scala.collection.TraversableLike$WithFilter.map(TraversableLike.scala:873)
> at akka.serialization.Serialization.<init>(Serialization.scala:346)
> at
> akka.serialization.SerializationExtension$.createExtension(SerializationExtension.scala:16)
> at
> akka.serialization.SerializationExtension$.createExtension(SerializationExtension.scala:13)
> at akka.actor.ActorSystemImpl.registerExtension(ActorSystem.scala:913)
> at
> akka.actor.ActorSystemImpl.$anonfun$loadExtensions$1(ActorSystem.scala:946)
> at scala.collection.Iterator.foreach(Iterator.scala:943)
> at scala.collection.Iterator.foreach$(Iterator.scala:943)
> at scala.collection.AbstractIterator.foreach(Iterator.scala:1431)
> at scala.collection.IterableLike.foreach(IterableLike.scala:74)
> at scala.collection.IterableLike.foreach$(IterableLike.scala:73)
> at scala.collection.AbstractIterable.foreach(Iterable.scala:56)
> at akka.actor.ActorSystemImpl.loadExtensions$1(ActorSystem.scala:944)
> at akka.actor.ActorSystemImpl.loadExtensions(ActorSystem.scala:961)
> at akka.actor.ActorSystemImpl.liftedTree2$1(ActorSystem.scala:833)
> at akka.actor.ActorSystemImpl._start$lzycompute(ActorSystem.scala:823)
> at akka.actor.ActorSystemImpl._start(ActorSystem.scala:823)
> at akka.actor.ActorSystemImpl.start(ActorSystem.scala:842)
> at akka.actor.RobustActorSystem$.internalApply(RobustActorSystem.scala:96)
> at akka.actor.RobustActorSystem$.apply(RobustActorSystem.scala:70)
> at akka.actor.RobustActorSystem$.create(RobustActorSystem.scala:55)
> at
> org.apache.flink.runtime.akka.AkkaUtils$.createActorSystem(AkkaUtils.scala:125)
> at
> org.apache.flink.runtime.akka.AkkaUtils.createActorSystem(AkkaUtils.scala)
> at
> org.apache.flink.runtime.clusterframework.BootstrapTools.startActorSystem(BootstrapTools.java:276)
> at
> org.apache.flink.runtime.clusterframework.BootstrapTools.startLocalActorSystem(BootstrapTools.java:260)
> ... 11 more
>
>
> This is my gradle:
>
> implementation files('lib/flink-avro-confluent-registry-1.12-SNAPSHOT.jar')
> implementation files('lib/flink-clients_2.11-1.12-SNAPSHOT.jar')
> implementation files('lib/flink-connector-kafka_2.11-1.12-SNAPSHOT.jar')
> implementation files('lib/flink-connector-wikiedits_2.11-1.12-SNAPSHOT.jar')
> implementation files('lib/flink-core-1.12-SNAPSHOT.jar')
> implementation files('lib/flink-java-1.12-SNAPSHOT.jar')
> implementation files('lib/flink-metrics-dropwizard-1.12-SNAPSHOT.jar')
> implementation files('lib/flink-streaming-java_2.11-1.12-SNAPSHOT.jar')
> implementation files('lib/flink-connector-kafka-base_2.11-1.12-SNAPSHOT.jar')
> implementation files('lib/flink-avro-1.12-SNAPSHOT.jar')
> implementation files('lib/flink-annotations-1.12-SNAPSHOT.jar')
> implementation files('lib/flink-runtime_2.11-1.12-SNAPSHOT.jar')
> implementation files('lib/flink-shaded-asm-7-7.1-11.0.jar')
> implementation files('lib/flink-metrics-core-1.12-SNAPSHOT.jar')
> implementation files('lib/flink-optimizer_2.11-1.12-SNAPSHOT.jar')
> implementation files('lib/flink-shaded-guava-18.0-11.0.jar')
>
> implementation group: 'org.scala-lang', name: 'scala-library', version:
> '2.12.12'
> implementation group: 'org.apache.commons', name: 'commons-lang3', version:
> '3.11'
> implementation group: 'commons-collections', name: 'commons-collections',
> version: '3.2.2'
> implementation group: 'com.esotericsoftware.kryo', name: 'kryo', version:
> '2.16'
> implementation group: 'com.typesafe', name: 'config', version: '1.3.2'
>
> implementation group: 'com.typesafe.akka', name: 'akka-remote_2.12', version:
> '2.5.12'
> implementation group: 'com.typesafe.akka', name: 'akka-slf4j_2.12', version:
> '2.5.12'
> implementation group: 'com.typesafe.akka', name: 'akka-stream_2.12', version:
> '2.5.12'
> implementation group: 'com.typesafe.akka', name: 'akka-protobuf_2.12',
> version: '2.5.12'
>
> Note: the row of "akka-slf4j" will automatically include "akka-actor" so I
> don't need to add "akka-actor" explicitly.
>
> Since Flink 1.12 will be released in Oct 2020, it makes sense for me to
> use it for development to save a lot of workaround code dealing with the
> avro 1.8 serialization issue [1] in older Flink. However, this exception is
> blocking me from doing so. Any idea is highly appreciated!
>
> Thanks
> Lian
>
>
> [1] https://issues.apache.org/jira/browse/FLINK-19339
>
>
>
