[Bug]: Loading Mixed-Iceberg format tables of internal catalog failed in Flink

[zhoujinsong]

What happened?

When I tried to write into a mixed format table in Flink, an error occurred when loading the table.

Catalog type: internal
Table format: Mixed iceberg
Engine: Flink

Affects Versions

master

What engines are you seeing the problem on?

Flink

How to reproduce

• Create an internal catalog with mixed iceberg format
• Create a table in the catalog
• Try to load the table in Flink environment

Relevant log output

2024-02-29 16:54:58,965 INFO  org.apache.flink.runtime.dispatcher.StandaloneDispatcher     [] - Could not archive completed job (application driver)(13bcbf5d19348cede80f356ae79c070c) to the history server.
java.util.concurrent.CompletionException: java.lang.RuntimeException: org.apache.hadoop.fs.FileAlreadyExistsException: Path already exists: jfs://autopai-bigdata/flink/completed-jobs/13bcbf5d19348cede80f356ae79c070c
	at java.util.concurrent.CompletableFuture.encodeThrowable(CompletableFuture.java:273) ~[?:1.8.0_221]
	at java.util.concurrent.CompletableFuture.completeThrowable(CompletableFuture.java:280) ~[?:1.8.0_221]
	at java.util.concurrent.CompletableFuture$AsyncRun.run(CompletableFuture.java:1629) ~[?:1.8.0_221]
	at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149) ~[?:1.8.0_221]
	at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624) ~[?:1.8.0_221]
	at java.lang.Thread.run(Thread.java:748) ~[?:1.8.0_221]
Caused by: java.lang.RuntimeException: org.apache.hadoop.fs.FileAlreadyExistsException: Path already exists: jfs://autopai-bigdata/flink/completed-jobs/13bcbf5d19348cede80f356ae79c070c
	at org.apache.flink.util.ExceptionUtils.rethrow(ExceptionUtils.java:321) ~[flink-dist-1.17.2.jar:1.17.2]
	at org.apache.flink.util.function.ThrowingRunnable.lambda$unchecked$0(ThrowingRunnable.java:51) ~[flink-dist-1.17.2.jar:1.17.2]
	at java.util.concurrent.CompletableFuture$AsyncRun.run(CompletableFuture.java:1626) ~[?:1.8.0_221]
	... 3 more
Caused by: org.apache.hadoop.fs.FileAlreadyExistsException: Path already exists: jfs://autopai-bigdata/flink/completed-jobs/13bcbf5d19348cede80f356ae79c070c
	at io.juicefs.JuiceFileSystemImpl.create(JuiceFileSystemImpl.java:1122) ~[juicefs-hadoop-1.0.2.jar:?]
	at io.juicefs.JuiceFileSystem.create(JuiceFileSystem.java:121) ~[juicefs-hadoop-1.0.2.jar:?]
	at org.apache.hadoop.fs.FileSystem.create(FileSystem.java:1175) ~[hadoop-common-3.3.1.jar:?]
	at org.apache.hadoop.fs.FileSystem.create(FileSystem.java:1064) ~[hadoop-common-3.3.1.jar:?]
	at org.apache.flink.runtime.fs.hdfs.HadoopFileSystem.create(HadoopFileSystem.java:154) ~[flink-dist-1.17.2.jar:1.17.2]
	at org.apache.flink.runtime.fs.hdfs.HadoopFileSystem.create(HadoopFileSystem.java:37) ~[flink-dist-1.17.2.jar:1.17.2]
	at org.apache.flink.runtime.history.FsJobArchivist.archiveJob(FsJobArchivist.java:74) ~[flink-dist-1.17.2.jar:1.17.2]
	at org.apache.flink.runtime.dispatcher.JsonResponseHistoryServerArchivist.lambda$archiveExecutionGraph$0(JsonResponseHistoryServerArchivist.java:58) ~[flink-dist-1.17.2.jar:1.17.2]
	at org.apache.flink.util.function.ThrowingRunnable.lambda$unchecked$0(ThrowingRunnable.java:49) ~[flink-dist-1.17.2.jar:1.17.2]
	at java.util.concurrent.CompletableFuture$AsyncRun.run(CompletableFuture.java:1626) ~[?:1.8.0_221]
	... 3 more
2024-02-29 16:54:58,973 INFO  org.apache.flink.client.deployment.application.ApplicationDispatcherBootstrap [] - Application FAILED: 
java.util.concurrent.CompletionException: org.apache.flink.client.deployment.application.UnsuccessfulExecutionException: Application Status: FAILED
	at org.apache.flink.client.deployment.application.ApplicationDispatcherBootstrap.lambda$unwrapJobResultException$7(ApplicationDispatcherBootstrap.java:403) ~[flink-dist-1.17.2.jar:1.17.2]
	at java.util.concurrent.CompletableFuture.uniApply(CompletableFuture.java:602) ~[?:1.8.0_221]
	at java.util.concurrent.CompletableFuture$UniApply.tryFire(CompletableFuture.java:577) ~[?:1.8.0_221]
	at java.util.concurrent.CompletableFuture.postComplete(CompletableFuture.java:474) ~[?:1.8.0_221]
	at java.util.concurrent.CompletableFuture.complete(CompletableFuture.java:1962) ~[?:1.8.0_221]
	at org.apache.flink.client.deployment.application.JobStatusPollingUtils.lambda$null$2(JobStatusPollingUtils.java:101) ~[flink-dist-1.17.2.jar:1.17.2]
	at java.util.concurrent.CompletableFuture.uniWhenComplete(CompletableFuture.java:760) ~[?:1.8.0_221]
	at java.util.concurrent.CompletableFuture$UniWhenComplete.tryFire(CompletableFuture.java:736) ~[?:1.8.0_221]
	at java.util.concurrent.CompletableFuture.postComplete(CompletableFuture.java:474) ~[?:1.8.0_221]
	at java.util.concurrent.CompletableFuture.complete(CompletableFuture.java:1962) ~[?:1.8.0_221]
	at org.apache.flink.runtime.rpc.akka.AkkaInvocationHandler.lambda$invokeRpc$1(AkkaInvocationHandler.java:267) ~[?:?]
	at java.util.concurrent.CompletableFuture.uniWhenComplete(CompletableFuture.java:760) ~[?:1.8.0_221]
	at java.util.concurrent.CompletableFuture$UniWhenComplete.tryFire(CompletableFuture.java:736) ~[?:1.8.0_221]
	at java.util.concurrent.CompletableFuture.postComplete(CompletableFuture.java:474) ~[?:1.8.0_221]
	at java.util.concurrent.CompletableFuture.complete(CompletableFuture.java:1962) ~[?:1.8.0_221]
	at org.apache.flink.util.concurrent.FutureUtils.doForward(FutureUtils.java:1300) ~[flink-dist-1.17.2.jar:1.17.2]
	at org.apache.flink.runtime.concurrent.akka.ClassLoadingUtils.lambda$null$1(ClassLoadingUtils.java:93) ~[flink-rpc-akka_515345c6-ace8-4266-8f0c-09c1bfbf1968.jar:1.17.2]
	at org.apache.flink.runtime.concurrent.akka.ClassLoadingUtils.runWithContextClassLoader(ClassLoadingUtils.java:68) ~[flink-rpc-akka_515345c6-ace8-4266-8f0c-09c1bfbf1968.jar:1.17.2]
	at org.apache.flink.runtime.concurrent.akka.ClassLoadingUtils.lambda$guardCompletionWithContextClassLoader$2(ClassLoadingUtils.java:92) ~[flink-rpc-akka_515345c6-ace8-4266-8f0c-09c1bfbf1968.jar:1.17.2]
	at java.util.concurrent.CompletableFuture.uniWhenComplete(CompletableFuture.java:760) ~[?:1.8.0_221]
	at java.util.concurrent.CompletableFuture$UniWhenComplete.tryFire(CompletableFuture.java:736) ~[?:1.8.0_221]
	at java.util.concurrent.CompletableFuture.postComplete(CompletableFuture.java:474) ~[?:1.8.0_221]
	at java.util.concurrent.CompletableFuture.complete(CompletableFuture.java:1962) ~[?:1.8.0_221]
	at org.apache.flink.runtime.concurrent.akka.AkkaFutureUtils$1.onComplete(AkkaFutureUtils.java:47) ~[flink-rpc-akka_515345c6-ace8-4266-8f0c-09c1bfbf1968.jar:1.17.2]
	at akka.dispatch.OnComplete.internal(Future.scala:300) ~[flink-rpc-akka_515345c6-ace8-4266-8f0c-09c1bfbf1968.jar:1.17.2]
	at akka.dispatch.OnComplete.internal(Future.scala:297) ~[flink-rpc-akka_515345c6-ace8-4266-8f0c-09c1bfbf1968.jar:1.17.2]
	at akka.dispatch.japi$CallbackBridge.apply(Future.scala:224) ~[flink-rpc-akka_515345c6-ace8-4266-8f0c-09c1bfbf1968.jar:1.17.2]
	at akka.dispatch.japi$CallbackBridge.apply(Future.scala:221) ~[flink-rpc-akka_515345c6-ace8-4266-8f0c-09c1bfbf1968.jar:1.17.2]
	at scala.concurrent.impl.CallbackRunnable.run(Promise.scala:64) ~[flink-rpc-akka_515345c6-ace8-4266-8f0c-09c1bfbf1968.jar:1.17.2]
	at org.apache.flink.runtime.concurrent.akka.AkkaFutureUtils$DirectExecutionContext.execute(AkkaFutureUtils.java:65) ~[flink-rpc-akka_515345c6-ace8-4266-8f0c-09c1bfbf1968.jar:1.17.2]
	at scala.concurrent.impl.CallbackRunnable.executeWithValue(Promise.scala:72) ~[flink-rpc-akka_515345c6-ace8-4266-8f0c-09c1bfbf1968.jar:1.17.2]
	at scala.concurrent.impl.Promise$DefaultPromise.$anonfun$tryComplete$1(Promise.scala:288) ~[flink-rpc-akka_515345c6-ace8-4266-8f0c-09c1bfbf1968.jar:1.17.2]
	at scala.concurrent.impl.Promise$DefaultPromise.$anonfun$tryComplete$1$adapted(Promise.scala:288) ~[flink-rpc-akka_515345c6-ace8-4266-8f0c-09c1bfbf1968.jar:1.17.2]
	at scala.concurrent.impl.Promise$DefaultPromise.tryComplete(Promise.scala:288) ~[flink-rpc-akka_515345c6-ace8-4266-8f0c-09c1bfbf1968.jar:1.17.2]
	at akka.pattern.PromiseActorRef.$bang(AskSupport.scala:622) ~[flink-rpc-akka_515345c6-ace8-4266-8f0c-09c1bfbf1968.jar:1.17.2]
	at akka.pattern.PipeToSupport$PipeableFuture$$anonfun$pipeTo$1.applyOrElse(PipeToSupport.scala:24) ~[flink-rpc-akka_515345c6-ace8-4266-8f0c-09c1bfbf1968.jar:1.17.2]
	at akka.pattern.PipeToSupport$PipeableFuture$$anonfun$pipeTo$1.applyOrElse(PipeToSupport.scala:23) ~[flink-rpc-akka_515345c6-ace8-4266-8f0c-09c1bfbf1968.jar:1.17.2]
	at scala.concurrent.Future.$anonfun$andThen$1(Future.scala:536) ~[flink-rpc-akka_515345c6-ace8-4266-8f0c-09c1bfbf1968.jar:1.17.2]
	at scala.concurrent.impl.Promise.liftedTree1$1(Promise.scala:33) ~[flink-rpc-akka_515345c6-ace8-4266-8f0c-09c1bfbf1968.jar:1.17.2]
	at scala.concurrent.impl.Promise.$anonfun$transform$1(Promise.scala:33) ~[flink-rpc-akka_515345c6-ace8-4266-8f0c-09c1bfbf1968.jar:1.17.2]
	at scala.concurrent.impl.CallbackRunnable.run(Promise.scala:64) ~[flink-rpc-akka_515345c6-ace8-4266-8f0c-09c1bfbf1968.jar:1.17.2]
	at akka.dispatch.BatchingExecutor$AbstractBatch.processBatch(BatchingExecutor.scala:63) ~[flink-rpc-akka_515345c6-ace8-4266-8f0c-09c1bfbf1968.jar:1.17.2]
	at akka.dispatch.BatchingExecutor$BlockableBatch.$anonfun$run$1(BatchingExecutor.scala:100) ~[flink-rpc-akka_515345c6-ace8-4266-8f0c-09c1bfbf1968.jar:1.17.2]
	at scala.runtime.java8.JFunction0$mcV$sp.apply(JFunction0$mcV$sp.java:23) [flink-rpc-akka_515345c6-ace8-4266-8f0c-09c1bfbf1968.jar:1.17.2]
	at scala.concurrent.BlockContext$.withBlockContext(BlockContext.scala:85) [flink-rpc-akka_515345c6-ace8-4266-8f0c-09c1bfbf1968.jar:1.17.2]
	at akka.dispatch.BatchingExecutor$BlockableBatch.run(BatchingExecutor.scala:100) [flink-rpc-akka_515345c6-ace8-4266-8f0c-09c1bfbf1968.jar:1.17.2]
	at akka.dispatch.TaskInvocation.run(AbstractDispatcher.scala:49) [flink-rpc-akka_515345c6-ace8-4266-8f0c-09c1bfbf1968.jar:1.17.2]
	at akka.dispatch.ForkJoinExecutorConfigurator$AkkaForkJoinTask.exec(ForkJoinExecutorConfigurator.scala:48) [flink-rpc-akka_515345c6-ace8-4266-8f0c-09c1bfbf1968.jar:1.17.2]
	at java.util.concurrent.ForkJoinTask.doExec(ForkJoinTask.java:289) [?:1.8.0_221]
	at java.util.concurrent.ForkJoinPool$WorkQueue.runTask(ForkJoinPool.java:1056) [?:1.8.0_221]
	at java.util.concurrent.ForkJoinPool.runWorker(ForkJoinPool.java:1692) [?:1.8.0_221]
	at java.util.concurrent.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:157) [?:1.8.0_221]
Caused by: org.apache.flink.client.deployment.application.UnsuccessfulExecutionException: Application Status: FAILED
	at org.apache.flink.client.deployment.application.UnsuccessfulExecutionException.fromJobResult(UnsuccessfulExecutionException.java:71) ~[flink-dist-1.17.2.jar:1.17.2]
	... 52 more
Caused by: org.apache.flink.runtime.client.JobExecutionException: Job execution failed.
	at org.apache.flink.runtime.jobmaster.JobResult.toJobExecutionResult(JobResult.java:144) ~[flink-dist-1.17.2.jar:1.17.2]
	at org.apache.flink.client.deployment.application.UnsuccessfulExecutionException.fromJobResult(UnsuccessfulExecutionException.java:60) ~[flink-dist-1.17.2.jar:1.17.2]
	... 52 more
Caused by: org.apache.flink.client.program.ProgramInvocationException: The main method caused an error: java.lang.NoClassDefFoundError: org/apache/hc/core5/http/ParseException
	at org.apache.flink.client.program.PackagedProgram.callMainMethod(PackagedProgram.java:372) ~[flink-dist-1.17.2.jar:1.17.2]
	at org.apache.flink.client.program.PackagedProgram.invokeInteractiveModeForExecution(PackagedProgram.java:222) ~[flink-dist-1.17.2.jar:1.17.2]
	at org.apache.flink.client.ClientUtils.executeProgram(ClientUtils.java:105) ~[flink-dist-1.17.2.jar:1.17.2]
	at org.apache.flink.client.deployment.application.ApplicationDispatcherBootstrap.runApplicationEntryPoint(ApplicationDispatcherBootstrap.java:301) ~[flink-dist-1.17.2.jar:1.17.2]
	at org.apache.flink.client.deployment.application.ApplicationDispatcherBootstrap.lambda$runApplicationAsync$2(ApplicationDispatcherBootstrap.java:254) ~[flink-dist-1.17.2.jar:1.17.2]
	at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511) ~[?:1.8.0_221]
	at java.util.concurrent.FutureTask.run(FutureTask.java:266) ~[?:1.8.0_221]
	at org.apache.flink.runtime.concurrent.akka.ActorSystemScheduledExecutorAdapter$ScheduledFutureTask.run(ActorSystemScheduledExecutorAdapter.java:171) ~[flink-rpc-akka_515345c6-ace8-4266-8f0c-09c1bfbf1968.jar:1.17.2]
	at org.apache.flink.runtime.concurrent.akka.ClassLoadingUtils.runWithContextClassLoader(ClassLoadingUtils.java:68) ~[flink-rpc-akka_515345c6-ace8-4266-8f0c-09c1bfbf1968.jar:1.17.2]
	at org.apache.flink.runtime.concurrent.akka.ClassLoadingUtils.lambda$withContextClassLoader$0(ClassLoadingUtils.java:41) ~[flink-rpc-akka_515345c6-ace8-4266-8f0c-09c1bfbf1968.jar:1.17.2]
	... 6 more
Caused by: java.lang.RuntimeException: java.lang.NoClassDefFoundError: org/apache/hc/core5/http/ParseException
	at com.autopai.bigdata.realtime.base.BaseStreamJob.run(BaseStreamJob.java:118) ~[?:?]
	at com.autopai.bigdata.realtime.amoro.ods.SDKLogServer.main(SDKLogServer.java:34) ~[?:?]
	at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method) ~[?:1.8.0_221]
	at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62) ~[?:1.8.0_221]
	at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43) ~[?:1.8.0_221]
	at java.lang.reflect.Method.invoke(Method.java:498) ~[?:1.8.0_221]
	at org.apache.flink.client.program.PackagedProgram.callMainMethod(PackagedProgram.java:355) ~[flink-dist-1.17.2.jar:1.17.2]
	at org.apache.flink.client.program.PackagedProgram.invokeInteractiveModeForExecution(PackagedProgram.java:222) ~[flink-dist-1.17.2.jar:1.17.2]
	at org.apache.flink.client.ClientUtils.executeProgram(ClientUtils.java:105) ~[flink-dist-1.17.2.jar:1.17.2]
	at org.apache.flink.client.deployment.application.ApplicationDispatcherBootstrap.runApplicationEntryPoint(ApplicationDispatcherBootstrap.java:301) ~[flink-dist-1.17.2.jar:1.17.2]
	at org.apache.flink.client.deployment.application.ApplicationDispatcherBootstrap.lambda$runApplicationAsync$2(ApplicationDispatcherBootstrap.java:254) ~[flink-dist-1.17.2.jar:1.17.2]
	at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511) ~[?:1.8.0_221]
	at java.util.concurrent.FutureTask.run(FutureTask.java:266) ~[?:1.8.0_221]
	at org.apache.flink.runtime.concurrent.akka.ActorSystemScheduledExecutorAdapter$ScheduledFutureTask.run(ActorSystemScheduledExecutorAdapter.java:171) ~[flink-rpc-akka_515345c6-ace8-4266-8f0c-09c1bfbf1968.jar:1.17.2]
	at org.apache.flink.runtime.concurrent.akka.ClassLoadingUtils.runWithContextClassLoader(ClassLoadingUtils.java:68) ~[flink-rpc-akka_515345c6-ace8-4266-8f0c-09c1bfbf1968.jar:1.17.2]
	at org.apache.flink.runtime.concurrent.akka.ClassLoadingUtils.lambda$withContextClassLoader$0(ClassLoadingUtils.java:41) ~[flink-rpc-akka_515345c6-ace8-4266-8f0c-09c1bfbf1968.jar:1.17.2]
	... 6 more

Anything else

Are you willing to submit a PR?

• Yes I am willing to submit a PR!

Code of Conduct

• I agree to follow this project's Code of Conduct