Flink QuickStart: On start up, running into ClassNotFoundException: org.apache.flink.streaming.runtime.tasks.OneInputStreamTask

[Prez Cannady-2]

I’m experimenting combining Spring with Flink.  I’ve successfully instrumented for Gradle, but Maven is emitting ClassNotFoundExceptions for items ostensibly on the class path.

Project is currently configured for:

1. Scala 2.10.4

2. Flink 0.9.1

I execute the following

```

# In one terminal

$ nc -lk -p 9999 --sh-exec "cat /usr/share/dict/words | head -n 10”

# In another terminal

$ mvn clean install spring-boot:run -Drun.arguments=“localhost,9999”

# observe output

```

The specific class not found is org.apache.flink.streaming.runtime.tasks.OneInputStreamTask.  However, Spring Boot Plugin is configured to repackage a fat jar, and I can see that the class is present in the included flink-streaming-core jar.  Additionally, LogBack shows that the flink-streaming-core jar is in my classpath.

I’m hoping I’m just missing something that should be obvious.  While I wish could move forward with just Gradle, unfortunately I have to support Maven builds.

For reference, the complete project is available here:

https://github.com/OCExercise/wordcount-processing

Additionally

1. pom.xml (https://github.com/OCExercise/wordcount-processing/blob/master/pom.xml)

2. build.grade (https://github.com/OCExercise/wordcount-processing/blob/master/build.gradle)

3. Gist containing the full exception (https://gist.github.com/revprez/2c1fb01c40e5d6790247)

Prez Cannady  

p: 617 500 3378  

e: [hidden email]  

GH: https://github.com/opencorrelate  

LI: https://www.linkedin.com/in/revprez  

[Stephan Ewen]

Hi!

Running this is Spring, the whole classloader configuration is probably a bit different than in Flink's standalone or YARN or local mode.

Can you try if the following solves your problem: 

At the end of the file "BlobLibraryCacheManager", there is the private class "FlinkUserCodeClassloader".

Can you replace the current FlinkUserCodeClassloader with this?

private static class FlinkUserCodeClassLoader extends URLClassLoader {

public FlinkUserCodeClassLoader(URL[] urls) {

super(urls, FlinkUserCodeClassLoader.class.getClassLoader());

}

}

You can also try and use instead of "FlinkUserCodeClassLoader.class.getClassLoader()" the statements "Thread.currentThread().getContextClassLoader()".

Let me know if one of the two solves the problem.

Greetings,

Stephan

On Wed, Jan 13, 2016 at 7:20 PM, Prez Cannady <[hidden email]> wrote:

I’m experimenting combining Spring with Flink.  I’ve successfully instrumented for Gradle, but Maven is emitting ClassNotFoundExceptions for items ostensibly on the class path.

Project is currently configured for:

1. Scala 2.10.4

2. Flink 0.9.1

I execute the following

```

# In one terminal

$ nc -lk -p 9999 --sh-exec "cat /usr/share/dict/words | head -n 10”

# In another terminal

$ mvn clean install spring-boot:run -Drun.arguments=“localhost,9999”

# observe output

```

The specific class not found is org.apache.flink.streaming.runtime.tasks.OneInputStreamTask.  However, Spring Boot Plugin is configured to repackage a fat jar, and I can see that the class is present in the included flink-streaming-core jar.  Additionally, LogBack shows that the flink-streaming-core jar is in my classpath.

I’m hoping I’m just missing something that should be obvious.  While I wish could move forward with just Gradle, unfortunately I have to support Maven builds.

For reference, the complete project is available here:

https://github.com/OCExercise/wordcount-processing

Additionally

1. pom.xml (https://github.com/OCExercise/wordcount-processing/blob/master/pom.xml)

2. build.grade (https://github.com/OCExercise/wordcount-processing/blob/master/build.gradle)

3. Gist containing the full exception (https://gist.github.com/revprez/2c1fb01c40e5d6790247)

Prez Cannady  

p: 617 500 3378  

e: [hidden email]  

GH: https://github.com/opencorrelate  

LI: https://www.linkedin.com/in/revprez  

[Prez Cannady-3]

Simply passing FlinkUserCodeClassLoader.class.getClassLoader to the parent constructor cleared the impasse. 

2016-01-13 20:06:43.637  INFO 35403 --- [           main] o.o.e.j.s.SocketTextStreamWordCount$     : Started SocketTextStreamWordCount. in 5.176 seconds (JVM running for 12.58)

[INFO] ------------------------------------------------------------------------

[INFO] BUILD SUCCESS

[INFO] ------------------------------------------------------------------------

[INFO] Total time: 11.734 s

[INFO] Finished at: 2016-01-13T20:06:43-05:00

[INFO] Final Memory: 49M/4986M

[INFO] ------------------------------------------------------------------------

2016-01-13 20:06:43.804  INFO 35403 --- [       Thread-3] s.c.a.AnnotationConfigApplicationContext : Closing org.springframework.context.annotation.AnnotationConfigApplicationContext@33248c18: startup date [Wed Jan 13 20:06:38 EST 2016]; root of context hierarchy

2016-01-13 20:06:43.806  INFO 35403 --- [       Thread-3] o.s.j.e.a.AnnotationMBeanExporter        : Unregistering JMX-exposed beans on shutdown

All tests in flink-runtime passed after the change `BlobLibraryCacheManager’, but I haven’t run the full test suite. 

Is this actually an appropriate fix, or just a way to highlight a configuration problem? 

I assume that injecting a parent class loader when registering a task might break things, but I don’t know nearly enough about Flink and this code to say one way or another.

Prez Cannady  

p: 617 500 3378  

e: [hidden email]  

GH: https://github.com/opencorrelate  

LI: https://www.linkedin.com/in/revprez  

On Jan 13, 2016, at 6:50 PM, Stephan Ewen <[hidden email]> wrote:

Hi!

Running this is Spring, the whole classloader configuration is probably a bit different than in Flink's standalone or YARN or local mode.

Can you try if the following solves your problem: 

At the end of the file "BlobLibraryCacheManager", there is the private class "FlinkUserCodeClassloader".

Can you replace the current FlinkUserCodeClassloader with this?

private static class FlinkUserCodeClassLoader extends URLClassLoader {

public FlinkUserCodeClassLoader(URL[] urls) {

super(urls, FlinkUserCodeClassLoader.class.getClassLoader());

}

}

You can also try and use instead of "FlinkUserCodeClassLoader.class.getClassLoader()" the statements "Thread.currentThread().getContextClassLoader()".

Let me know if one of the two solves the problem.

Greetings,

Stephan

On Wed, Jan 13, 2016 at 7:20 PM, Prez Cannady <[hidden email]> wrote:

I’m experimenting combining Spring with Flink.  I’ve successfully instrumented for Gradle, but Maven is emitting ClassNotFoundExceptions for items ostensibly on the class path.

Project is currently configured for:

1. Scala 2.10.4

2. Flink 0.9.1

I execute the following

```

# In one terminal

$ nc -lk -p 9999 --sh-exec "cat /usr/share/dict/words | head -n 10”

# In another terminal

$ mvn clean install spring-boot:run -Drun.arguments=“localhost,9999”

# observe output

```

The specific class not found is org.apache.flink.streaming.runtime.tasks.OneInputStreamTask.  However, Spring Boot Plugin is configured to repackage a fat jar, and I can see that the class is present in the included flink-streaming-core jar.  Additionally, LogBack shows that the flink-streaming-core jar is in my classpath.

I’m hoping I’m just missing something that should be obvious.  While I wish could move forward with just Gradle, unfortunately I have to support Maven builds.

For reference, the complete project is available here:

https://github.com/OCExercise/wordcount-processing

Additionally

1. pom.xml (https://github.com/OCExercise/wordcount-processing/blob/master/pom.xml)

2. build.grade (https://github.com/OCExercise/wordcount-processing/blob/master/build.gradle)

3. Gist containing the full exception (https://gist.github.com/revprez/2c1fb01c40e5d6790247)

Prez Cannady  

p: 617 500 3378  

e: [hidden email]  

GH: https://github.com/opencorrelate  

LI: https://www.linkedin.com/in/revprez  

[Stephan Ewen]

Hi!

I think this is a solid fix. Adding the classloader that loads Flink's classes as the parent is a good.

Do you want to open a pull request with that?

Greetings,

Stephan

On Thu, Jan 14, 2016 at 2:26 AM, Prez Cannady <[hidden email]> wrote:

Simply passing FlinkUserCodeClassLoader.class.getClassLoader to the parent constructor cleared the impasse. 

2016-01-13 20:06:43.637  INFO 35403 --- [           main] o.o.e.j.s.SocketTextStreamWordCount$     : Started SocketTextStreamWordCount. in 5.176 seconds (JVM running for 12.58)

[INFO] ------------------------------------------------------------------------

[INFO] BUILD SUCCESS

[INFO] ------------------------------------------------------------------------

[INFO] Total time: 11.734 s

[INFO] Finished at: 2016-01-13T20:06:43-05:00

[INFO] Final Memory: 49M/4986M

[INFO] ------------------------------------------------------------------------

2016-01-13 20:06:43.804  INFO 35403 --- [       Thread-3] s.c.a.AnnotationConfigApplicationContext : Closing org.springframework.context.annotation.AnnotationConfigApplicationContext@33248c18: startup date [Wed Jan 13 20:06:38 EST 2016]; root of context hierarchy

2016-01-13 20:06:43.806  INFO 35403 --- [       Thread-3] o.s.j.e.a.AnnotationMBeanExporter        : Unregistering JMX-exposed beans on shutdown

All tests in flink-runtime passed after the change `BlobLibraryCacheManager’, but I haven’t run the full test suite. 

Is this actually an appropriate fix, or just a way to highlight a configuration problem? 

I assume that injecting a parent class loader when registering a task might break things, but I don’t know nearly enough about Flink and this code to say one way or another.

Prez Cannady  

p: 617 500 3378  

e: [hidden email]  

GH: https://github.com/opencorrelate  

LI: https://www.linkedin.com/in/revprez  

On Jan 13, 2016, at 6:50 PM, Stephan Ewen <[hidden email]> wrote:

Hi!

Running this is Spring, the whole classloader configuration is probably a bit different than in Flink's standalone or YARN or local mode.

Can you try if the following solves your problem: 

At the end of the file "BlobLibraryCacheManager", there is the private class "FlinkUserCodeClassloader".

Can you replace the current FlinkUserCodeClassloader with this?

private static class FlinkUserCodeClassLoader extends URLClassLoader {

public FlinkUserCodeClassLoader(URL[] urls) {

super(urls, FlinkUserCodeClassLoader.class.getClassLoader());

}

}

You can also try and use instead of "FlinkUserCodeClassLoader.class.getClassLoader()" the statements "Thread.currentThread().getContextClassLoader()".

Let me know if one of the two solves the problem.

Greetings,

Stephan

On Wed, Jan 13, 2016 at 7:20 PM, Prez Cannady <[hidden email]> wrote:

I’m experimenting combining Spring with Flink.  I’ve successfully instrumented for Gradle, but Maven is emitting ClassNotFoundExceptions for items ostensibly on the class path.

Project is currently configured for:

1. Scala 2.10.4

2. Flink 0.9.1

I execute the following

```

# In one terminal

$ nc -lk -p 9999 --sh-exec "cat /usr/share/dict/words | head -n 10”

# In another terminal

$ mvn clean install spring-boot:run -Drun.arguments=“localhost,9999”

# observe output

```

The specific class not found is org.apache.flink.streaming.runtime.tasks.OneInputStreamTask.  However, Spring Boot Plugin is configured to repackage a fat jar, and I can see that the class is present in the included flink-streaming-core jar.  Additionally, LogBack shows that the flink-streaming-core jar is in my classpath.

I’m hoping I’m just missing something that should be obvious.  While I wish could move forward with just Gradle, unfortunately I have to support Maven builds.

For reference, the complete project is available here:

https://github.com/OCExercise/wordcount-processing

Additionally

1. pom.xml (https://github.com/OCExercise/wordcount-processing/blob/master/pom.xml)

2. build.grade (https://github.com/OCExercise/wordcount-processing/blob/master/build.gradle)

3. Gist containing the full exception (https://gist.github.com/revprez/2c1fb01c40e5d6790247)

Prez Cannady  

p: 617 500 3378  

e: [hidden email]  

GH: https://github.com/opencorrelate  

LI: https://www.linkedin.com/in/revprez  

[Prez Cannady-3]

Sure thing.  Opened pull requests #1506 against master.  Also submitted pull request #1507 against release-0.9.1.rc1 seeing as we’re still pinned to 0.9.1 for our project.  Not sure if you guys are interested in hot fixes to previous releases, but there you have it.

Prez Cannady  

p: 617 500 3378  

e: [hidden email]  

GH: https://github.com/opencorrelate  

LI: https://www.linkedin.com/in/revprez  

On Jan 14, 2016, at 4:07 AM, Stephan Ewen <[hidden email]> wrote:

Hi!

I think this is a solid fix. Adding the classloader that loads Flink's classes as the parent is a good.

Do you want to open a pull request with that?

Greetings,

Stephan

On Thu, Jan 14, 2016 at 2:26 AM, Prez Cannady <[hidden email]> wrote:

Simply passing FlinkUserCodeClassLoader.class.getClassLoader to the parent constructor cleared the impasse. 

2016-01-13 20:06:43.637  INFO 35403 --- [           main] o.o.e.j.s.SocketTextStreamWordCount$     : Started SocketTextStreamWordCount. in 5.176 seconds (JVM running for 12.58)

[INFO] ------------------------------------------------------------------------

[INFO] BUILD SUCCESS

[INFO] ------------------------------------------------------------------------

[INFO] Total time: 11.734 s

[INFO] Finished at: 2016-01-13T20:06:43-05:00

[INFO] Final Memory: 49M/4986M

[INFO] ------------------------------------------------------------------------

2016-01-13 20:06:43.804  INFO 35403 --- [       Thread-3] s.c.a.AnnotationConfigApplicationContext : Closing org.springframework.context.annotation.AnnotationConfigApplicationContext@33248c18: startup date [Wed Jan 13 20:06:38 EST 2016]; root of context hierarchy

2016-01-13 20:06:43.806  INFO 35403 --- [       Thread-3] o.s.j.e.a.AnnotationMBeanExporter        : Unregistering JMX-exposed beans on shutdown

All tests in flink-runtime passed after the change `BlobLibraryCacheManager’, but I haven’t run the full test suite. 

Is this actually an appropriate fix, or just a way to highlight a configuration problem? 

I assume that injecting a parent class loader when registering a task might break things, but I don’t know nearly enough about Flink and this code to say one way or another.

Prez Cannady  

p: 617 500 3378  

e: [hidden email]  

GH: https://github.com/opencorrelate  

LI: https://www.linkedin.com/in/revprez  

On Jan 13, 2016, at 6:50 PM, Stephan Ewen <[hidden email]> wrote:

Hi!

Running this is Spring, the whole classloader configuration is probably a bit different than in Flink's standalone or YARN or local mode.

Can you try if the following solves your problem: 

At the end of the file "BlobLibraryCacheManager", there is the private class "FlinkUserCodeClassloader".

Can you replace the current FlinkUserCodeClassloader with this?

private static class FlinkUserCodeClassLoader extends URLClassLoader {

public FlinkUserCodeClassLoader(URL[] urls) {

super(urls, FlinkUserCodeClassLoader.class.getClassLoader());

}

}

You can also try and use instead of "FlinkUserCodeClassLoader.class.getClassLoader()" the statements "Thread.currentThread().getContextClassLoader()".

Let me know if one of the two solves the problem.

Greetings,

Stephan

On Wed, Jan 13, 2016 at 7:20 PM, Prez Cannady <[hidden email]> wrote:

I’m experimenting combining Spring with Flink.  I’ve successfully instrumented for Gradle, but Maven is emitting ClassNotFoundExceptions for items ostensibly on the class path.

Project is currently configured for:

1. Scala 2.10.4

2. Flink 0.9.1

I execute the following

```

# In one terminal

$ nc -lk -p 9999 --sh-exec "cat /usr/share/dict/words | head -n 10”

# In another terminal

$ mvn clean install spring-boot:run -Drun.arguments=“localhost,9999”

# observe output

```

The specific class not found is org.apache.flink.streaming.runtime.tasks.OneInputStreamTask.  However, Spring Boot Plugin is configured to repackage a fat jar, and I can see that the class is present in the included flink-streaming-core jar.  Additionally, LogBack shows that the flink-streaming-core jar is in my classpath.

I’m hoping I’m just missing something that should be obvious.  While I wish could move forward with just Gradle, unfortunately I have to support Maven builds.

For reference, the complete project is available here:

https://github.com/OCExercise/wordcount-processing

Additionally

1. pom.xml (https://github.com/OCExercise/wordcount-processing/blob/master/pom.xml)

2. build.grade (https://github.com/OCExercise/wordcount-processing/blob/master/build.gradle)

3. Gist containing the full exception (https://gist.github.com/revprez/2c1fb01c40e5d6790247)

Prez Cannady  

p: 617 500 3378  

e: [hidden email]  

GH: https://github.com/opencorrelate  

LI: https://www.linkedin.com/in/revprez