[FLINK-30377] CommonExecSink does not use ClientWrapperClassLoader while extracting Type from KeyedStream

• Type: Bug
• Status: Resolved
• Priority: Major
• Resolution: Fixed
• Affects Version/s: 1.16.0
• Fix Version/s: 1.17.0, 1.16.1
• Component/s: Table SQL / Planner
• Labels: pull-request-available
• Assignee: Samrat Deb
• Reporter: Prabhu Joseph
• Created: 12/Dec/22 11:35
• Updated: 27/Jan/23 14:53
• Resolved: 27/Jan/23 14:53

Description

CommonExecSink does not use ClientWrapperClassLoader while extracting Type from KeyedStream. This will lead to ClassNotFoundException on user classes added through add jar command. This is working fine on Flink 1.15.

 

Caused by: java.lang.ClassNotFoundException: org.apache.hudi.common.model.HoodieRecord
 at java.net.URLClassLoader.findClass(URLClassLoader.java:387) ~[?:1.8.0_352]
 at java.lang.ClassLoader.loadClass(ClassLoader.java:418) ~[?:1.8.0_352]
 at sun.misc.Launcher$AppClassLoader.loadClass(Launcher.java:352) ~[?:1.8.0_352]
 at java.lang.ClassLoader.loadClass(ClassLoader.java:351) ~[?:1.8.0_352]
 at java.lang.Class.forName0(Native Method) ~[?:1.8.0_352]
 at java.lang.Class.forName(Class.java:348) ~[?:1.8.0_352]
 at org.apache.flink.api.java.typeutils.TypeExtractionUtils.checkAndExtractLambda(TypeExtractionUtils.java:143) ~[flink-dist-1.16.0.jar:1.16.0]
 at org.apache.flink.api.java.typeutils.TypeExtractor.getUnaryOperatorReturnType(TypeExtractor.java:539) ~[flink-dist-1.16.0.jar:1.16.0]
 at org.apache.flink.api.java.typeutils.TypeExtractor.getKeySelectorTypes(TypeExtractor.java:415) ~[flink-dist-1.16.0.jar:1.16.0]
 at org.apache.flink.api.java.typeutils.TypeExtractor.getKeySelectorTypes(TypeExtractor.java:406) ~[flink-dist-1.16.0.jar:1.16.0]
 at org.apache.flink.streaming.api.datastream.KeyedStream.<init>(KeyedStream.java:116) ~[flink-dist-1.16.0.jar:1.16.0]
 at org.apache.flink.streaming.api.datastream.DataStream.keyBy(DataStream.java:300) ~[flink-dist-1.16.0.jar:1.16.0]
 at org.apache.hudi.sink.utils.Pipelines.hoodieStreamWrite(Pipelines.java:339) ~[?:?]
 at org.apache.hudi.table.HoodieTableSink.lambda$getSinkRuntimeProvider$0(HoodieTableSink.java:104) ~[?:?]
 at org.apache.hudi.adapter.DataStreamSinkProviderAdapter.consumeDataStream(DataStreamSinkProviderAdapter.java:35) ~[?:?]
 at org.apache.flink.table.planner.plan.nodes.exec.common.CommonExecSink.applySinkProvider(CommonExecSink.java:483) ~[?:?]
 at org.apache.flink.table.planner.plan.nodes.exec.common.CommonExecSink.createSinkTransformation(CommonExecSink.java:203) ~[?:?]
 at org.apache.flink.table.planner.plan.nodes.exec.stream.StreamExecSink.translateToPlanInternal(StreamExecSink.java:176) ~[?:?]
 at org.apache.flink.table.planner.plan.nodes.exec.ExecNodeBase.translateToPlan(ExecNodeBase.java:158) ~[?:?]
 at org.apache.flink.table.planner.delegation.StreamPlanner.$anonfun$translateToPlan$1(StreamPlanner.scala:85) ~[?:?]
 at scala.collection.TraversableLike.$anonfun$map$1(TraversableLike.scala:233) ~[flink-scala_2.12-1.16.0.jar:1.16.0]
 at scala.collection.Iterator.foreach(Iterator.scala:937) ~[flink-scala_2.12-1.16.0.jar:1.16.0]
 at scala.collection.Iterator.foreach$(Iterator.scala:937) ~[flink-scala_2.12-1.16.0.jar:1.16.0]
 at scala.collection.AbstractIterator.foreach(Iterator.scala:1425) ~[flink-scala_2.12-1.16.0.jar:1.16.0]
 at scala.collection.IterableLike.foreach(IterableLike.scala:70) ~[flink-scala_2.12-1.16.0.jar:1.16.0]
 at scala.collection.IterableLike.foreach$(IterableLike.scala:69) ~[flink-scala_2.12-1.16.0.jar:1.16.0]
 at scala.collection.AbstractIterable.foreach(Iterable.scala:54) ~[flink-scala_2.12-1.16.0.jar:1.16.0]
 at scala.collection.TraversableLike.map(TraversableLike.scala:233) ~[flink-scala_2.12-1.16.0.jar:1.16.0]
 at scala.collection.TraversableLike.map$(TraversableLike.scala:226) ~[flink-scala_2.12-1.16.0.jar:1.16.0]
 at scala.collection.AbstractTraversable.map(Traversable.scala:104) ~[flink-scala_2.12-1.16.0.jar:1.16.0]
 at org.apache.flink.table.planner.delegation.StreamPlanner.translateToPlan(StreamPlanner.scala:84) ~[?:?]
 at org.apache.flink.table.planner.delegation.PlannerBase.translate(PlannerBase.scala:197) ~[?:?]
 at org.apache.flink.table.api.internal.TableEnvironmentImpl.translate(TableEnvironmentImpl.java:1733) ~[flink-table-api-java-uber-1.16.0.jar:1.16.0]
 at org.apache.flink.table.api.internal.TableEnvironmentImpl.executeInternal(TableEnvironmentImpl.java:825) ~[flink-table-api-java-uber-1.16.0.jar:1.16.0]
 at org.apache.flink.table.client.gateway.local.LocalExecutor.executeModifyOperations(LocalExecutor.java:219) ~[flink-sql-client-1.16.0.jar:1.16.0]

 

 

 

Issue Links

• links to: GitHub Pull Request #21497
• links to: GitHub Pull Request #21766