You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by "Youjun Yuan (Jira)" <ji...@apache.org> on 2025/04/15 22:40:00 UTC
[jira] [Created] (FLINK-37684) ClassCastException: cannot assign instance of java.lang.invoke.SerializedLambda to field org.apache.flink.streaming.runtime.partitioner.KeyGroupStreamPartitioner.keySelector
Youjun Yuan created FLINK-37684:
-----------------------------------
Summary: ClassCastException: cannot assign instance of java.lang.invoke.SerializedLambda to field org.apache.flink.streaming.runtime.partitioner.KeyGroupStreamPartitioner.keySelector
Key: FLINK-37684
URL: https://issues.apache.org/jira/browse/FLINK-37684
Project: Flink
Issue Type: Bug
Components: API / DataStream
Affects Versions: 1.14.4
Reporter: Youjun Yuan
Flink job failed to run (in k8s) with below error.
The job just has 4 operators, see below. One thing interesting, is that if I remove the second one (id=74, which is just a map operator), then it's fine.
Topo:
{code:java}
{
"nodes" : [ {
"id" : 73,
"type" : "Source: click-tracking-record-kafka_source",
"pact" : "Data Source",
"contents" : "Source: click-tracking-record-kafka_source",
"parallelism" : 1
}, {
"id" : 74,
"type" : "Proto-Deserialization",
"pact" : "Operator",
"contents" : "Proto-Deserialization",
"parallelism" : 1,
"predecessors" : [ {
"id" : 73,
"ship_strategy" : "FORWARD",
"side" : "second"
} ]
}, {
"id" : 76,
"type" : "KeyedProcess",
"pact" : "Operator",
"contents" : "KeyedProcess",
"parallelism" : 1,
"predecessors" : [ {
"id" : 74,
"ship_strategy" : "HASH",
"side" : "second"
} ]
}, {
"id" : 79,
"type" : "Sink uf-aggregate_-_to_kafka_-_click-tracking-record_-_as_-_int-dj-live-user-features-dev03",
"pact" : "Operator",
"contents" : "Sink uf-aggregate_-_to_kafka_-_click-tracking-record_-_as_-_int-dj-live-user-features-dev03",
"parallelism" : 1,
"predecessors" : [ {
"id" : 76,
"ship_strategy" : "FORWARD",
"side" : "second"
} ]
} ]
}{code}
Full call stack:
{code:java}
Caused by: java.lang.ClassCastException: cannot assign instance of java.lang.invoke.SerializedLambda to field org.apache.flink.streaming.runtime.partitioner.KeyGroupStreamPartitioner.keySelector of type org.apache.flink.api.java.functions.KeySelector in instance of org.apache.flink.streaming.runtime.partitioner.KeyGroupStreamPartitioner
at java.io.ObjectStreamClass$FieldReflector.setObjFieldValues(ObjectStreamClass.java:2301) ~[?:1.8.0_322]
at java.io.ObjectStreamClass.setObjFieldValues(ObjectStreamClass.java:1431) ~[?:1.8.0_322]
at java.io.ObjectInputStream.defaultReadFields(ObjectInputStream.java:2437) ~[?:1.8.0_322]
at java.io.ObjectInputStream.readSerialData(ObjectInputStream.java:2355) ~[?:1.8.0_322]
at java.io.ObjectInputStream.readOrdinaryObject(ObjectInputStream.java:2213) ~[?:1.8.0_322]
at java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1669) ~[?:1.8.0_322]
at java.io.ObjectInputStream.defaultReadFields(ObjectInputStream.java:2431) ~[?:1.8.0_322]
at java.io.ObjectInputStream.readSerialData(ObjectInputStream.java:2355) ~[?:1.8.0_322]
at java.io.ObjectInputStream.readOrdinaryObject(ObjectInputStream.java:2213) ~[?:1.8.0_322]
at java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1669) ~[?:1.8.0_322]
at java.io.ObjectInputStream.readObject(ObjectInputStream.java:503) ~[?:1.8.0_322]
at java.io.ObjectInputStream.readObject(ObjectInputStream.java:461) ~[?:1.8.0_322]
at java.util.ArrayList.readObject(ArrayList.java:799) ~[?:1.8.0_322]
at sun.reflect.GeneratedMethodAccessor24.invoke(Unknown Source) ~[?:?]
at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43) ~[?:1.8.0_322]
at java.lang.reflect.Method.invoke(Method.java:498) ~[?:1.8.0_322]
at java.io.ObjectStreamClass.invokeReadObject(ObjectStreamClass.java:1184) ~[?:1.8.0_322]
at java.io.ObjectInputStream.readSerialData(ObjectInputStream.java:2322) ~[?:1.8.0_322]
at java.io.ObjectInputStream.readOrdinaryObject(ObjectInputStream.java:2213) ~[?:1.8.0_322]
at java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1669) ~[?:1.8.0_322]
at java.io.ObjectInputStream.readObject(ObjectInputStream.java:503) ~[?:1.8.0_322]
at java.io.ObjectInputStream.readObject(ObjectInputStream.java:461) ~[?:1.8.0_322]
at org.apache.flink.util.InstantiationUtil.deserializeObject(InstantiationUtil.java:617) ~[flink-dist_2.12-1.14.4.jar:1.14.4]
at org.apache.flink.util.InstantiationUtil.deserializeObject(InstantiationUtil.java:602) ~[flink-dist_2.12-1.14.4.jar:1.14.4]
at org.apache.flink.util.InstantiationUtil.deserializeObject(InstantiationUtil.java:589) ~[flink-dist_2.12-1.14.4.jar:1.14.4]
at org.apache.flink.util.InstantiationUtil.readObjectFromConfig(InstantiationUtil.java:543) ~[flink-dist_2.12-1.14.4.jar:1.14.4]
at org.apache.flink.streaming.api.graph.StreamConfig.getNonChainedOutputs(StreamConfig.java:387) ~[flink-dist_2.12-1.14.4.jar:1.14.4]
at org.apache.flink.streaming.api.graph.StreamConfig.toString(StreamConfig.java:714) ~[flink-dist_2.12-1.14.4.jar:1.14.4]
at java.lang.String.valueOf(String.java:2994) ~[?:1.8.0_322]
at java.lang.StringBuilder.append(StringBuilder.java:136) ~[?:1.8.0_322]
at java.util.AbstractMap.toString(AbstractMap.java:559) ~[?:1.8.0_322]
at java.io.ObjectOutputStream.writeOrdinaryObject(ObjectOutputStream.java:1421) ~[?:1.8.0_322]
at java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1178) ~[?:1.8.0_322]
at java.io.ObjectOutputStream.writeObject(ObjectOutputStream.java:348) ~[?:1.8.0_322]
at org.apache.flink.util.InstantiationUtil.serializeObject(InstantiationUtil.java:632) ~[flink-dist_2.12-1.14.4.jar:1.14.4]
at org.apache.flink.util.InstantiationUtil.writeObjectToConfig(InstantiationUtil.java:548) ~[flink-dist_2.12-1.14.4.jar:1.14.4]
at org.apache.flink.streaming.api.graph.StreamConfig.setTransitiveChainedTaskConfigs(StreamConfig.java:495) ~[flink-dist_2.12-1.14.4.jar:1.14.4]
at org.apache.flink.streaming.api.graph.StreamingJobGraphGenerator.createChain(StreamingJobGraphGenerator.java:479) ~[flink-dist_2.12-1.14.4.jar:1.14.4]
at org.apache.flink.streaming.api.graph.StreamingJobGraphGenerator.setChaining(StreamingJobGraphGenerator.java:377) ~[flink-dist_2.12-1.14.4.jar:1.14.4]
at org.apache.flink.streaming.api.graph.StreamingJobGraphGenerator.createJobGraph(StreamingJobGraphGenerator.java:178) ~[flink-dist_2.12-1.14.4.jar:1.14.4]
at org.apache.flink.streaming.api.graph.StreamingJobGraphGenerator.createJobGraph(StreamingJobGraphGenerator.java:116) ~[flink-dist_2.12-1.14.4.jar:1.14.4]
at org.apache.flink.streaming.api.graph.StreamGraph.getJobGraph(StreamGraph.java:985) ~[flink-dist_2.12-1.14.4.jar:1.14.4]
at org.apache.flink.client.StreamGraphTranslator.translateToJobGraph(StreamGraphTranslator.java:50) ~[flink-dist_2.12-1.14.4.jar:1.14.4]
at org.apache.flink.client.FlinkPipelineTranslationUtil.getJobGraph(FlinkPipelineTranslationUtil.java:39) ~[flink-dist_2.12-1.14.4.jar:1.14.4]
at org.apache.flink.client.deployment.executors.PipelineExecutorUtils.getJobGraph(PipelineExecutorUtils.java:56) ~[flink-dist_2.12-1.14.4.jar:1.14.4]
at org.apache.flink.client.deployment.application.executors.EmbeddedExecutor.submitAndGetJobClientFuture(EmbeddedExecutor.java:122) ~[flink-dist_2.12-1.14.4.jar:1.14.4]
at org.apache.flink.client.deployment.application.executors.EmbeddedExecutor.execute(EmbeddedExecutor.java:104) ~[flink-dist_2.12-1.14.4.jar:1.14.4]
at org.apache.flink.streaming.api.environment.StreamExecutionEnvironment.executeAsync(StreamExecutionEnvironment.java:2042) ~[flink-dist_2.12-1.14.4.jar:1.14.4]
at org.apache.flink.client.program.StreamContextEnvironment.executeAsync(StreamContextEnvironment.java:137) ~[flink-dist_2.12-1.14.4.jar:1.14.4]
at org.apache.flink.client.program.StreamContextEnvironment.execute(StreamContextEnvironment.java:76) ~[flink-dist_2.12-1.14.4.jar:1.14.4]
at org.apache.flink.streaming.api.environment.StreamExecutionEnvironment.execute(StreamExecutionEnvironment.java:1916) ~[flink-dist_2.12-1.14.4.jar:1.14.4]
at com.thetradedesk.streaming.services.databahnjunction.DatabahnJunction.main(DatabahnJunction.java:56) ~[?:?]
at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method) ~[?:1.8.0_322]
at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62) ~[?:1.8.0_322]
at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43) ~[?:1.8.0_322]
at java.lang.reflect.Method.invoke(Method.java:498) ~[?:1.8.0_322]
at org.apache.flink.client.program.PackagedProgram.callMainMethod(PackagedProgram.java:355) {code}
--
This message was sent by Atlassian Jira
(v8.20.10#820010)