[inlong] branch master updated: [INLONG-7172][Sort] Fix new table write into iceberg failed (#7173)

[zi...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

zirui pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/inlong.git


The following commit(s) were added to refs/heads/master by this push:
     new cfefd68c1 [INLONG-7172][Sort] Fix new table write into iceberg failed (#7173)
cfefd68c1 is described below

commit cfefd68c127c2afd22f35a8c901992b3f53f0166
Author: emhui <11...@users.noreply.github.com>
AuthorDate: Fri Jan 6 16:23:19 2023 +0800

    [INLONG-7172][Sort] Fix new table write into iceberg failed (#7173)
---
 .../inlong/sort/iceberg/sink/multiple/IcebergSingleFileCommiter.java   | 3 ++-
 1 file changed, 2 insertions(+), 1 deletion(-)

diff --git a/inlong-sort/sort-connectors/iceberg/src/main/java/org/apache/inlong/sort/iceberg/sink/multiple/IcebergSingleFileCommiter.java b/inlong-sort/sort-connectors/iceberg/src/main/java/org/apache/inlong/sort/iceberg/sink/multiple/IcebergSingleFileCommiter.java
index 4f13c8ec6..6c5e14fe7 100644
--- a/inlong-sort/sort-connectors/iceberg/src/main/java/org/apache/inlong/sort/iceberg/sink/multiple/IcebergSingleFileCommiter.java
+++ b/inlong-sort/sort-connectors/iceberg/src/main/java/org/apache/inlong/sort/iceberg/sink/multiple/IcebergSingleFileCommiter.java
@@ -160,7 +160,8 @@ public class IcebergSingleFileCommiter extends IcebergProcessFunction<WriteResul
 
         this.checkpointsState = context.getOperatorStateStore().getListState(stateDescriptor);
         this.jobIdState = context.getOperatorStateStore().getListState(jobIdDescriptor);
-        if (context.isRestored()) {
+        // New table doesn't have state, so it doesn't need to do restore operation.
+        if (context.isRestored() && jobIdState.get().iterator().hasNext()) {
             String restoredFlinkJobId = jobIdState.get().iterator().next();
             Preconditions.checkState(!Strings.isNullOrEmpty(restoredFlinkJobId),
                     "Flink job id parsed from checkpoint snapshot shouldn't be null or empty");