flink and akka HTTP
Locked
 
|
Good evening, According to Flink's documentation I have excluded the Flink runtime library from the runtime dependencies of my project: dependencies { compileOnly group: 'org.apache.flink', name: 'flink-core', version: '1.4.2' compileOnly group: 'org.apache.flink', name: 'flink-java', version: '1.4.2' compileOnly group: 'org.apache.flink', name: 'flink-streaming-java_2.11', version: '1.4.2' implementation group: 'org.apache.flink', name: 'flink-connector-kafka-0.11_2.11', version: '1.4.2' ... } Unfortunately I get the following error: Caused by: java.lang.ClassCastException: interface akka.serialization.Serializer is not assignable from class akka.remote.serialization.MiscMessageSerializer at akka.actor.ReflectiveDynamicAccess$$anonfun$getClassFor$1.apply(ReflectiveDynamicAccess.scala:23) at akka.actor.ReflectiveDynamicAccess$$anonfun$getClassFor$1.apply(ReflectiveDynamicAccess.scala:20) at scala.util.Try$.apply(Try.scala:192) at akka.actor.ReflectiveDynamicAccess.getClassFor(ReflectiveDynamicAccess.scala:20) at akka.actor.ReflectiveDynamicAccess.createInstanceFor(ReflectiveDynamicAccess.scala:38) at akka.serialization.Serialization.serializerOf(Serialization.scala:301) at akka.serialization.Serialization$$anonfun$6.apply(Serialization.scala:327) at akka.serialization.Serialization$$anonfun$6.apply(Serialization.scala:327) at scala.collection.TraversableLike$WithFilter$$anonfun$map$2.apply(TraversableLike.scala:683) at scala.collection.immutable.HashMap$HashMap1.foreach(HashMap.scala:221) at scala.collection.immutable.HashMap$HashTrieMap.foreach(HashMap.scala:428) at scala.collection.TraversableLike$WithFilter.map(TraversableLike.scala:682) at akka.serialization.Serialization.<init>(Serialization.scala:327) at akka.serialization.SerializationExtension$.createExtension(SerializationExtension.scala:15) at akka.serialization.SerializationExtension$.createExtension(SerializationExtension.scala:12) at akka.actor.ActorSystemImpl.registerExtension(ActorSystem.scala:913) at akka.actor.ActorSystemImpl$$anonfun$loadExtensions$1$1.apply(ActorSystem.scala:946) at akka.actor.ActorSystemImpl$$anonfun$loadExtensions$1$1.apply(ActorSystem.scala:944) at scala.collection.Iterator$class.foreach(Iterator.scala:891) at scala.collection.AbstractIterator.foreach(Iterator.scala:1334) at scala.collection.IterableLike$class.foreach(IterableLike.scala:72) at scala.collection.AbstractIterable.foreach(Iterable.scala:54) 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.ActorSystem$.apply(ActorSystem.scala:246) at akka.actor.ActorSystem$.apply(ActorSystem.scala:289) at akka.actor.ActorSystem$.apply(ActorSystem.scala:234) at akka.actor.ActorSystem$.apply(ActorSystem.scala:225) at akka.actor.ActorSystem$.create(ActorSystem.scala:160) at akka.actor.ActorSystem.create(ActorSystem.scala) at de.eso.swarm.rest.client.akka.AkkaRestClient.<clinit>(AkkaRestClient.java:43) ... 12 more My application needs to initialize the AKKA ActorSystem because it uses an HTTP client that I have developed using akka-http and akka-stream. Here are the dependencies of my HTTP client: dependencies { compile project(':platform-sdk-java-core') testCompile project(':platform-sdk-java-testing') implementation group: 'com.typesafe.akka', name: 'akka-http_2.11', version: '10.1.2' implementation group: 'com.typesafe.akka', name: 'akka-stream_2.11', version: '2.5.11' implementation group: 'com.google.code.gson', name: 'gson', version: '2.8.4' implementation group: 'com.google.protobuf', name: 'protobuf-java', version: '3.5.1' testImplementation group: 'junit', name: 'junit', version: '4.12' } On the other hand, when I add the flink runtime to my runtime dependencies, everything works: dependencies { implementation group: 'org.apache.flink', name: 'flink-core', version: '1.4.2' implementation group: 'org.apache.flink', name: 'flink-java', version: '1.4.2' implementation group: 'org.apache.flink', name: 'flink-streaming-java_2.11', version: '1.4.2' implementation group: 'org.apache.flink', name: 'flink-connector-kafka-0.11_2.11', version: '1.4.2' } Relocation of AKKA did not solve the problem, because AKKA did no longer find its serialization classes.
Using an older version of akka http and akka streams is also not an option, because the client won't compile in that case. Are the any other ideas? Thanks in advance Martin ---------------------------------------------------------------------------------- Martin Gäckler
Entwicklung Schwarmdienste Im Auftrag der
e.solutions GmbH Despagstr. 4a 85055 Ingolstadt Germany Registered Office: Despagstr. 4a 85055 Ingolstadt Germany Phone +49 8458 3332 145 e.solutions GmbH Managing Directors Uwe Reder, Dr. Riclef Schmidt-Clausen
Register Court Ingolstadt HRB 5221 |
|
|
Hi, I assume that you have an Akka dependency conflict. By adding the Akka dependency version to your user jar and enabling child first class loading you should be able to control which Akka version is loaded. The only thing you have to check is whether Flink works with a newer version of Akka. Cheers, Till On Fri, Jun 15, 2018 at 8:15 PM Gäckler Martin <[hidden email]> wrote:
|
|
|
Good morning, Thanks for the hint. AKKA streams and http are added to our user jar. If not, our HTTP client won't work. According to
the flink documentation here:
https://flink.apache.org/news/2017/12/12/release-1.4.0.html#changes-to-dynamic-class-loading-of-user-code
child first class loading is enabled by default and we did not change any settings here. Cheers Martin ---------------------------------------------------------------------------------- Martin Gäckler
Entwicklung Schwarmdienste Im Auftrag der
e.solutions GmbH Despagstr. 4a 85055 Ingolstadt Germany Registered Office: Despagstr. 4a 85055 Ingolstadt Germany Phone +49 8458 3332 145 e.solutions GmbH Managing Directors Uwe Reder, Dr. Riclef Schmidt-Clausen
Register Court Ingolstadt HRB 5221 From: Till Rohrmann [mailto:[hidden email]]
Hi, I assume that you have an Akka dependency conflict. By adding the Akka dependency version to your user jar and enabling child first class loading you should be able to control which Akka version is loaded. The only thing you have to check
is whether Flink works with a newer version of Akka. Cheers, Till On Fri, Jun 15, 2018 at 8:15 PM Gäckler Martin <[hidden email]> wrote:
|
|
|
Is it then working or not? On Mon, Jun 18, 2018, 11:52 Gäckler Martin <[hidden email]> wrote:
|
|
|
No, without including the flink runtime to my own jar, nothing works. Regards Martin ---------------------------------------------------------------------------------- Martin Gäckler
Entwicklung Schwarmdienste Im Auftrag der
e.solutions GmbH Despagstr. 4a 85055 Ingolstadt Germany Registered Office: Despagstr. 4a 85055 Ingolstadt Germany Phone +49 8458 3332 145 e.solutions GmbH Managing Directors Uwe Reder, Dr. Riclef Schmidt-Clausen
Register Court Ingolstadt HRB 5221 From: Till Rohrmann [mailto:[hidden email]]
Is it then working or not? On Mon, Jun 18, 2018, 11:52 Gäckler Martin <[hidden email]> wrote:
|
|
|
Just for clarification: The user jar contains the required Akka dependency (including akka-remote and akka-actor)? Which version of Akka do you require? On Mon, Jun 18, 2018 at 1:07 PM Gäckler Martin <[hidden email]> wrote:
|
|
|
Here are the dependencies: runtimeClasspath - Runtime classpath of source set 'main'. +--- org.apache.flink:flink-connector-kafka-0.11_2.11:1.4.2 | +--- org.apache.flink:flink-connector-kafka-0.10_2.11:1.4.2 | | +--- org.apache.flink:flink-connector-kafka-0.9_2.11:1.4.2 | | | +--- org.apache.flink:flink-connector-kafka-base_2.11:1.4.2 | | | | \--- org.apache.flink:force-shading:1.4.2 | | | +--- org.apache.kafka:kafka-clients:0.9.0.1 -> 0.11.0.2 | | | | +--- net.jpountz.lz4:lz4:1.3.0 | | | | +--- org.xerial.snappy:snappy-java:1.1.2.6 | | | | \--- org.slf4j:slf4j-api:1.7.25 | | | \--- org.apache.flink:force-shading:1.4.2 | | +--- org.apache.kafka:kafka-clients:0.10.2.1 -> 0.11.0.2 (*) | | \--- org.apache.flink:force-shading:1.4.2 | +--- org.apache.kafka:kafka-clients:0.11.0.2 (*) | \--- org.apache.flink:force-shading:1.4.2 +--- openlr:map:1.4.2 | +--- commons-configuration:commons-configuration:1.9 -> 1.10 | | +--- commons-lang:commons-lang:2.6 | | \--- commons-logging:commons-logging:1.1.1 -> 1.2 | +--- log4j:log4j:1.2.17 | \--- commons-lang:commons-lang:2.6 +--- com.twitter:chill-protobuf:0.9.2 | +--- com.twitter:chill-java:0.9.2 | | \--- com.esotericsoftware:kryo-shaded:4.0.0 | | +--- com.esotericsoftware:minlog:1.3.0 | | \--- org.objenesis:objenesis:2.2 | \--- com.esotericsoftware:kryo-shaded:4.0.0 (*) +--- de.eso.swarm:platform-sdk-java-core:latest.integration -> 0.0.1-SNAPSHOT | +--- org.apache.logging.log4j:log4j-api:2.7 -> 2.11.0 | \--- org.apache.logging.log4j:log4j-core:2.7 -> 2.11.0 | \--- org.apache.logging.log4j:log4j-api:2.11.0 +--- de.eso.swarm:ncfs-openlrmap-sdk-java:latest.integration -> 0.0.1-SNAPSHOT | +--- de.eso.swarm:platform-sdk-java-core:latest.integration -> 0.0.1-SNAPSHOT (*) | +--- mysql:mysql-connector-java:5.1.39 | +--- org.apache.logging.log4j:log4j-1.2-api:2.7 -> 2.11.0 | | +--- org.apache.logging.log4j:log4j-api:2.11.0 | | \--- org.apache.logging.log4j:log4j-core:2.11.0 (*) | +--- org.apache.logging.log4j:log4j-api:2.7 -> 2.11.0 | +--- org.apache.logging.log4j:log4j-core:2.7 -> 2.11.0 (*) | +--- commons-configuration:commons-configuration:1.10 (*) | \--- openlr:map:1.4.2 (*) +--- de.eso:tpeg-sdk-java:latest.integration -> 0.0.1-SNAPSHOT | +--- org.apache.logging.log4j:log4j-1.2-api:latest.integration -> 2.11.0 (*) | +--- org.apache.logging.log4j:log4j-api:latest.integration -> 2.11.0 | +--- org.apache.logging.log4j:log4j-core:latest.integration -> 2.11.0 (*) | +--- openlr:map:1.4.2 (*) | +--- openlr:encoder:1.4.2 | | +--- commons-lang:commons-lang:2.6 | | +--- openlr:map:1.4.2 (*) | | +--- openlr:data:1.4.2 | | | +--- openlr:map:1.4.2 (*) | | | +--- commons-lang:commons-lang:2.6 | | | +--- commons-configuration:commons-configuration:1.9 -> 1.10 (*) | | | \--- log4j:log4j:1.2.17 | | +--- log4j:log4j:1.2.17 | | \--- commons-configuration:commons-configuration:1.9 -> 1.10 (*) | +--- openlr:decoder:1.4.2 | | +--- log4j:log4j:1.2.17 | | +--- openlr:map:1.4.2 (*) | | +--- openlr:data:1.4.2 (*) | | \--- commons-configuration:commons-configuration:1.9 -> 1.10 (*) | \--- openlr:data:1.4.2 (*) +--- de.eso.swarm.aqp:aqp-sdk-java:latest.integration -> 0.0.1-SNAPSHOT | +--- org.apache.avro:avro:1.8.2 | | +--- org.codehaus.jackson:jackson-core-asl:1.9.13 | | +--- org.codehaus.jackson:jackson-mapper-asl:1.9.13 | | | \--- org.codehaus.jackson:jackson-core-asl:1.9.13 | | +--- com.thoughtworks.paranamer:paranamer:2.7 | | +--- org.xerial.snappy:snappy-java:1.1.1.3 -> 1.1.2.6 | | +--- org.apache.commons:commons-compress:1.8.1 | | +--- org.tukaani:xz:1.5 | | \--- org.slf4j:slf4j-api:1.7.7 -> 1.7.25 +--- org.apache.commons:commons-lang3:3.7 | +--- com.google.protobuf:protobuf-java:3.5.1 | +--- de.eso.swarm:platform-sdk-java-core:latest.integration -> 0.0.1-SNAPSHOT (*) | \--- xfcd.codec:xfcd-sdk-java-codec:latest.integration -> 0.0.1-SNAPSHOT | \--- com.google.protobuf:protobuf-java:3.2.0 -> 3.5.1 +--- com.here:here-data-ingestion-sdk-java-hazard-events:latest.integration -> 0.0.1-SNAPSHOT | \--- com.google.protobuf:protobuf-java:3.2.0 -> 3.5.1 \--- de.eso.swarm:here-data-ingestion-sdk-java-client:latest.integration -> 0.0.1-SNAPSHOT +--- com.here.account:here-oauth-client:latest.integration -> 0.4.13 | +--- org.ini4j:ini4j:0.5.1 | +--- commons-codec:commons-codec:1.10 | +--- com.fasterxml.jackson.core:jackson-databind:2.8.1 | | +--- com.fasterxml.jackson.core:jackson-annotations:2.8.0 | | \--- com.fasterxml.jackson.core:jackson-core:2.8.1 | \--- org.apache.httpcomponents:httpclient:4.5.2 | +--- org.apache.httpcomponents:httpcore:4.4.4 | +--- commons-logging:commons-logging:1.2 | \--- commons-codec:commons-codec:1.9 -> 1.10 +--- de.eso.swarm:platform-sdk-java-core:latest.integration -> 0.0.1-SNAPSHOT (*) +--- de.eso.swarm:platform-sdk-java-rest-client:latest.integration -> 0.0.1-SNAPSHOT | +--- de.eso.swarm:platform-sdk-java-core:0.0.1-SNAPSHOT (*) | +--- org.apache.logging.log4j:log4j-api:2.7 -> 2.11.0 | +--- org.apache.logging.log4j:log4j-core:2.7 -> 2.11.0 (*) | +--- com.typesafe.akka:akka-http_2.11:10.1.2 | | +--- org.scala-lang:scala-library:2.11.12 | | \--- com.typesafe.akka:akka-http-core_2.11:10.1.2 | | +--- org.scala-lang:scala-library:2.11.12 | | \--- com.typesafe.akka:akka-parsing_2.11:10.1.2 | | \--- org.scala-lang:scala-library:2.11.12 | +--- com.typesafe.akka:akka-stream_2.11:2.5.11 | | +--- org.scala-lang:scala-library:2.11.12 | | +--- com.typesafe.akka:akka-actor_2.11:2.5.11 | | | +--- org.scala-lang:scala-library:2.11.12 | | | +--- com.typesafe:config:1.3.2 | | | \--- org.scala-lang.modules:scala-java8-compat_2.11:0.7.0 | | | \--- org.scala-lang:scala-library:2.11.7 -> 2.11.12 | | +--- com.typesafe.akka:akka-protobuf_2.11:2.5.11 | | | \--- org.scala-lang:scala-library:2.11.12 | | +--- org.reactivestreams:reactive-streams:1.0.2 | | \--- com.typesafe:ssl-config-core_2.11:0.2.2 | | +--- org.scala-lang:scala-library:2.11.8 -> 2.11.12 | | +--- com.typesafe:config:1.2.0 -> 1.3.2 | | \--- org.scala-lang.modules:scala-parser-combinators_2.11:1.0.4 | | \--- org.scala-lang:scala-library:2.11.6 -> 2.11.12 | +--- com.google.code.gson:gson:2.8.4 | \--- com.google.protobuf:protobuf-java:3.5.1 \--- com.here:here-data-ingestion-sdk-java-hazard-events:0.0.1-SNAPSHOT (*) There is no akka-remote Martin ---------------------------------------------------------------------------------- Martin Gäckler
Entwicklung Schwarmdienste Im Auftrag der
e.solutions GmbH Despagstr. 4a 85055 Ingolstadt Germany Registered Office: Despagstr. 4a 85055 Ingolstadt Germany Phone +49 8458 3332 145 e.solutions GmbH Managing Directors Uwe Reder, Dr. Riclef Schmidt-Clausen
Register Court Ingolstadt HRB 5221 From: Till Rohrmann [mailto:[hidden email]]
Just for clarification: The user jar contains the required Akka dependency (including akka-remote and akka-actor)? Which version of Akka do you require? On Mon, Jun 18, 2018 at 1:07 PM Gäckler Martin <[hidden email]> wrote:
|
|
|
Try adding akka-remote:2.5.11 to see whether it resolves the issues. The problem is that you have akka-actor as a dependency which is also required by Flink. If Flink loads akka-actor:2.5.11 and then tries to use akka-remote:2.4.x, this might cause the problem. Cheers, Till On Mon, Jun 18, 2018 at 2:17 PM Gäckler Martin <[hidden email]> wrote:
|
|
|
Hi, akka-remote was the golden keyword. This line in my gradle file runtimeOnly group: 'com.typesafe.akka', name: 'akka-remote_2.11', version: '2.5.11' solved the problem. Thank You very much. Martin ---------------------------------------------------------------------------------- Martin Gäckler
Entwicklung Schwarmdienste Im Auftrag der
e.solutions GmbH Despagstr. 4a 85055 Ingolstadt Germany Registered Office: Despagstr. 4a 85055 Ingolstadt Germany Phone +49 8458 3332 145 e.solutions GmbH Managing Directors Uwe Reder, Dr. Riclef Schmidt-Clausen
Register Court Ingolstadt HRB 5221 From: Till Rohrmann [mailto:[hidden email]]
Try adding akka-remote:2.5.11 to see whether it resolves the issues. The problem is that you have akka-actor as a dependency which is also required by Flink. If Flink loads akka-actor:2.5.11 and then tries to use akka-remote:2.4.x, this
might cause the problem. Cheers, Till On Mon, Jun 18, 2018 at 2:17 PM Gäckler Martin <[hidden email]> wrote:
|
| Free forum by Nabble | Edit this page |