[jira] [Created] (FLINK-18459) ClassNotFoundException in ProcessFunction property object's callback and promise

[Shang Yuanchun (Jira)]

Xinhao Li created FLINK-18459:
---------------------------------

             Summary: ClassNotFoundException in ProcessFunction property object's callback and promise
                 Key: FLINK-18459
                 URL: https://issues.apache.org/jira/browse/FLINK-18459
             Project: Flink
          Issue Type: Bug
          Components: Build System
    Affects Versions: 1.10.0
         Environment: The environment is:
Flink version is 1.10.0 and using shaded {{netty}}, {{hadoop}}, {{guava}} and {{jackson}}.
Using cassandra-driver-mapping: 3.9.0 and shaded cassandra-driver-core: 3.9.0.
            Reporter: Xinhao Li


I am developing a Flink application and it will use Cassandra Driver to interact with Cassandra DB. The Cassandra Driver is implemented in Singleton fashion and multiple Flink process functions will interact with it to get data from Cassandra. I also add a future callback to each {{Session.executeAsync}}'s {{ResultSetFuture}}. The app is run on Kubernetes through Docker containers.
 
All dependencies are packaged in a single {{jar}} using {{Bazel}}. Before starting the Flink app, I check all the required classes are in the {{jar}} and are correct and complete. And I use the shaded dependency in order to avoid class loading conflict in JVM. But, when I start and run the Flink app. I keep seeing the following ClassNotFoundException in the Taskmanager logs.

I also notice that these issues are easier to reproduce when giving more resources and parallelism to the Flink app and the process functions. And the issues are most likely happen in the future callback or Promise.
 java.lang.NoClassDefFoundError: com/datastax/driver/core/SessionManager$State
    at com.datastax.driver.core.SessionManager.getState(SessionManager.java:211)
    at io.uhana.cassandra.CassandraDriver.sessionNeedsReconnect(CassandraDriver.java:508)
    at io.uhana.cassandra.CassandraDriver.access$000(CassandraDriver.java:61)
    at io.uhana.cassandra.CassandraDriver$1.onFailure(CassandraDriver.java:518)
    at com.google.common.util.concurrent.Futures$CallbackListener.run(Futures.java:1387)
    at com.google.common.util.concurrent.AbstractFuture.executeListener(AbstractFuture.java:1015)
    at com.google.common.util.concurrent.AbstractFuture.complete(AbstractFuture.java:868)
    at com.google.common.util.concurrent.AbstractFuture.setException(AbstractFuture.java:713)
    at com.datastax.driver.core.DefaultResultSetFuture.onSet(DefaultResultSetFuture.java:230)
    at com.datastax.driver.core.RequestHandler.setFinalResult(RequestHandler.java:235)
    at com.datastax.driver.core.RequestHandler.access$2600(RequestHandler.java:61)
    at com.datastax.driver.core.RequestHandler$SpeculativeExecution.setFinalResult(RequestHandler.java:1011)
    at com.datastax.driver.core.RequestHandler$SpeculativeExecution.onSet(RequestHandler.java:647)
    at com.datastax.driver.core.Connection$Dispatcher.channelRead0(Connection.java:1262)
    at com.datastax.driver.core.Connection$Dispatcher.channelRead0(Connection.java:1180)
    at com.datastax.shaded.netty.channel.SimpleChannelInboundHandler.channelRead(SimpleChannelInboundHandler.java:105)
    at com.datastax.shaded.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:356)
    at com.datastax.shaded.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:342)
    at com.datastax.shaded.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:335)
    at com.datastax.shaded.netty.handler.timeout.IdleStateHandler.channelRead(IdleStateHandler.java:286)
    at com.datastax.shaded.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:356)
    at com.datastax.shaded.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:342)
    at com.datastax.shaded.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:335)
    at com.datastax.shaded.netty.handler.codec.MessageToMessageDecoder.channelRead(MessageToMessageDecoder.java:102)
    at com.datastax.shaded.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:356)
    at com.datastax.shaded.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:342)
    at com.datastax.shaded.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:335)
    at com.datastax.shaded.netty.handler.codec.ByteToMessageDecoder.fireChannelRead(ByteToMessageDecoder.java:312)
    at com.datastax.shaded.netty.handler.codec.ByteToMessageDecoder.channelRead(ByteToMessageDecoder.java:286)
    at com.datastax.shaded.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:356)
    at com.datastax.shaded.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:342)
    at com.datastax.shaded.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:335)
    at com.datastax.shaded.netty.channel.ChannelInboundHandlerAdapter.channelRead(ChannelInboundHandlerAdapter.java:86)
    at com.datastax.driver.core.InboundTrafficMeter.channelRead(InboundTrafficMeter.java:38)
    at com.datastax.shaded.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:356)
    at com.datastax.shaded.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:342)
    at com.datastax.shaded.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:335)
    at com.datastax.shaded.netty.channel.DefaultChannelPipeline$HeadContext.channelRead(DefaultChannelPipeline.java:1304)
    at com.datastax.shaded.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:356)
    at com.datastax.shaded.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:342)
    at com.datastax.shaded.netty.channel.DefaultChannelPipeline.fireChannelRead(DefaultChannelPipeline.java:921)
    at com.datastax.shaded.netty.channel.nio.AbstractNioByteChannel$NioByteUnsafe.read(AbstractNioByteChannel.java:135)
    at com.datastax.shaded.netty.channel.nio.NioEventLoop.processSelectedKey(NioEventLoop.java:646)
    at com.datastax.shaded.netty.channel.nio.NioEventLoop.processSelectedKeysPlain(NioEventLoop.java:546)
    at com.datastax.shaded.netty.channel.nio.NioEventLoop.processSelectedKeys(NioEventLoop.java:500)
    at com.datastax.shaded.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:460)
    at com.datastax.shaded.netty.util.concurrent.SingleThreadEventExecutor$2.run(SingleThreadEventExecutor.java:131)
    at com.datastax.shaded.netty.util.concurrent.FastThreadLocalRunnable.run(FastThreadLocalRunnable.java:30)
    at java.base/java.lang.Thread.run(Thread.java:834)
Caused by: java.lang.ClassNotFoundException: com.datastax.driver.core.SessionManager$State
    at java.base/java.net.URLClassLoader.findClass(URLClassLoader.java:471)
    at java.base/java.lang.ClassLoader.loadClass(ClassLoader.java:588)
    at org.apache.flink.util.ChildFirstClassLoader.loadClass(ChildFirstClassLoader.java:69)
    at java.base/java.lang.ClassLoader.loadClass(ClassLoader.java:521)
    ... 49 more
and
ConstantReconnectionPolicy$ConstantSchedule' [enable DEBUG level for full stacktrace] was thrown by a user handler's exceptionCaught() method while handling the following exception:
java.lang.NoClassDefFoundError: com/datastax/shaded/netty/handler/timeout/IdleState
    at com.datastax.shaded.netty.handler.timeout.IdleStateHandler$ReaderIdleTimeoutTask.run(IdleStateHandler.java:493)
    at com.datastax.shaded.netty.handler.timeout.IdleStateHandler$AbstractIdleTask.run(IdleStateHandler.java:466)
    at com.datastax.shaded.netty.util.concurrent.PromiseTask$RunnableAdapter.call(PromiseTask.java:38)
    at com.datastax.shaded.netty.util.concurrent.ScheduledFutureTask.run(ScheduledFutureTask.java:120)
    at com.datastax.shaded.netty.util.concurrent.SingleThreadEventExecutor.runAllTasks(SingleThreadEventExecutor.java:399)
    at com.datastax.shaded.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:464)
    at com.datastax.shaded.netty.util.concurrent.SingleThreadEventExecutor$2.run(SingleThreadEventExecutor.java:131)
    at com.datastax.shaded.netty.util.concurrent.FastThreadLocalRunnable.run(FastThreadLocalRunnable.java:30)
    at java.base/java.lang.Thread.run(Thread.java:834)
Caused by: java.lang.ClassNotFoundException: com.datastax.shaded.netty.handler.timeout.IdleState
    at java.base/java.net.URLClassLoader.findClass(URLClassLoader.java:471)
    at java.base/java.lang.ClassLoader.loadClass(ClassLoader.java:588)
    at org.apache.flink.util.ChildFirstClassLoader.loadClass(ChildFirstClassLoader.java:69)
    at java.base/java.lang.ClassLoader.loadClass(ClassLoader.java:521)
    ... 9 more {{}}



--
This message was sent by Atlassian Jira
(v8.3.4#803005)