Hi 👋 I recently upgrade to flink 1.17.1 from 1.16.1 and I and submitting sql queries which does some streaming from Kafka and write back to Kafka. I am getting a null pointer exception when running the job via command line using

flink run -c

com.org`.project.MainClass /Users/some/workspace/project/build/libs/project_fat.jar --flinkRestUrl` http://localhost:8081

--environment local-jar --suffix some-suffix --s3_region us-east-1 --dag_label some_label

command. My StreamingEnv is initialized using

Configuration config = new Configuration();
		config.set(ExecutionCheckpointingOptions.ENABLE_CHECKPOINTS_AFTER_TASKS_FINISH, true);
		config.set(PipelineOptions.NAME, globalParams.get("job_name"));

		CheckpointService checkpointService = injector.getInstance(CheckpointService.class);
		String checkpointPath = checkpointService.getLatestCheckpointLocation(globalParams.get("checkpoint_path"));
		if (checkpointPath != null) {
			<http://log.info|log.info>("Restarting from the latest checkpoint location :{}", checkpointPath);
			// this needs to set on a Configuration and has to supplied to StreamExecutionEnvironment.getExecutionEnvironment(config);, without this the job is unable to restart from a previous checkpoint.
			config.set(SAVEPOINT_PATH, checkpointPath);
		}

		StreamExecutionEnvironment streamExecutionEnvironment;
		if (globalParams.get("environment").equals("local")) {
			config.set(RestOptions.PORT, 8085);
			streamExecutionEnvironment = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(config);
		} else {
			streamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment(config);
		}

		// should show user defined param in flink ui
		streamExecutionEnvironment.getConfig().setGlobalJobParameters(ParameterTool.fromMap(globalParams));
		//set default configs for all jobs
		streamExecutionEnvironment.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.AT_LEAST_ONCE);
		streamExecutionEnvironment.getCheckpointConfig().setMaxConcurrentCheckpoints(1);
		streamExecutionEnvironment.getCheckpointConfig().setExternalizedCheckpointCleanup(
				CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
		streamExecutionEnvironment.setStateBackend(new EmbeddedRocksDBStateBackend(true));

		String sinkParallelism = globalParams.get("sink_parallelism");
		int parallelism = sinkParallelism != null ? Integer.parseInt(sinkParallelism) : Runtime.getRuntime().availableProcessors();
		streamExecutionEnvironment.setParallelism(parallelism);

		// domain event specific configs
		streamExecutionEnvironment.enableCheckpointing(Long.parseLong(globalParams.get("checkpoint_interval_ms")), CheckpointingMode.AT_LEAST_ONCE);
		streamExecutionEnvironment.getCheckpointConfig().setCheckpointTimeout(Long.parseLong(globalParams.get("checkpoint_completion_timeout_ms")));
		streamExecutionEnvironment.getCheckpointConfig().setMinPauseBetweenCheckpoints(Long.parseLong(globalParams.get("checkpoint_pause_interval_ms")));
		streamExecutionEnvironment.getCheckpointConfig().setCheckpointStorage(globalParams.get("checkpoint_path"));

		//restart 10 times, with a delay of 10 seconds between each run on any exception.
		streamExecutionEnvironment.setRestartStrategy(RestartStrategies.fixedDelayRestart(10, 10000L));

and StreamTableEnvironment using

this.streamTableEnvironment = StreamTableEnvironment.create(getStreamExecutionEnvironment());

		// needed to handle all timestamp in to UTC
		streamTableEnvironment.getConfig().setLocalTimeZone(ZoneOffset.UTC);
		// needed to handle null values in column that are defined not null
		streamTableEnvironment.getConfig().getConfiguration().setString("table.exec.sink.not-null-enforcer", "DROP");

The exception is

org.apache.flink.runtime.rest.util.RestClientException: [Internal server error., <Exception on server side:
java.lang.NullPointerException
	at org.apache.flink.api.common.ExecutionConfig.getExecutionMode(ExecutionConfig.java:561)
	at org.apache.flink.api.common.ArchivedExecutionConfig.<init>(ArchivedExecutionConfig.java:46)
	at org.apache.flink.api.common.ExecutionConfig.archive(ExecutionConfig.java:1008)
	at org.apache.flink.runtime.executiongraph.DefaultExecutionGraph.getArchivedExecutionConfig(DefaultExecutionGraph.java:1114)
	at org.apache.flink.runtime.executiongraph.ArchivedExecutionGraph.createFrom(ArchivedExecutionGraph.java:344)
	at org.apache.flink.runtime.executiongraph.ArchivedExecutionGraph.createFrom(ArchivedExecutionGraph.java:291)
	at org.apache.flink.runtime.scheduler.SchedulerBase.requestJob(SchedulerBase.java:785)
	at org.apache.flink.runtime.jobmaster.JobMaster.requestJob(JobMaster.java:855)
	at jdk.internal.reflect.GeneratedMethodAccessor44.invoke(Unknown Source)
	at java.base/jdk.internal.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
	at java.base/java.lang.reflect.Method.invoke(Method.java:566)
	at org.apache.flink.runtime.rpc.akka.AkkaRpcActor.lambda$handleRpcInvocation$1(AkkaRpcActor.java:309)
	at org.apache.flink.runtime.concurrent.akka.ClassLoadingUtils.runWithContextClassLoader(ClassLoadingUtils.java:83)
	at org.apache.flink.runtime.rpc.akka.AkkaRpcActor.handleRpcInvocation(AkkaRpcActor.java:307)
	at org.apache.flink.runtime.rpc.akka.AkkaRpcActor.handleRpcMessage(AkkaRpcActor.java:222)
	at org.apache.flink.runtime.rpc.akka.FencedAkkaRpcActor.handleRpcMessage(FencedAkkaRpcActor.java:84)
	at org.apache.flink.runtime.rpc.akka.AkkaRpcActor.handleMessage(AkkaRpcActor.java:168)
	at akka.japi.pf.UnitCaseStatement.apply(CaseStatements.scala:24)
	at akka.japi.pf.UnitCaseStatement.apply(CaseStatements.scala:20)
	at scala.PartialFunction.applyOrElse(PartialFunction.scala:127)
	at scala.PartialFunction.applyOrElse$(PartialFunction.scala:126)
	at akka.japi.pf.UnitCaseStatement.applyOrElse(CaseStatements.scala:20)
	at scala.PartialFunction$OrElse.applyOrElse(PartialFunction.scala:175)
	at scala.PartialFunction$OrElse.applyOrElse(PartialFunction.scala:176)
	at scala.PartialFunction$OrElse.applyOrElse(PartialFunction.scala:176)
	at akka.actor.Actor.aroundReceive(Actor.scala:537)
	at akka.actor.Actor.aroundReceive$(Actor.scala:535)
	at akka.actor.AbstractActor.aroundReceive(AbstractActor.scala:220)
	at akka.actor.ActorCell.receiveMessage(ActorCell.scala:579)
	at akka.actor.ActorCell.invoke(ActorCell.scala:547)
	at akka.dispatch.Mailbox.processMailbox(Mailbox.scala:270)
	at akka.dispatch.Mailbox.run(Mailbox.scala:231)
	at akka.dispatch.Mailbox.exec(Mailbox.scala:243)
	at java.base/java.util.concurrent.ForkJoinTask.doExec(ForkJoinTask.java:290)
	at java.base/java.util.concurrent.ForkJoinPool$WorkQueue.topLevelExec(ForkJoinPool.java:1020)
	at java.base/java.util.concurrent.ForkJoinPool.scan(ForkJoinPool.java:1656)
	at java.base/java.util.concurrent.ForkJoinPool.runWorker(ForkJoinPool.java:1594)
	at java.base/java.util.concurrent.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:183)

End of exception on server side>]
	at org.apache.flink.runtime.rest.RestClient.parseResponse(RestClient.java:534)
	at org.apache.flink.runtime.rest.RestClient.lambda$submitRequest$3(RestClient.java:514)
	at java.base/java.util.concurrent.CompletableFuture$UniCompose.tryFire(CompletableFuture.java:1072)!
	at java.base/java.util.concurrent.CompletableFuture$Completion.run(CompletableFuture.java:478)
	at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128)
	at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628)
	at java.base/java.lang.Thread.run(Thread.java:829)

Thanks for the help!