title: Flink throwing com.esotericsoftware.kryo.KryoException: java.lang.UnsupportedOperationException
body: <p>I am trying to deserialize kafka events in my flink stream job. This is my code:</p>
  <pre><code>...
  case class URLResponse (status: Int, domain: String, url: String, queue: String, html: String)
  ...
  val schema: Schema = AvroSchema[URLResponse]
  ...
  val stream = env.addSource(new FlinkKafkaConsumer[GenericRecord](kafkaTopic, ConfluentRegistryAvroDeserializationSchema.forGeneric(schema, schemaRegistryURL), properties))
  </code></pre>
  <p>And the job throwing this exception during the runtime:</p>
  <pre><code>...
  Caused by: com.esotericsoftware.kryo.KryoException: java.lang.UnsupportedOperationException
  Serialization trace:
  reserved (org.apache.avro.Schema$Field)
  fieldMap (org.apache.avro.Schema$RecordSchema)
  schema (org.apache.avro.generic.GenericData$Record)
      at com.esotericsoftware.kryo.serializers.ObjectField.read(ObjectField.java:125)
      at com.esotericsoftware.kryo.serializers.FieldSerializer.read(FieldSerializer.java:528)
      at com.esotericsoftware.kryo.Kryo.readClassAndObject(Kryo.java:761)
      at com.esotericsoftware.kryo.serializers.MapSerializer.read(MapSerializer.java:143)
      at com.esotericsoftware.kryo.serializers.MapSerializer.read(MapSerializer.java:21)
      at com.esotericsoftware.kryo.Kryo.readObject(Kryo.java:679)
      at com.esotericsoftware.kryo.serializers.ObjectField.read(ObjectField.java:106)
      at com.esotericsoftware.kryo.serializers.FieldSerializer.read(FieldSerializer.java:528)
      at com.esotericsoftware.kryo.Kryo.readObject(Kryo.java:679)
      at com.esotericsoftware.kryo.serializers.ObjectField.read(ObjectField.java:106)
      at com.esotericsoftware.kryo.serializers.FieldSerializer.read(FieldSerializer.java:528)
      at com.esotericsoftware.kryo.Kryo.readObject(Kryo.java:657)
      at org.apache.flink.api.java.typeutils.runtime.kryo.KryoSerializer.copy(KryoSerializer.java:262)
      at org.apache.flink.streaming.runtime.tasks.CopyingChainingOutput.pushToOperator(CopyingChainingOutput.java:69)
      at org.apache.flink.streaming.runtime.tasks.CopyingChainingOutput.collect(CopyingChainingOutput.java:46)
      at org.apache.flink.streaming.runtime.tasks.CopyingChainingOutput.collect(CopyingChainingOutput.java:26)
      at org.apache.flink.streaming.api.operators.CountingOutput.collect(CountingOutput.java:52)
      at org.apache.flink.streaming.api.operators.CountingOutput.collect(CountingOutput.java:30)
      at org.apache.flink.streaming.api.operators.StreamSourceContexts$ManualWatermarkContext.processAndCollectWithTimestamp(StreamSourceContexts.java:310)
      at org.apache.flink.streaming.api.operators.StreamSourceContexts$WatermarkContext.collectWithTimestamp(StreamSourceContexts.java:409)
      at org.apache.flink.streaming.connectors.kafka.internals.AbstractFetcher.emitRecordsWithTimestamps(AbstractFetcher.java:352)
      at org.apache.flink.streaming.connectors.kafka.internals.KafkaFetcher.partitionConsumerRecordsHandler(KafkaFetcher.java:181)
      at org.apache.flink.streaming.connectors.kafka.internals.KafkaFetcher.runFetchLoop(KafkaFetcher.java:137)
      at org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumerBase.run(FlinkKafkaConsumerBase.java:761)
      at org.apache.flink.streaming.api.operators.StreamSource.run(StreamSource.java:100)
      at org.apache.flink.streaming.api.operators.StreamSource.run(StreamSource.java:63)
      at org.apache.flink.streaming.runtime.tasks.SourceStreamTask$LegacySourceFunctionThread.run(SourceStreamTask.java:215)
  Caused by: java.lang.UnsupportedOperationException
      at java.util.Collections$UnmodifiableCollection.add(Collections.java:1057)
      at com.esotericsoftware.kryo.serializers.CollectionSerializer.read(CollectionSerializer.java:109)
      at com.esotericsoftware.kryo.serializers.CollectionSerializer.read(CollectionSerializer.java:22)
      at com.esotericsoftware.kryo.Kryo.readObject(Kryo.java:679)
      at com.esotericsoftware.kryo.serializers.ObjectField.read(ObjectField.java:106)
      ... 26 more
  
  Process finished with exit code 1
  </code></pre>
  <p>I read that I should not use <code>Kryo</code> but I have no idea how to do it. I tried:</p>
  <pre><code>executionConfig.enableForceAvro()
  executionConfig.disableForceKryo()
  </code></pre>
  <p>but it doesn't help.</p>