title: Using own jackson versions in Flink causes VerifyError
body: <p>I'm using Apache Flink (v1.11) with Scala and added an own DeserializationSchema for Kafka connector. Therefore i would like to use my own packages and versions of jackson (v2.12.0).</p>
  <p>But i got the following error:</p>
  <pre><code>Exception in thread &quot;main&quot; java.lang.VerifyError: Cannot inherit from final class
      at java.lang.ClassLoader.defineClass1(Native Method)
      at java.lang.ClassLoader.defineClass(ClassLoader.java:756)
      at java.security.SecureClassLoader.defineClass(SecureClassLoader.java:142)
      at java.net.URLClassLoader.defineClass(URLClassLoader.java:468)
      at java.net.URLClassLoader.access$100(URLClassLoader.java:74)
      at java.net.URLClassLoader$1.run(URLClassLoader.java:369)
      at java.net.URLClassLoader$1.run(URLClassLoader.java:363)
      at java.security.AccessController.doPrivileged(Native Method)
      at java.net.URLClassLoader.findClass(URLClassLoader.java:362)
      at java.lang.ClassLoader.loadClass(ClassLoader.java:418)
      at sun.misc.Launcher$AppClassLoader.loadClass(Launcher.java:352)
      at java.lang.ClassLoader.loadClass(ClassLoader.java:351)
      at com.fasterxml.jackson.dataformat.csv.CsvMapper.&lt;init&gt;(CsvMapper.java:108)
      at de.integration_factory.datastream.types.CovidEventSchema.&lt;init&gt;(CovidEventSchema.scala:14)
      at de.integration_factory.datastream.Aggregate_Datastream$.main(Aggregate_Datastream.scala:34)
      at de.integration_factory.datastream.Aggregate_Datastream.main(Aggregate_Datastream.scala)
  </code></pre>
  <p>This is my EventSchema:</p>
  <pre><code>import com.fasterxml.jackson.dataformat.csv.CsvMapper
  import com.fasterxml.jackson.datatype.joda.JodaModule
  import org.apache.flink.api.common.serialization.{DeserializationSchema, SerializationSchema}
  import org.apache.flink.api.common.typeinfo.TypeInformation
  
  @SerialVersionUID(6154188370181669758L)
  class CovidEventSchema extends DeserializationSchema[CovidEvent] with SerializationSchema[CovidEvent] {
  
    private val mapper = new CsvMapper
    mapper.registerModule(new JodaModule)
  
    val csvSchema = mapper
      .schemaFor(classOf[CovidEvent])
      .withLineSeparator(&quot;,&quot;)
      .withoutHeader()
    val  reader = mapper.readerWithSchemaFor(classOf[CovidEvent])
  
    def serialize(event: CovidEvent): Array[Byte] = mapper.writer(csvSchema).writeValueAsBytes()
  
    @throws[IOException]
    def deserialize(message: Array[Byte]): CovidEvent = reader.readValue[CovidEvent](message)
  
  
    def isEndOfStream(nextElement: CovidEvent) = false
  
    def getProducedType: TypeInformation[CovidEvent] = TypeInformation.of(classOf[CovidEvent])
  }
  </code></pre>
  <p>This is my PoJo for schema:</p>
  <pre><code>import com.fasterxml.jackson.annotation.JsonFormat;
  import lombok.AllArgsConstructor;
  import lombok.Data;
  import lombok.NoArgsConstructor;
  import org.joda.time.DateTime;
  
  
  @Data
  @NoArgsConstructor
  @AllArgsConstructor
  public class CovidEvent {
  
      private long objectId;
      private int bundeslandId;
      private String bundesland;
      private String landkreis;
      private String altersgruppe;
      private String geschlecht;
      private int anzahlFall;
      private int anzahlTodesfall;
      @JsonFormat(shape = JsonFormat.Shape.STRING, pattern = &quot;yyyy-MM-dd HH:mm:ss&quot;, timezone = &quot;UTC&quot;)
      private DateTime meldedatum;
      private int landkreisId;
      private String datenstand;
      private int neuerFall;
      private int neuerTodesfall;
      private String refDatum;
      private int neuGenesen;
      private int anzahlGenesen;
      @JsonFormat(shape = JsonFormat.Shape.NUMBER)
      private boolean istErkrankungsbeginn;
      private String altersGruppe2;
  
      public long getEventtime() {
          return meldedatum.getMillis();
      }
  
  }
  </code></pre>
  <p>After some research <a href="https://stackoverflow.com/questions/7766614/jackson-tomcat-java-lang-verifyerror-cannot-inherit-from-final-class">I found out</a> that the error is probably caused by different Jackson versions in the classpath.</p>
  <p>I thought it would be possible to use own version of Jackson, because Flink shaded the own versions.</p>
  <p>What am I doing wrong?</p>
  <p><strong>UPDATE</strong>: If i import the jackson classes from shaded flink package it is working</p>
  <pre><code>org.apache.flink.shaded.jackson2.com.fasterxml.jackson.dataformat.csv.CsvMapper
  </code></pre>
  <p>But so i am dependent on the flink shaded jackson version.</p>
  <p><strong>UPDATE</strong>: So a better implementation by using <code>open</code> is something like this ?</p>
  <pre><code>class CovidEventSchema extends DeserializationSchema[CovidEvent] with SerializationSchema[CovidEvent] {
  
     private  var  reader: ObjectReader = null
  
    private var writer: ObjectWriter = null
  
    override def open(context: SerializationSchema.InitializationContext): Unit = {
  
       val mapper  = new CsvMapper()
  
       val csvSchema = mapper
        .schemaFor(classOf[CovidEvent])
        .withLineSeparator(&quot;,&quot;)
        .withoutHeader()
  
      this.reader = mapper.readerFor(classOf[CovidEvent]).`with`(csvSchema)
      this.writer = mapper.writer(csvSchema)
      super.open(context)
    }
  }
  </code></pre>