Hello, flink experts and friends!
It is my first time to write flink application in my company. But I met the following error when I used a elasticsearch as my sink. I searched the solution for it and found a jira
https://issues.apache.org/jira/browse/FLINK-7133 . then, I added the PR to my code like the following. But when I ran the flink program again, error is still there. Why ? When I used a filesystem as a sink ,nothing error happened, but when i returned to elasticsearch, error loves me. bad! Could you help me please ?
I think the two lines of code of Log is not relevant to the error. But I still put it here as a reference. It is just a PB object.
21:41:09,397 INFO org.apache.flink.api.java.typeutils.TypeExtractor - class com.vip.data.cleaning.logic.mars.activity.info.ActivityInfoProtos$ActivityInfo does not contain a setter for field unknownFields
21:41:09,400 INFO org.apache.flink.api.java.typeutils.TypeExtractor - class com.vip.data.cleaning.logic.mars.activity.info.ActivityInfoProtos$ActivityInfo is not a valid POJO type because not all fields are valid POJO fields.
Exception in thread "main" java.lang.IllegalArgumentException
at org.apache.flink.shaded.org.objectweb.asm.ClassVisitor.<init>(Unknown Source)
at org.apache.flink.shaded.org.objectweb.asm.ClassVisitor.<init>(Unknown Source)
at org.apache.flink.api.scala.InnerClosureFinder.<init>(ClosureCleaner.scala:279)
at org.apache.flink.api.scala.ClosureCleaner$.getInnerClasses(ClosureCleaner.scala:95)
at org.apache.flink.api.scala.ClosureCleaner$.clean(ClosureCleaner.scala:115)
at org.apache.flink.streaming.api.scala.StreamExecutionEnvironment.scalaClean(StreamExecutionEnvironment.scala:670)
at org.apache.flink.streaming.api.scala.StreamExecutionEnvironment.addSource(StreamExecutionEnvironment.scala:600)
at com.vip.data.cleaning.validation.mars.activity.info.WhileListFilter$.main(WhileListFilter.scala:79)
at com.vip.data.cleaning.validation.mars.activity.info.WhileListFilter.main(WhileListFilter.scala)
PR here:
<dependency>
<groupId>org.elasticsearch</groupId>
<artifactId>elasticsearch</artifactId>
<version>${elasticsearch.version}</version>
<exclusions>
<exclusion>
<groupId>org.ow2.asm</groupId>
<artifactId>*</artifactId>
</exclusion>
</exclusions>
</dependency>
Regards
mingleizhang