Great to hear that it works now :-)
On Fri, Oct 2, 2020 at 2:17 AM Lian Jiang <jiangok2...@gmail.com> wrote:
> Thanks Till. Making the scala version consistent using 2.11 solved the
> ClassNotFoundException.
>
> On Tue, Sep 29, 2020 at 11:58 PM Till Rohrmann <trohrm...@apache.org>
> wrote:
>
>> 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
>>>
>>>
>>>
>
> --
>
> Create your own email signature
> <https://www.wisestamp.com/signature-in-email/?utm_source=promotion&utm_medium=signature&utm_campaign=create_your_own&srcid=5234462839406592>
>
