title: Not able to run flink application which deserializes avro data coming from a kafka topic
body: <p>Trying to read avro data from a kafka topic using a flink application. Getting the below error while running the flink app. This is my first time working with flink/kafka, and couldn't fix this from days.</p>
  <blockquote>
  <p>org.apache.flink.runtime.client.JobInitializationException: Could not start the JobMaster.
  at org.apache.flink.runtime.jobmaster.DefaultJobMasterServiceProcess.lambda$new$0(DefaultJobMasterServiceProcess.java:97)
  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$AsyncSupply.run(CompletableFuture.java:1609)
  at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
  at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
  at java.lang.Thread.run(Thread.java:748)
  Caused by: java.util.concurrent.CompletionException: java.lang.IllegalStateException: Failed to rollback to checkpoint/savepoint s3://rinc-ingestion-service/flink-savepoints/namespaces/default/deployments/e8f1afd0-9236-43c2-93cb-db16604da594/1a6716af-6357-4ae3-95da-dad0bdb1f7cc/savepoint-f101ee-feb4e8023b09. Cannot map checkpoint/savepoint state for operator 7df19f87deec5680128845fd9a6ca18d to the new program, because the operator is not available in the new program. If you want to allow to skip this, you can set the --allowNonRestoredState option on the CLI.
  at java.util.concurrent.CompletableFuture.encodeThrowable(CompletableFuture.java:273)
  at java.util.concurrent.CompletableFuture.completeThrowable(CompletableFuture.java:280)
  at java.util.concurrent.CompletableFuture$AsyncSupply.run(CompletableFuture.java:1606)
  ... 3 more
  Caused by: java.lang.IllegalStateException: Failed to rollback to checkpoint/savepoint s3://rinc-ingestion-service/flink-savepoints/namespaces/default/deployments/e8f1afd0-9236-43c2-93cb-db16604da594/1a6716af-6357-4ae3-95da-dad0bdb1f7cc/savepoint-f101ee-feb4e8023b09. Cannot map checkpoint/savepoint state for operator 7df19f87deec5680128845fd9a6ca18d to the new program, because the operator is not available in the new program. If you want to allow to skip this, you can set the --allowNonRestoredState option on the CLI.
  at org.apache.flink.runtime.checkpoint.Checkpoints.throwNonRestoredStateException(Checkpoints.java:230)
  at org.apache.flink.runtime.checkpoint.Checkpoints.loadAndValidateCheckpoint(Checkpoints.java:194)
  at org.apache.flink.runtime.checkpoint.CheckpointCoordinator.restoreSavepoint(CheckpointCoordinator.java:1648)
  at org.apache.flink.runtime.scheduler.DefaultExecutionGraphFactory.tryRestoreExecutionGraphFromSavepoint(DefaultExecutionGraphFactory.java:163)
  at org.apache.flink.runtime.scheduler.DefaultExecutionGraphFactory.createAndRestoreExecutionGraph(DefaultExecutionGraphFactory.java:138)
  at org.apache.flink.runtime.scheduler.SchedulerBase.createAndRestoreExecutionGraph(SchedulerBase.java:335)
  at org.apache.flink.runtime.scheduler.SchedulerBase.(SchedulerBase.java:191)
  at org.apache.flink.runtime.scheduler.DefaultScheduler.(DefaultScheduler.java:140)
  at org.apache.flink.runtime.scheduler.DefaultSchedulerFactory.createInstance(DefaultSchedulerFactory.java:134)
  at org.apache.flink.runtime.jobmaster.DefaultSlotPoolServiceSchedulerFactory.createScheduler(DefaultSlotPoolServiceSchedulerFactory.java:110)
  at org.apache.flink.runtime.jobmaster.JobMaster.createScheduler(JobMaster.java:346)
  at org.apache.flink.runtime.jobmaster.JobMaster.(JobMaster.java:323)
  at org.apache.flink.runtime.jobmaster.factories.DefaultJobMasterServiceFactory.internalCreateJobMasterService(DefaultJobMasterServiceFactory.java:106)
  at org.apache.flink.runtime.jobmaster.factories.DefaultJobMasterServiceFactory.lambda$createJobMasterService$0(DefaultJobMasterServiceFactory.java:94)
  at org.apache.flink.util.function.FunctionUtils.lambda$uncheckedSupplier$4(FunctionUtils.java:112)
  at java.util.concurrent.CompletableFuture$AsyncSupply.run(CompletableFuture.java:1604)
  ... 3 more</p>
  </blockquote>
  <p>The flink app code is below.</p>
  <pre><code>import io.confluent.kafka.serializers.KafkaAvroDeserializerConfig;
  import org.apache.flink.api.common.eventtime.WatermarkStrategy;
  import org.apache.flink.connector.kafka.source.KafkaSource;
  import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer;
  import org.apache.flink.streaming.api.datastream.DataStreamSource;
  import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
  import testRecord.DataRecordAvro;
  public class KafkaAvroDeserialize {
      public static void main(String[] args) throws Exception {
          StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
          KafkaSource&lt;DataRecordAvro&gt; source = KafkaSource.&lt;DataRecordAvro&gt;builder()
                  .setBootstrapServers(&quot;pkc-2396y.us-east-1.aws.confluent.cloud:9092&quot;)
                  .setTopics(&quot;test&quot;)
                  .setGroupId(&quot;demo-consumer-avro-1&quot;)
                  .setProperty(KafkaAvroDeserializerConfig.SCHEMA_REGISTRY_URL_CONFIG, &quot;https://psrc-1wydj.us-east-2.aws.confluent.cloud&quot;)
                  .setProperty(&quot;basic.auth.credentials.source&quot;, &quot;USER_INFO&quot;)
                  .setProperty(&quot;basic.auth.user.info&quot;, &quot;5CDO5ZR4HDJSGOHT:v/YJxrw6iR+ASbIwMGefZN9SIzPccuyyTQC8EXFn8cOcJCeC1EASEfvHzFRflyXd&quot;)
                  .setProperty(&quot;advertised.host.name&quot;, &quot;pkc-2396y.us-east-1.aws.confluent.cloud:9092&quot;)
                  .setValueOnlyDeserializer(new AvroDeserializer&lt;&gt;(DataRecordAvro.class))
                  .setStartingOffsets(OffsetsInitializer.earliest())
                  .build();
          DataStreamSource&lt;DataRecordAvro&gt; input = env.fromSource(source, WatermarkStrategy.noWatermarks(), &quot;Kafka Source&quot;);
          input.map(record -&gt; &quot;DESERIALIZED: &quot; + record.getMsgType() + &quot;-&quot; + record.get(0) + &quot;-&quot; + record.get(1) + &quot;-&quot; + record.get(2)).print();
          env.execute(&quot;Printing the payload&quot;);
      }
  }
  </code></pre>