This is an automated email from the ASF dual-hosted git repository. dockerzhang pushed a commit to branch branch-1.5 in repository https://gitbox.apache.org/repos/asf/inlong.git
commit 7cb6f77f3d3cfc25ed7865aeab8df485b039e234 Author: emhui <[email protected]> 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");
