title: Apache flink Confluent org.apache.avro.generic.GenericData$Record cannot be cast to java.lang.String
body: <p>I have a Apache Flink Application, where I want to filter the data by Country which gets read from topic v01 and write the filtered data into the topic v02. For testing purposes I tried to write everything in uppercase.</p>
  <p>My Code:</p>
  <pre><code>package org.example;
  
  import org.apache.avro.Schema;
  import org.apache.flink.api.common.functions.MapFunction;
  import org.apache.flink.formats.avro.registry.confluent.ConfluentRegistryAvroDeserializationSchema;
  import org.apache.flink.formats.avro.registry.confluent.ConfluentRegistryAvroSerializationSchema;
  import org.apache.flink.streaming.api.datastream.DataStream;
  import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
  import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
  import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer;
  import java.util.Properties;
  
  public class KafkaRead {
  
  public static void main(String[] args) throws Exception {
  
      Properties properties = new Properties();
      properties.setProperty(&quot;bootstrap.servers&quot;, &quot;192.168.100.100:9092&quot;);
  
      properties.setProperty(&quot;group.id&quot;, &quot;luft&quot;);
      String schemaRegistryUrl = &quot;http://192.168.100.100:8081&quot;;
  
      String valueSchema = &quot;{\&quot;connect.name\&quot;:\&quot;com.github.jcustenborder.kafka.connect.model.Value\&quot;,\&quot;fields\&quot;:[{\&quot;default\&quot;:null,\&quot;name\&quot;:\&quot;Datum\&quot;,\&quot;type\&quot;:[\&quot;null\&quot;,\&quot;string\&quot;]},{\&quot;default\&quot;:null,\&quot;name\&quot;:\&quot;Country\&quot;,\&quot;type\&quot;:[\&quot;null\&quot;,\&quot;string\&quot;]},{\&quot;default\&quot;:null,\&quot;name\&quot;:\&quot;City\&quot;,\&quot;type\&quot;:[\&quot;null\&quot;,\&quot;string\&quot;]},{\&quot;default\&quot;:null,\&quot;name\&quot;:\&quot;Specie\&quot;,\&quot;type\&quot;:[\&quot;null\&quot;,\&quot;string\&quot;]},{\&quot;default\&quot;:null,\&quot;name\&quot;:\&quot;count\&quot;,\&quot;type\&quot;:[\&quot;null\&quot;,\&quot;string\&quot;]},{\&quot;default\&quot;:null,\&quot;name\&quot;:\&quot;min\&quot;,\&quot;type\&quot;:[\&quot;null\&quot;,\&quot;string\&quot;]},{\&quot;default\&quot;:null,\&quot;name\&quot;:\&quot;max\&quot;,\&quot;type\&quot;:[\&quot;null\&quot;,\&quot;string\&quot;]},{\&quot;default\&quot;:null,\&quot;name\&quot;:\&quot;median\&quot;,\&quot;type\&quot;:[\&quot;null\&quot;,\&quot;string\&quot;]},{\&quot;default\&quot;:null,\&quot;name\&quot;:\&quot;variance\&quot;,\&quot;type\&quot;:[\&quot;null\&quot;,\&quot;string\&quot;]}],\&quot;name\&quot;:\&quot;Value\&quot;,\&quot;namespace\&quot;:\&quot;com.github.jcustenborder.kafka.connect.model\&quot;,\&quot;type\&quot;:\&quot;record\&quot;}&quot;;
      Schema schema = new Schema.Parser().parse(valueSchema);
  
      final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
  
      FlinkKafkaConsumer kafkaConsumer = new FlinkKafkaConsumer&lt;&gt; (&quot;v01&quot;, ConfluentRegistryAvroDeserializationSchema.forGeneric(schema, schemaRegistryUrl) , properties);
      kafkaConsumer.setStartFromEarliest();
      
      DataStream&lt;String&gt; streamIn = env.addSource(kafkaConsumer);
  
      FlinkKafkaProducer kafkaProducer = new FlinkKafkaProducer(&quot;v02&quot;,ConfluentRegistryAvroSerializationSchema.forGeneric(&quot;v02-value&quot;,schema,schemaRegistryUrl),properties);
  
      DataStream&lt;String&gt; streamOut = streamIn.map(new MapFunction&lt;String, String&gt;() { //&lt;--Error in this line
          @Override
          public String map(String value) throws Exception {
              return value.toUpperCase();
          }
      });
      streamOut.addSink(kafkaProducer);
      env.execute(&quot;Flink Streaming In/Out Kafka&quot;);
  }
  </code></pre>
  <p>}</p>
  <p>When executing I get following error:</p>
  <pre><code>Caused by: java.lang.ClassCastException: org.apache.avro.generic.GenericData$Record cannot be cast to java.lang.String
  at org.example.KafkaRead$1.map(KafkaRead.java:45)
  </code></pre>
  <p>It works without problems if I dont use a map function and use my Input as my Output:</p>
  <pre><code>DataStream&lt;String&gt; streamOut = streamIn;
  </code></pre>
  <p>For Context:</p>
  <p>The data that gets read in the first place looks like this</p>
  <pre><code>Date,Country,City,Specie,count,min,max,median,variance
  2020-03-24,DE,Hamburg,humidity,288,26.0,54.0,36.5,966.48
  2020-03-26,DE,Hamburg,humidity,288,25.0,71.5,44.0,1847.14
  2020-04-01,DE,Hamburg,humidity,288,61.0,83.0,75.0,418.07
  </code></pre>
  <p>The csv file are read via the SpoolDirCsvSourceConnector in Kafka. Apache Flink reads from topic v01, which gets an automatic generated schema via the connector, which is saved as avro. In the next step I want to analyse it with filters inside flink and after that I want to write back to v02. For example filtering via country.</p>
  <p>Complete Error:</p>
  <pre><code>    Exception in thread &quot;main&quot; org.apache.flink.runtime.client.JobExecutionException: Job execution failed.
      at org.apache.flink.runtime.jobmaster.JobResult.toJobExecutionResult(JobResult.java:147)
      at org.apache.flink.runtime.minicluster.MiniClusterJobClient.lambda$getJobExecutionResult$2(MiniClusterJobClient.java:119)
      at java.util.concurrent.CompletableFuture.uniApply(CompletableFuture.java:616)
      at java.util.concurrent.CompletableFuture$UniApply.tryFire(CompletableFuture.java:591)
      at java.util.concurrent.CompletableFuture.postComplete(CompletableFuture.java:488)
      at java.util.concurrent.CompletableFuture.complete(CompletableFuture.java:1975)
      at org.apache.flink.runtime.rpc.akka.AkkaInvocationHandler.lambda$invokeRpc$0(AkkaInvocationHandler.java:229)
      at java.util.concurrent.CompletableFuture.uniWhenComplete(CompletableFuture.java:774)
      at java.util.concurrent.CompletableFuture$UniWhenComplete.tryFire(CompletableFuture.java:750)
      at java.util.concurrent.CompletableFuture.postComplete(CompletableFuture.java:488)
      at java.util.concurrent.CompletableFuture.complete(CompletableFuture.java:1975)
      at org.apache.flink.runtime.concurrent.FutureUtils$1.onComplete(FutureUtils.java:996)
      at akka.dispatch.OnComplete.internal(Future.scala:264)
      at akka.dispatch.OnComplete.internal(Future.scala:261)
      at akka.dispatch.japi$CallbackBridge.apply(Future.scala:191)
      at akka.dispatch.japi$CallbackBridge.apply(Future.scala:188)
      at scala.concurrent.impl.CallbackRunnable.run(Promise.scala:36)
      at org.apache.flink.runtime.concurrent.Executors$DirectExecutionContext.execute(Executors.java:74)
      at scala.concurrent.impl.CallbackRunnable.executeWithValue(Promise.scala:44)
      at scala.concurrent.impl.Promise$DefaultPromise.tryComplete(Promise.scala:252)
      at akka.pattern.PromiseActorRef.$bang(AskSupport.scala:572)
      at akka.pattern.PipeToSupport$PipeableFuture$$anonfun$pipeTo$1.applyOrElse(PipeToSupport.scala:22)
      at akka.pattern.PipeToSupport$PipeableFuture$$anonfun$pipeTo$1.applyOrElse(PipeToSupport.scala:21)
      at scala.concurrent.Future$$anonfun$andThen$1.apply(Future.scala:436)
      at scala.concurrent.Future$$anonfun$andThen$1.apply(Future.scala:435)
      at scala.concurrent.impl.CallbackRunnable.run(Promise.scala:36)
      at akka.dispatch.BatchingExecutor$AbstractBatch.processBatch(BatchingExecutor.scala:55)
      at akka.dispatch.BatchingExecutor$BlockableBatch$$anonfun$run$1.apply$mcV$sp(BatchingExecutor.scala:91)
      at akka.dispatch.BatchingExecutor$BlockableBatch$$anonfun$run$1.apply(BatchingExecutor.scala:91)
      at akka.dispatch.BatchingExecutor$BlockableBatch$$anonfun$run$1.apply(BatchingExecutor.scala:91)
      at scala.concurrent.BlockContext$.withBlockContext(BlockContext.scala:72)
      at akka.dispatch.BatchingExecutor$BlockableBatch.run(BatchingExecutor.scala:90)
      at akka.dispatch.TaskInvocation.run(AbstractDispatcher.scala:40)
      at akka.dispatch.ForkJoinExecutorConfigurator$AkkaForkJoinTask.exec(ForkJoinExecutorConfigurator.scala:44)
      at akka.dispatch.forkjoin.ForkJoinTask.doExec(ForkJoinTask.java:260)
      at akka.dispatch.forkjoin.ForkJoinPool$WorkQueue.runTask(ForkJoinPool.java:1339)
      at akka.dispatch.forkjoin.ForkJoinPool.runWorker(ForkJoinPool.java:1979)
      at akka.dispatch.forkjoin.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:107)
  Caused by: org.apache.flink.runtime.JobException: Recovery is suppressed by NoRestartBackoffTimeStrategy
      at org.apache.flink.runtime.executiongraph.failover.flip1.ExecutionFailureHandler.handleFailure(ExecutionFailureHandler.java:116)
      at org.apache.flink.runtime.executiongraph.failover.flip1.ExecutionFailureHandler.getFailureHandlingResult(ExecutionFailureHandler.java:78)
      at org.apache.flink.runtime.scheduler.DefaultScheduler.handleTaskFailure(DefaultScheduler.java:224)
      at org.apache.flink.runtime.scheduler.DefaultScheduler.maybeHandleTaskFailure(DefaultScheduler.java:217)
      at org.apache.flink.runtime.scheduler.DefaultScheduler.updateTaskExecutionStateInternal(DefaultScheduler.java:208)
      at org.apache.flink.runtime.scheduler.SchedulerBase.updateTaskExecutionState(SchedulerBase.java:610)
      at org.apache.flink.runtime.scheduler.SchedulerNG.updateTaskExecutionState(SchedulerNG.java:89)
      at org.apache.flink.runtime.jobmaster.JobMaster.updateTaskExecutionState(JobMaster.java:419)
      at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
      at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
      at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
      at java.lang.reflect.Method.invoke(Method.java:498)
      at org.apache.flink.runtime.rpc.akka.AkkaRpcActor.handleRpcInvocation(AkkaRpcActor.java:286)
      at org.apache.flink.runtime.rpc.akka.AkkaRpcActor.handleRpcMessage(AkkaRpcActor.java:201)
      at org.apache.flink.runtime.rpc.akka.FencedAkkaRpcActor.handleRpcMessage(FencedAkkaRpcActor.java:74)
      at org.apache.flink.runtime.rpc.akka.AkkaRpcActor.handleMessage(AkkaRpcActor.java:154)
      at akka.japi.pf.UnitCaseStatement.apply(CaseStatements.scala:26)
      at akka.japi.pf.UnitCaseStatement.apply(CaseStatements.scala:21)
      at scala.PartialFunction$class.applyOrElse(PartialFunction.scala:123)
      at akka.japi.pf.UnitCaseStatement.applyOrElse(CaseStatements.scala:21)
      at scala.PartialFunction$OrElse.applyOrElse(PartialFunction.scala:170)
      at scala.PartialFunction$OrElse.applyOrElse(PartialFunction.scala:171)
      at scala.PartialFunction$OrElse.applyOrElse(PartialFunction.scala:171)
      at akka.actor.Actor$class.aroundReceive(Actor.scala:517)
      at akka.actor.AbstractActor.aroundReceive(AbstractActor.scala:225)
      at akka.actor.ActorCell.receiveMessage(ActorCell.scala:592)
      at akka.actor.ActorCell.invoke(ActorCell.scala:561)
      at akka.dispatch.Mailbox.processMailbox(Mailbox.scala:258)
      at akka.dispatch.Mailbox.run(Mailbox.scala:225)
      at akka.dispatch.Mailbox.exec(Mailbox.scala:235)
      ... 4 more
  Caused by: java.lang.ClassCastException: org.apache.avro.generic.GenericData$Record cannot be cast to java.lang.String
      at org.example.KafkaRead$1.map(KafkaRead.java:45)
      at org.apache.flink.streaming.api.operators.StreamMap.processElement(StreamMap.java:41)
      at org.apache.flink.streaming.runtime.tasks.CopyingChainingOutput.pushToOperator(CopyingChainingOutput.java:71)
      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)
  </code></pre>