[GitHub] [incubator-iceberg] waterlx opened a new pull request #856: [WIP] Flink Iceberg sink

[GitBox <gi...@apache.org>]

waterlx opened a new pull request #856:
URL: https://github.com/apache/incubator-iceberg/pull/856


   Flink Iceberg sink, trying to address #567, modeled after [nfflink-connector-iceberg](https://github.com/Netflix-Skunkworks/nfflink-connector-iceberg)
   
   A workable version which could pass checkstyle, compiled with the latest master code, and has basic functions verified:
   - DataFile generated
   - Committed successfully
   
   Being improve and polished.
   
   [The design doc](https://docs.google.com/document/d/19M-sP6FlTVm7BV7MM4Om1n_MVo1xCy7GyDl_9ZAjVNQ/edit?usp=sharing), which contains:
   - Design extracted from nfflink-connector-iceberg
   - Future work and improvements to do
   - Sampe code


----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org



---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org

[GitHub] [incubator-iceberg] stevenzwu commented on a change in pull request #856: [WIP] Flink Iceberg sink

[GitBox <gi...@apache.org>]

stevenzwu commented on a change in pull request #856:
URL: https://github.com/apache/incubator-iceberg/pull/856#discussion_r420914501



##########
File path: flink/src/main/java/org/apache/iceberg/flink/connector/sink/IcebergCommitter.java
##########
@@ -0,0 +1,642 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iceberg.flink.connector.sink;
+
+import com.google.common.base.Joiner;
+import com.google.common.base.Strings;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.stream.Collectors;
+import javax.annotation.Nullable;
+import org.apache.flink.annotation.VisibleForTesting;
+import org.apache.flink.api.common.JobExecutionResult;
+import org.apache.flink.api.common.JobID;
+import org.apache.flink.api.common.state.ListState;
+import org.apache.flink.api.common.state.ListStateDescriptor;
+import org.apache.flink.api.java.ExecutionEnvironment;
+import org.apache.flink.configuration.CheckpointingOptions;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.runtime.state.CheckpointListener;
+import org.apache.flink.runtime.state.FunctionInitializationContext;
+import org.apache.flink.runtime.state.FunctionSnapshotContext;
+import org.apache.flink.streaming.api.checkpoint.CheckpointedFunction;
+import org.apache.flink.streaming.api.environment.CheckpointConfig;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
+import org.apache.flink.util.Preconditions;
+import org.apache.hadoop.hive.conf.HiveConf.ConfVars;
+import org.apache.iceberg.AppendFiles;
+import org.apache.iceberg.DataFile;
+import org.apache.iceberg.ManifestFile;
+import org.apache.iceberg.ManifestWriter;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.Snapshot;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.Transaction;
+import org.apache.iceberg.UpdateProperties;
+import org.apache.iceberg.catalog.Catalog;
+import org.apache.iceberg.catalog.TableIdentifier;
+import org.apache.iceberg.flink.connector.IcebergConnectorConstant;
+import org.apache.iceberg.flink.connector.model.CommitMetadata;
+import org.apache.iceberg.flink.connector.model.CommitMetadataUtil;
+import org.apache.iceberg.flink.connector.model.FlinkManifestFile;
+import org.apache.iceberg.flink.connector.model.FlinkManifestFileUtil;
+import org.apache.iceberg.flink.connector.model.GenericFlinkManifestFile;
+import org.apache.iceberg.flink.connector.model.ManifestFileState;
+import org.apache.iceberg.hadoop.HadoopCatalog;
+import org.apache.iceberg.hive.HiveCatalogs;
+import org.apache.iceberg.io.FileIO;
+import org.apache.iceberg.io.OutputFile;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * This operator commit data files to Iceberg table.
+ * <p>
+ * This operator should always run with parallelism of 1.
+ * Because Iceberg lib perform optimistic concurrency control,
+ * this can help reduce contention and retries
+ * when committing files to Iceberg table.
+ * <p>
+ * Here are some known contentions
+ * 1) Flink jobs running in multiple regions,
+ * since Iceberg metadata service and commit only happens in us-east-1.
+ * 2) auto tuning and auto lift services may update Iceberg table infrequently.
+ */
+@SuppressWarnings("checkstyle:HiddenField")
+public class IcebergCommitter extends RichSinkFunction<FlinkDataFile>

Review comment:
       yeah. OperatorCoordinator seems like the better way to go. One drawback of the old two-stage operators model is that it can invalidate an embarrassingly parallel DAG. I brought this up to Becket Qin. This is the reverse/mirroring problem of FLIP-27 (source interface redesign). glad that we can use the OperatorCoordinator for sink too. It will probably be a relative big refactoring. might be better to track it in a separate issue.




----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org



---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org

[GitHub] [incubator-iceberg] stevenzwu commented on a change in pull request #856: [WIP] Flink Iceberg sink

[GitBox <gi...@apache.org>]

stevenzwu commented on a change in pull request #856:
URL: https://github.com/apache/incubator-iceberg/pull/856#discussion_r420917740



##########
File path: flink/src/main/java/org/apache/iceberg/flink/connector/sink/IcebergCommitter.java
##########
@@ -0,0 +1,642 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iceberg.flink.connector.sink;
+
+import com.google.common.base.Joiner;
+import com.google.common.base.Strings;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.stream.Collectors;
+import javax.annotation.Nullable;
+import org.apache.flink.annotation.VisibleForTesting;
+import org.apache.flink.api.common.JobExecutionResult;
+import org.apache.flink.api.common.JobID;
+import org.apache.flink.api.common.state.ListState;
+import org.apache.flink.api.common.state.ListStateDescriptor;
+import org.apache.flink.api.java.ExecutionEnvironment;
+import org.apache.flink.configuration.CheckpointingOptions;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.runtime.state.CheckpointListener;
+import org.apache.flink.runtime.state.FunctionInitializationContext;
+import org.apache.flink.runtime.state.FunctionSnapshotContext;
+import org.apache.flink.streaming.api.checkpoint.CheckpointedFunction;
+import org.apache.flink.streaming.api.environment.CheckpointConfig;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
+import org.apache.flink.util.Preconditions;
+import org.apache.hadoop.hive.conf.HiveConf.ConfVars;
+import org.apache.iceberg.AppendFiles;
+import org.apache.iceberg.DataFile;
+import org.apache.iceberg.ManifestFile;
+import org.apache.iceberg.ManifestWriter;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.Snapshot;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.Transaction;
+import org.apache.iceberg.UpdateProperties;
+import org.apache.iceberg.catalog.Catalog;
+import org.apache.iceberg.catalog.TableIdentifier;
+import org.apache.iceberg.flink.connector.IcebergConnectorConstant;
+import org.apache.iceberg.flink.connector.model.CommitMetadata;
+import org.apache.iceberg.flink.connector.model.CommitMetadataUtil;
+import org.apache.iceberg.flink.connector.model.FlinkManifestFile;
+import org.apache.iceberg.flink.connector.model.FlinkManifestFileUtil;
+import org.apache.iceberg.flink.connector.model.GenericFlinkManifestFile;
+import org.apache.iceberg.flink.connector.model.ManifestFileState;
+import org.apache.iceberg.hadoop.HadoopCatalog;
+import org.apache.iceberg.hive.HiveCatalogs;
+import org.apache.iceberg.io.FileIO;
+import org.apache.iceberg.io.OutputFile;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * This operator commit data files to Iceberg table.
+ * <p>
+ * This operator should always run with parallelism of 1.
+ * Because Iceberg lib perform optimistic concurrency control,
+ * this can help reduce contention and retries
+ * when committing files to Iceberg table.
+ * <p>
+ * Here are some known contentions
+ * 1) Flink jobs running in multiple regions,
+ * since Iceberg metadata service and commit only happens in us-east-1.
+ * 2) auto tuning and auto lift services may update Iceberg table infrequently.
+ */
+@SuppressWarnings("checkstyle:HiddenField")
+public class IcebergCommitter extends RichSinkFunction<FlinkDataFile>
+    implements CheckpointedFunction, CheckpointListener {
+  private static final Logger LOG = LoggerFactory.getLogger(IcebergCommitter.class);
+
+  private static final String COMMIT_REGION_KEY = "flink.commit.region";
+  private static final String COMMIT_MANIFEST_HASHES_KEY = "flink.commit.manifest.hashes";
+  private static final String VTTS_WATERMARK_PROP_KEY_PREFIX = "flink.watermark.";
+
+  private Configuration config;
+  private final String namespace;
+  private final String tableName;
+
+  private final String region;
+  private final boolean vttsWatermarkEnabled;
+  private final String vttsWatermarkPropKey;
+  private final long snapshotRetentionHours;
+  private final boolean commitRestoredManifestFiles;
+  private final String icebergManifestFileDir;
+  private final PartitionSpec spec;
+  private final FileIO io;
+  private final String flinkJobId;
+
+  private transient Table table;
+  private transient List<FlinkDataFile> pendingDataFiles;
+  private transient List<FlinkManifestFile> flinkManifestFiles;
+  private transient ListState<ManifestFileState> manifestFilesState;
+  private transient CommitMetadata metadata;
+  private transient ListState<CommitMetadata> metadataState;
+
+  public IcebergCommitter(Table table, Configuration config) {
+    this.config = config;
+
+    // current Iceberg sink implementation can't work with concurrent checkpoints.
+    // We disable concurrent checkpoints by default as min pause is set to 60s by default.
+    // Add an assertion to fail explicit in case job enables concurrent checkpoints.
+    CheckpointConfig checkpointConfig = StreamExecutionEnvironment.getExecutionEnvironment().getCheckpointConfig();
+    if (checkpointConfig.getMaxConcurrentCheckpoints() > 1) {
+      throw new IllegalArgumentException("Iceberg sink doesn't support concurrent checkpoints");
+    }
+
+    namespace = config.getString(IcebergConnectorConstant.NAMESPACE, "");
+    tableName = config.getString(IcebergConnectorConstant.TABLE, "");
+
+    String region = System.getenv("EC2_REGION");
+    if (Strings.isNullOrEmpty(region)) {
+      region = "us-east-1";
+      LOG.info("Iceberg committer {}.{} default region to us-east-1", namespace, tableName);
+    }
+    this.region = region;
+
+    vttsWatermarkEnabled = !Strings.isNullOrEmpty(
+        config.getString(IcebergConnectorConstant.VTTS_WATERMARK_TIMESTAMP_FIELD, ""));
+    vttsWatermarkPropKey = VTTS_WATERMARK_PROP_KEY_PREFIX + region;
+    snapshotRetentionHours = config.getLong(IcebergConnectorConstant.SNAPSHOT_RETENTION_HOURS,
+        IcebergConnectorConstant.DEFAULT_SNAPSHOT_RETENTION_HOURS);
+    commitRestoredManifestFiles = config.getBoolean(IcebergConnectorConstant.COMMIT_RESTORED_MANIFEST_FILES,
+        IcebergConnectorConstant.DEFAULT_COMMIT_RESTORED_MANIFEST_FILES);
+    icebergManifestFileDir = getIcebergManifestFileDir(config);
+
+    // The only final fields yielded by table inputted
+    spec = table.spec();
+    io = table.io();
+
+    final JobExecutionResult jobExecutionResult
+        = ExecutionEnvironment.getExecutionEnvironment().getLastJobExecutionResult();
+    if (jobExecutionResult != null) {
+      flinkJobId = jobExecutionResult.getJobID().toString();
+      LOG.info("Get Flink job ID from execution environment: {}", flinkJobId);
+    } else {
+      flinkJobId = new JobID().toString();
+      LOG.info("Execution environment doesn't have executed job. Generate a random job ID : {}", flinkJobId);
+    }
+    LOG.info("Iceberg committer {}.{} created with sink config", namespace, tableName);
+    LOG.info("Iceberg committer {}.{} loaded table partition spec: {}", namespace, tableName, spec);
+  }
+
+  @VisibleForTesting
+  List<FlinkDataFile> getPendingDataFiles() {
+    return pendingDataFiles;
+  }
+
+  @VisibleForTesting
+  List<FlinkManifestFile> getFlinkManifestFiles() {
+    return flinkManifestFiles;
+  }
+
+  @VisibleForTesting
+  CommitMetadata getMetadata() {
+    return metadata;
+  }
+
+  private String getIcebergManifestFileDir(Configuration config) {
+    final String checkpointDir = config.getString(
+        CheckpointingOptions.CHECKPOINTS_DIRECTORY, null);
+    if (null == checkpointDir) {
+      throw new IllegalArgumentException("checkpoint dir is null");
+    }
+
+    return String.format("%s/iceberg/manifest/", checkpointDir);
+  }
+
+  @Override
+  public void close() throws Exception {
+    super.close();
+  }
+
+  void init() {
+    // TODO: duplicate logic, to extract
+    org.apache.hadoop.conf.Configuration hadoopConf = new org.apache.hadoop.conf.Configuration();
+    String catalogType = config.getString(IcebergConnectorConstant.CATALOG_TYPE,
+                                          IcebergConnectorConstant.CATALOG_TYPE_DEFAULT);
+    Catalog catalog = null;
+    switch (catalogType.toUpperCase()) {
+      case IcebergConnectorConstant.HIVE_CATALOG:
+        hadoopConf.set(ConfVars.METASTOREURIS.varname, config.getString(ConfVars.METASTOREURIS.varname, ""));
+        catalog = HiveCatalogs.loadCatalog(hadoopConf);
+        break;
+
+      case IcebergConnectorConstant.HADOOP_CATALOG:
+        catalog = new HadoopCatalog(hadoopConf,
+                                    config.getString(IcebergConnectorConstant.HADOOP_CATALOG_WAREHOUSE_LOCATION, ""));
+        break;
+
+      default:
+        throw new UnsupportedOperationException("Unknown catalog type or not set: " + catalogType);
+    }
+
+    this.table = catalog.loadTable(TableIdentifier.parse(namespace + "." + tableName));
+
+    pendingDataFiles = new ArrayList<>();
+    flinkManifestFiles = new ArrayList<>();
+    metadata = CommitMetadata.newBuilder()
+        .setLastCheckpointId(0)
+        .setLastCheckpointTimestamp(0)
+        .setLastCommitTimestamp(System.currentTimeMillis())
+        .build();
+  }
+
+  @Override
+  public void initializeState(FunctionInitializationContext context) throws Exception {
+    init();
+
+    Preconditions.checkState(manifestFilesState == null,
+        "checkpointedFilesState has already been initialized.");
+    Preconditions.checkState(metadataState == null,
+        "metadataState has already been initialized.");
+    manifestFilesState = context.getOperatorStateStore().getListState(new ListStateDescriptor<>(
+        "iceberg-committer-manifest-files-state", ManifestFileState.class));
+    metadataState = context.getOperatorStateStore().getListState(new ListStateDescriptor<>(
+        "iceberg-committer-metadata-state", CommitMetadata.class));
+
+    if (context.isRestored()) {
+      final Iterable<CommitMetadata> restoredMetadata = metadataState.get();
+      if (null != restoredMetadata) {
+        LOG.info("Iceberg committer {}.{} restoring metadata", namespace, tableName);
+        List<CommitMetadata> metadataList = new ArrayList<>();
+        for (CommitMetadata entry : restoredMetadata) {
+          metadataList.add(entry);
+        }
+        Preconditions.checkState(1 == metadataList.size(),
+            "metadata list size should be 1. got " + metadataList.size());
+        metadata = metadataList.get(0);
+        LOG.info("Iceberg committer {}.{} restored metadata: {}",
+            namespace, tableName, CommitMetadataUtil.getInstance().encodeAsJson(metadata));
+      } else {
+        LOG.info("Iceberg committer {}.{} has nothing to restore for metadata", namespace, tableName);
+      }
+
+      Iterable<ManifestFileState> restoredManifestFileStates = manifestFilesState.get();
+      if (null != restoredManifestFileStates) {
+        LOG.info("Iceberg committer {}.{} restoring manifest files",
+            namespace, tableName);
+        for (ManifestFileState manifestFileState : restoredManifestFileStates) {
+          flinkManifestFiles.add(GenericFlinkManifestFile.fromState(manifestFileState));
+        }
+        LOG.info("Iceberg committer {}.{} restored {} manifest files: {}",
+            namespace, tableName, flinkManifestFiles.size(), flinkManifestFiles);
+        final long now = System.currentTimeMillis();
+        if (now - metadata.getLastCheckpointTimestamp() > TimeUnit.HOURS.toMillis(snapshotRetentionHours)) {
+          flinkManifestFiles.clear();
+          LOG.info("Iceberg committer {}.{} cleared restored manifest files as checkpoint timestamp is too old: " +
+                  "checkpointTimestamp = {}, now = {}, snapshotRetentionHours = {}",
+              namespace, tableName, metadata.getLastCheckpointTimestamp(), now, snapshotRetentionHours);
+        } else {
+          flinkManifestFiles = removeCommittedManifests(flinkManifestFiles);
+          if (flinkManifestFiles.isEmpty()) {
+            LOG.info("Iceberg committer {}.{} has zero uncommitted manifest files from restored state",
+                namespace, tableName);
+          } else {
+            if (commitRestoredManifestFiles) {
+              commitRestoredManifestFiles();
+            } else {
+              LOG.info("skip commit of restored manifest files");
+            }
+          }
+        }
+      } else {
+        LOG.info("Iceberg committer {}.{} has nothing to restore for manifest files", namespace, tableName);
+      }
+    }
+  }
+
+  @VisibleForTesting
+  void commitRestoredManifestFiles() throws Exception {
+    LOG.info("Iceberg committer {}.{} committing last uncompleted transaction upon recovery: " +
+            "metadata = {}, flink manifest files ({}) = {}", namespace, tableName,
+        CommitMetadataUtil.getInstance().encodeAsJson(metadata),
+        flinkManifestFiles.size(), flinkManifestFiles);
+    commit();
+    LOG.info("Iceberg committer {}.{} committed last uncompleted transaction upon recovery: " +
+            "metadata = {}, flink manifest files ({}) = {}", namespace, tableName,
+        CommitMetadataUtil.getInstance().encodeAsJson(metadata),
+        flinkManifestFiles.size(), flinkManifestFiles);
+    postCommitSuccess();
+  }
+
+  private List<FlinkManifestFile> removeCommittedManifests(List<FlinkManifestFile> flinkManifestFiles) {
+    int snapshotCount = 0;
+    String result = "succeeded";
+    final long start = System.currentTimeMillis();
+    try {
+      final Set<String> manifestHashes = flinkManifestFiles.stream()
+          .map(f -> f.hash())
+          .collect(Collectors.toSet());
+      final Set<String> committedHashes = new HashSet<>(flinkManifestFiles.size());
+      final Iterable<Snapshot> snapshots = table.snapshots();
+      for (Snapshot snapshot : snapshots) {
+        ++snapshotCount;
+        final Map<String, String> summary = snapshot.summary();
+        final List<String> hashes = FlinkManifestFileUtil.hashesStringToList(summary.get(COMMIT_MANIFEST_HASHES_KEY));
+        for (String hash : hashes) {
+          if (manifestHashes.contains(hash)) {
+            committedHashes.add(hash);
+          }
+        }
+      }
+      final List<FlinkManifestFile> uncommittedManifestFiles = flinkManifestFiles.stream()
+          .filter(f -> !committedHashes.contains(f.hash()))
+          .collect(Collectors.toList());
+      return uncommittedManifestFiles;
+    } catch (Throwable t) {
+      result = "failed";
+      //LOG.error(String.format("Iceberg committer %s.%s failed to check transaction completed", database, tableName),
+      //          t);
+      LOG.error("Iceberg committer {}.{} failed to check transaction completed. Throwable = {}",
+                namespace, tableName, t);
+      throw t;
+    } finally {
+      final long duration = System.currentTimeMillis() - start;
+      LOG.info("Iceberg committer {}.{} {} to check transaction completed" +
+               " after iterating {} snapshots and {} milli-seconds",
+               namespace, tableName, result, snapshotCount, duration);
+    }
+  }
+
+  @Override
+  public void snapshotState(FunctionSnapshotContext context) throws Exception {
+    LOG.info("Iceberg committer {}.{} snapshot state: checkpointId = {}, triggerTime = {}",
+        namespace, tableName, context.getCheckpointId(), context.getCheckpointTimestamp());
+    Preconditions.checkState(null != manifestFilesState,
+        "manifest files state has not been properly initialized.");
+    Preconditions.checkState(null != metadataState,
+        "metadata state has not been properly initialized.");
+
+    // set transaction to null to indicate a start of a new checkpoint/commit/transaction
+    synchronized (this) {
+      snapshot(context, pendingDataFiles);
+      checkpointState(flinkManifestFiles, metadata);
+      postSnapshotSuccess();
+    }
+  }
+
+  @VisibleForTesting
+  void snapshot(FunctionSnapshotContext context, List<FlinkDataFile> pendingDataFiles) throws Exception {
+    FlinkManifestFile flinkManifestFile = null;
+    if (!pendingDataFiles.isEmpty()) {
+      flinkManifestFile = createManifestFile(context, pendingDataFiles);
+      flinkManifestFiles.add(flinkManifestFile);
+    }
+    metadata = updateMetadata(metadata, context, flinkManifestFile);
+  }
+
+  private FlinkManifestFile createManifestFile(
+      FunctionSnapshotContext context, List<FlinkDataFile> pendingDataFiles) throws Exception {
+    LOG.info("Iceberg committer {}.{} checkpointing {} pending data files}",
+        namespace, tableName, pendingDataFiles.size());
+    String result = "succeeded";
+    final long start = System.currentTimeMillis();
+    try {
+      final String manifestFileName = Joiner.on("_")
+          .join(flinkJobId, context.getCheckpointId(), context.getCheckpointTimestamp());
+      // Iceberg requires file format suffix right now
+      final String manifestFileNameWithSuffix = manifestFileName + ".avro";
+      OutputFile outputFile = io.newOutputFile(icebergManifestFileDir + manifestFileNameWithSuffix);
+      ManifestWriter manifestWriter = ManifestWriter.write(spec, outputFile);
+
+      // stats
+      long recordCount = 0;
+      long byteCount = 0;
+      long lowWatermark = Long.MAX_VALUE;
+      long highWatermark = Long.MIN_VALUE;
+      for (FlinkDataFile flinkDataFile : pendingDataFiles) {
+        DataFile dataFile = flinkDataFile.getIcebergDataFile();
+        manifestWriter.add(dataFile);
+        // update stas
+        recordCount += dataFile.recordCount();
+        byteCount += dataFile.fileSizeInBytes();
+        if (flinkDataFile.getLowWatermark() < lowWatermark) {
+          lowWatermark = flinkDataFile.getLowWatermark();
+        }
+        if (flinkDataFile.getHighWatermark() > highWatermark) {
+          highWatermark = flinkDataFile.getHighWatermark();
+        }
+        LOG.debug("Data file with size of {} bytes added to manifest", dataFile.fileSizeInBytes());
+      }
+      manifestWriter.close();
+      ManifestFile manifestFile = manifestWriter.toManifestFile();
+
+      FlinkManifestFile flinkManifestFile = GenericFlinkManifestFile.builder()
+          .setPath(manifestFile.path())
+          .setLength(manifestFile.length())
+          .setSpecId(manifestFile.partitionSpecId())
+          .setCheckpointId(context.getCheckpointId())
+          .setCheckpointTimestamp(context.getCheckpointTimestamp())
+          .setDataFileCount(pendingDataFiles.size())
+          .setRecordCount(recordCount)
+          .setByteCount(byteCount)
+          .setLowWatermark(lowWatermark)
+          .setHighWatermark(highWatermark)
+          .build();
+
+      // don't want to log a giant list at one line.
+      // split the complete list into smaller chunks with 50 files.
+      final AtomicInteger counter = new AtomicInteger(0);
+      Collection<List<String>> listOfFileList = pendingDataFiles.stream()
+          .map(flinkDataFile -> flinkDataFile.toCompactDump())
+          .collect(Collectors.groupingBy(it -> counter.getAndIncrement() / 50))
+          .values();
+      for (List<String> fileList : listOfFileList) {
+        LOG.info("Iceberg committer {}.{} created manifest file {} for {}/{} data files: {}",
+            namespace, tableName, manifestFile.path(), fileList.size(), pendingDataFiles.size(), fileList);
+      }
+      return flinkManifestFile;
+    } catch (Throwable t) {
+      result = "failed";
+      //LOG.error(String.format("Iceberg committer %s.%s failed to create manifest file for %d pending data files",
+      //          database, tableName, pendingDataFiles.size()), t);
+      LOG.error("Iceberg committer {}.{} failed to create manifest file for {} pending data files. Throwable={}",
+          namespace, tableName, pendingDataFiles.size(), t);
+      throw t;
+    } finally {
+      final long duration = System.currentTimeMillis() - start;
+      LOG.info("Iceberg committer {}.{} {} to create manifest file with {} data files after {} milli-seconds",
+          namespace, tableName, result, pendingDataFiles.size(), duration);
+    }
+  }
+
+  private CommitMetadata updateMetadata(
+      CommitMetadata oldMetadata,
+      FunctionSnapshotContext context,
+      @Nullable FlinkManifestFile flinkManifestFile) {
+    LOG.info("Iceberg committer {}.{} updating metadata {} with manifest file {}",
+        namespace, tableName, CommitMetadataUtil.getInstance().encodeAsJson(oldMetadata), flinkManifestFile);
+    CommitMetadata.Builder metadataBuilder = CommitMetadata.newBuilder(oldMetadata)
+        .setLastCheckpointId(context.getCheckpointId())
+        .setLastCheckpointTimestamp(context.getCheckpointTimestamp());
+    if (vttsWatermarkEnabled) {
+      Long vttsWatermark = oldMetadata.getVttsWatermark();
+      if (null == flinkManifestFile) {
+        // DPS-412: when there is no data to be committed
+        // use elapsed wall clock time to move the VTTS watermark forward.
+        if (null != vttsWatermark) {
+          final long elapsedTimeMs = System.currentTimeMillis() - oldMetadata.getLastCommitTimestamp();
+          vttsWatermark += elapsedTimeMs;
+        } else {
+          vttsWatermark = System.currentTimeMillis();
+        }
+      } else {
+        // use lowWatermark.
+        if (null == flinkManifestFile.lowWatermark()) {
+          throw new IllegalArgumentException("VTTS is enabled but lowWatermark is null");
+        }
+        // in case one container/slot is lagging behind,
+        // we want to move watermark forward based on the slowest.
+        final long newWatermark = flinkManifestFile.lowWatermark();
+        // make sure VTTS watermark doesn't go back in time
+        if (null == vttsWatermark || newWatermark > vttsWatermark) {
+          vttsWatermark = newWatermark;
+        }
+      }
+      metadataBuilder.setVttsWatermark(vttsWatermark);
+    }
+    CommitMetadata metadata = metadataBuilder.build();
+    LOG.info("Iceberg committer {}.{} updated metadata {} with manifest file {}",
+        namespace, tableName, CommitMetadataUtil.getInstance().encodeAsJson(metadata), flinkManifestFile);
+    return metadata;
+  }
+
+  private void checkpointState(List<FlinkManifestFile> flinkManifestFiles, CommitMetadata metadata) throws Exception {
+    LOG.info("Iceberg committer {}.{} checkpointing state", namespace, tableName);
+    List<ManifestFileState> manifestFileStates = flinkManifestFiles.stream()
+        .map(f -> f.toState())
+        .collect(Collectors.toList());
+    manifestFilesState.clear();
+    manifestFilesState.addAll(manifestFileStates);
+    metadataState.clear();
+    metadataState.add(metadata);
+    LOG.info("Iceberg committer {}.{} checkpointed state: metadata = {}, flinkManifestFiles({}) = {}",
+        namespace, tableName, CommitMetadataUtil.getInstance().encodeAsJson(metadata),
+        flinkManifestFiles.size(), flinkManifestFiles);
+  }
+
+  private void postSnapshotSuccess() {
+    pendingDataFiles.clear();
+    LOG.debug("Un-committed manifest file count {}, containing data file count {}, record count {} and byte count {}",
+        flinkManifestFiles.size(),
+        FlinkManifestFileUtil.getDataFileCount(flinkManifestFiles),
+        FlinkManifestFileUtil.getRecordCount(flinkManifestFiles),
+        FlinkManifestFileUtil.getByteCount(flinkManifestFiles)
+    );
+  }
+
+  @Override
+  public void notifyCheckpointComplete(long checkpointId) {
+    LOG.info("Iceberg committer {}.{} checkpoint {} completed", namespace, tableName, checkpointId);
+    synchronized (this) {
+      if (checkpointId == metadata.getLastCheckpointId()) {
+        try {
+          commit();
+          postCommitSuccess();
+        } catch (Exception t) {
+          // swallow the exception to avoid job restart in case of commit failure
+          //LOG.error(String.format("Iceberg committer %s.%s failed to do post checkpoint commit", database, tableName),
+          //          t);
+          LOG.error("Iceberg committer {}.{} failed to do post checkpoint commit. Throwable = ",
+              namespace, tableName, t);
+        }
+      } else {
+        // TODO: it would be nice to fix this and allow concurrent checkpoint
+        LOG.info("Iceberg committer {}.{} skip committing transaction: " +

Review comment:
       yeah. it is a problem if we allow concurrent checkpoint, which we never use at Netflix.




----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org



---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org

[GitHub] [incubator-iceberg] waterlx commented on a change in pull request #856: [WIP] Flink Iceberg sink

[GitBox <gi...@apache.org>]

waterlx commented on a change in pull request #856:
URL: https://github.com/apache/incubator-iceberg/pull/856#discussion_r419200546



##########
File path: flink/src/main/java/org/apache/iceberg/flink/connector/sink/RecordPartitioner.java
##########
@@ -110,7 +110,7 @@ public void partition(Record record) {
     if (isOptional) {
       // the wrapped position handles null layers
       return new WrappedPositionAccessor(p, accessor);
-    } else if (accessor instanceof PositionAccessor) {
+    } else if (accessor != null && accessor.getClass() == PositionAccessor.class) {

Review comment:
       This is a temporary fix. Please check https://github.com/apache/incubator-iceberg/issues/994 or its PR.




----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org



---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org

[GitHub] [incubator-iceberg] waterlx edited a comment on pull request #856: [WIP] Flink Iceberg sink

[GitBox <gi...@apache.org>]

waterlx edited a comment on pull request #856:
URL: https://github.com/apache/incubator-iceberg/pull/856#issuecomment-617074928


   @rdblue Thanks for sharing your thoughts on the reasons why Table/BaseTable is not serializable. Totally agree. But I am currently in the dilemma where there might be a need to call high-level operations in Flink tasks,  like table.newTransaction() when trying to commit DataFiles accumulated from the streaming inputs. Currently code limits the parallelism to 1 so that the commit won't be performed in parallel.  
   
   For now, it is not a blocker because I could pass the namespace and table name by config and call loadTable() of Catalog to build the table when there is a need. But the implementation is not that good as the table informations(like namespace, table name, it is HiveCatalog or HadoopCatalog) passes eveywhere, while some of them are not needed.
   
   I am also thinking about passing the path as a string (db.table for Hive Catalog while full qualified path for HadoopTables) instead of passing table instance, but the purpose is also to re-build the table instance so as to call some high-level operations.


----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org



---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org

[GitHub] [incubator-iceberg] Jiayi-Liao commented on a change in pull request #856: [WIP] Flink Iceberg sink

[GitBox <gi...@apache.org>]

Jiayi-Liao commented on a change in pull request #856:
URL: https://github.com/apache/incubator-iceberg/pull/856#discussion_r426225073



##########
File path: flink/src/main/java/org/apache/iceberg/flink/connector/sink/IcebergCommitter.java
##########
@@ -0,0 +1,642 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iceberg.flink.connector.sink;
+
+import com.google.common.base.Joiner;
+import com.google.common.base.Strings;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.stream.Collectors;
+import javax.annotation.Nullable;
+import org.apache.flink.annotation.VisibleForTesting;
+import org.apache.flink.api.common.JobExecutionResult;
+import org.apache.flink.api.common.JobID;
+import org.apache.flink.api.common.state.ListState;
+import org.apache.flink.api.common.state.ListStateDescriptor;
+import org.apache.flink.api.java.ExecutionEnvironment;
+import org.apache.flink.configuration.CheckpointingOptions;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.runtime.state.CheckpointListener;
+import org.apache.flink.runtime.state.FunctionInitializationContext;
+import org.apache.flink.runtime.state.FunctionSnapshotContext;
+import org.apache.flink.streaming.api.checkpoint.CheckpointedFunction;
+import org.apache.flink.streaming.api.environment.CheckpointConfig;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
+import org.apache.flink.util.Preconditions;
+import org.apache.hadoop.hive.conf.HiveConf.ConfVars;
+import org.apache.iceberg.AppendFiles;
+import org.apache.iceberg.DataFile;
+import org.apache.iceberg.ManifestFile;
+import org.apache.iceberg.ManifestWriter;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.Snapshot;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.Transaction;
+import org.apache.iceberg.UpdateProperties;
+import org.apache.iceberg.catalog.Catalog;
+import org.apache.iceberg.catalog.TableIdentifier;
+import org.apache.iceberg.flink.connector.IcebergConnectorConstant;
+import org.apache.iceberg.flink.connector.model.CommitMetadata;
+import org.apache.iceberg.flink.connector.model.CommitMetadataUtil;
+import org.apache.iceberg.flink.connector.model.FlinkManifestFile;
+import org.apache.iceberg.flink.connector.model.FlinkManifestFileUtil;
+import org.apache.iceberg.flink.connector.model.GenericFlinkManifestFile;
+import org.apache.iceberg.flink.connector.model.ManifestFileState;
+import org.apache.iceberg.hadoop.HadoopCatalog;
+import org.apache.iceberg.hive.HiveCatalogs;
+import org.apache.iceberg.io.FileIO;
+import org.apache.iceberg.io.OutputFile;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * This operator commit data files to Iceberg table.
+ * <p>
+ * This operator should always run with parallelism of 1.
+ * Because Iceberg lib perform optimistic concurrency control,
+ * this can help reduce contention and retries
+ * when committing files to Iceberg table.
+ * <p>
+ * Here are some known contentions
+ * 1) Flink jobs running in multiple regions,
+ * since Iceberg metadata service and commit only happens in us-east-1.
+ * 2) auto tuning and auto lift services may update Iceberg table infrequently.
+ */
+@SuppressWarnings("checkstyle:HiddenField")
+public class IcebergCommitter extends RichSinkFunction<FlinkDataFile>
+    implements CheckpointedFunction, CheckpointListener {
+  private static final Logger LOG = LoggerFactory.getLogger(IcebergCommitter.class);
+
+  private static final String COMMIT_REGION_KEY = "flink.commit.region";
+  private static final String COMMIT_MANIFEST_HASHES_KEY = "flink.commit.manifest.hashes";
+  private static final String VTTS_WATERMARK_PROP_KEY_PREFIX = "flink.watermark.";
+
+  private Configuration config;
+  private final String namespace;
+  private final String tableName;
+
+  private final String region;
+  private final boolean vttsWatermarkEnabled;
+  private final String vttsWatermarkPropKey;
+  private final long snapshotRetentionHours;
+  private final boolean commitRestoredManifestFiles;
+  private final String icebergManifestFileDir;
+  private final PartitionSpec spec;
+  private final FileIO io;
+  private final String flinkJobId;
+
+  private transient Table table;
+  private transient List<FlinkDataFile> pendingDataFiles;
+  private transient List<FlinkManifestFile> flinkManifestFiles;
+  private transient ListState<ManifestFileState> manifestFilesState;
+  private transient CommitMetadata metadata;
+  private transient ListState<CommitMetadata> metadataState;
+
+  public IcebergCommitter(Table table, Configuration config) {
+    this.config = config;
+
+    // current Iceberg sink implementation can't work with concurrent checkpoints.
+    // We disable concurrent checkpoints by default as min pause is set to 60s by default.
+    // Add an assertion to fail explicit in case job enables concurrent checkpoints.
+    CheckpointConfig checkpointConfig = StreamExecutionEnvironment.getExecutionEnvironment().getCheckpointConfig();
+    if (checkpointConfig.getMaxConcurrentCheckpoints() > 1) {
+      throw new IllegalArgumentException("Iceberg sink doesn't support concurrent checkpoints");
+    }
+
+    namespace = config.getString(IcebergConnectorConstant.NAMESPACE, "");
+    tableName = config.getString(IcebergConnectorConstant.TABLE, "");
+
+    String region = System.getenv("EC2_REGION");
+    if (Strings.isNullOrEmpty(region)) {
+      region = "us-east-1";
+      LOG.info("Iceberg committer {}.{} default region to us-east-1", namespace, tableName);
+    }
+    this.region = region;
+
+    vttsWatermarkEnabled = !Strings.isNullOrEmpty(
+        config.getString(IcebergConnectorConstant.VTTS_WATERMARK_TIMESTAMP_FIELD, ""));
+    vttsWatermarkPropKey = VTTS_WATERMARK_PROP_KEY_PREFIX + region;
+    snapshotRetentionHours = config.getLong(IcebergConnectorConstant.SNAPSHOT_RETENTION_HOURS,
+        IcebergConnectorConstant.DEFAULT_SNAPSHOT_RETENTION_HOURS);
+    commitRestoredManifestFiles = config.getBoolean(IcebergConnectorConstant.COMMIT_RESTORED_MANIFEST_FILES,
+        IcebergConnectorConstant.DEFAULT_COMMIT_RESTORED_MANIFEST_FILES);
+    icebergManifestFileDir = getIcebergManifestFileDir(config);
+
+    // The only final fields yielded by table inputted
+    spec = table.spec();
+    io = table.io();
+
+    final JobExecutionResult jobExecutionResult
+        = ExecutionEnvironment.getExecutionEnvironment().getLastJobExecutionResult();
+    if (jobExecutionResult != null) {
+      flinkJobId = jobExecutionResult.getJobID().toString();
+      LOG.info("Get Flink job ID from execution environment: {}", flinkJobId);
+    } else {
+      flinkJobId = new JobID().toString();
+      LOG.info("Execution environment doesn't have executed job. Generate a random job ID : {}", flinkJobId);
+    }
+    LOG.info("Iceberg committer {}.{} created with sink config", namespace, tableName);
+    LOG.info("Iceberg committer {}.{} loaded table partition spec: {}", namespace, tableName, spec);
+  }
+
+  @VisibleForTesting
+  List<FlinkDataFile> getPendingDataFiles() {
+    return pendingDataFiles;
+  }
+
+  @VisibleForTesting
+  List<FlinkManifestFile> getFlinkManifestFiles() {
+    return flinkManifestFiles;
+  }
+
+  @VisibleForTesting
+  CommitMetadata getMetadata() {
+    return metadata;
+  }
+
+  private String getIcebergManifestFileDir(Configuration config) {
+    final String checkpointDir = config.getString(
+        CheckpointingOptions.CHECKPOINTS_DIRECTORY, null);
+    if (null == checkpointDir) {
+      throw new IllegalArgumentException("checkpoint dir is null");
+    }
+
+    return String.format("%s/iceberg/manifest/", checkpointDir);
+  }
+
+  @Override
+  public void close() throws Exception {
+    super.close();
+  }
+
+  void init() {
+    // TODO: duplicate logic, to extract
+    org.apache.hadoop.conf.Configuration hadoopConf = new org.apache.hadoop.conf.Configuration();
+    String catalogType = config.getString(IcebergConnectorConstant.CATALOG_TYPE,
+                                          IcebergConnectorConstant.CATALOG_TYPE_DEFAULT);
+    Catalog catalog = null;
+    switch (catalogType.toUpperCase()) {
+      case IcebergConnectorConstant.HIVE_CATALOG:
+        hadoopConf.set(ConfVars.METASTOREURIS.varname, config.getString(ConfVars.METASTOREURIS.varname, ""));
+        catalog = HiveCatalogs.loadCatalog(hadoopConf);
+        break;
+
+      case IcebergConnectorConstant.HADOOP_CATALOG:
+        catalog = new HadoopCatalog(hadoopConf,
+                                    config.getString(IcebergConnectorConstant.HADOOP_CATALOG_WAREHOUSE_LOCATION, ""));
+        break;
+
+      default:
+        throw new UnsupportedOperationException("Unknown catalog type or not set: " + catalogType);
+    }
+
+    this.table = catalog.loadTable(TableIdentifier.parse(namespace + "." + tableName));
+
+    pendingDataFiles = new ArrayList<>();
+    flinkManifestFiles = new ArrayList<>();
+    metadata = CommitMetadata.newBuilder()
+        .setLastCheckpointId(0)
+        .setLastCheckpointTimestamp(0)
+        .setLastCommitTimestamp(System.currentTimeMillis())
+        .build();
+  }
+
+  @Override
+  public void initializeState(FunctionInitializationContext context) throws Exception {
+    init();
+
+    Preconditions.checkState(manifestFilesState == null,
+        "checkpointedFilesState has already been initialized.");
+    Preconditions.checkState(metadataState == null,
+        "metadataState has already been initialized.");
+    manifestFilesState = context.getOperatorStateStore().getListState(new ListStateDescriptor<>(
+        "iceberg-committer-manifest-files-state", ManifestFileState.class));
+    metadataState = context.getOperatorStateStore().getListState(new ListStateDescriptor<>(
+        "iceberg-committer-metadata-state", CommitMetadata.class));
+
+    if (context.isRestored()) {
+      final Iterable<CommitMetadata> restoredMetadata = metadataState.get();
+      if (null != restoredMetadata) {
+        LOG.info("Iceberg committer {}.{} restoring metadata", namespace, tableName);
+        List<CommitMetadata> metadataList = new ArrayList<>();
+        for (CommitMetadata entry : restoredMetadata) {
+          metadataList.add(entry);
+        }
+        Preconditions.checkState(1 == metadataList.size(),
+            "metadata list size should be 1. got " + metadataList.size());
+        metadata = metadataList.get(0);
+        LOG.info("Iceberg committer {}.{} restored metadata: {}",
+            namespace, tableName, CommitMetadataUtil.getInstance().encodeAsJson(metadata));
+      } else {
+        LOG.info("Iceberg committer {}.{} has nothing to restore for metadata", namespace, tableName);
+      }
+
+      Iterable<ManifestFileState> restoredManifestFileStates = manifestFilesState.get();
+      if (null != restoredManifestFileStates) {
+        LOG.info("Iceberg committer {}.{} restoring manifest files",
+            namespace, tableName);
+        for (ManifestFileState manifestFileState : restoredManifestFileStates) {
+          flinkManifestFiles.add(GenericFlinkManifestFile.fromState(manifestFileState));
+        }
+        LOG.info("Iceberg committer {}.{} restored {} manifest files: {}",
+            namespace, tableName, flinkManifestFiles.size(), flinkManifestFiles);
+        final long now = System.currentTimeMillis();
+        if (now - metadata.getLastCheckpointTimestamp() > TimeUnit.HOURS.toMillis(snapshotRetentionHours)) {
+          flinkManifestFiles.clear();
+          LOG.info("Iceberg committer {}.{} cleared restored manifest files as checkpoint timestamp is too old: " +
+                  "checkpointTimestamp = {}, now = {}, snapshotRetentionHours = {}",
+              namespace, tableName, metadata.getLastCheckpointTimestamp(), now, snapshotRetentionHours);
+        } else {
+          flinkManifestFiles = removeCommittedManifests(flinkManifestFiles);
+          if (flinkManifestFiles.isEmpty()) {
+            LOG.info("Iceberg committer {}.{} has zero uncommitted manifest files from restored state",
+                namespace, tableName);
+          } else {
+            if (commitRestoredManifestFiles) {
+              commitRestoredManifestFiles();
+            } else {
+              LOG.info("skip commit of restored manifest files");
+            }
+          }
+        }
+      } else {
+        LOG.info("Iceberg committer {}.{} has nothing to restore for manifest files", namespace, tableName);
+      }
+    }
+  }
+
+  @VisibleForTesting
+  void commitRestoredManifestFiles() throws Exception {
+    LOG.info("Iceberg committer {}.{} committing last uncompleted transaction upon recovery: " +
+            "metadata = {}, flink manifest files ({}) = {}", namespace, tableName,
+        CommitMetadataUtil.getInstance().encodeAsJson(metadata),
+        flinkManifestFiles.size(), flinkManifestFiles);
+    commit();
+    LOG.info("Iceberg committer {}.{} committed last uncompleted transaction upon recovery: " +
+            "metadata = {}, flink manifest files ({}) = {}", namespace, tableName,
+        CommitMetadataUtil.getInstance().encodeAsJson(metadata),
+        flinkManifestFiles.size(), flinkManifestFiles);
+    postCommitSuccess();
+  }
+
+  private List<FlinkManifestFile> removeCommittedManifests(List<FlinkManifestFile> flinkManifestFiles) {
+    int snapshotCount = 0;
+    String result = "succeeded";
+    final long start = System.currentTimeMillis();
+    try {
+      final Set<String> manifestHashes = flinkManifestFiles.stream()
+          .map(f -> f.hash())
+          .collect(Collectors.toSet());
+      final Set<String> committedHashes = new HashSet<>(flinkManifestFiles.size());
+      final Iterable<Snapshot> snapshots = table.snapshots();
+      for (Snapshot snapshot : snapshots) {
+        ++snapshotCount;
+        final Map<String, String> summary = snapshot.summary();
+        final List<String> hashes = FlinkManifestFileUtil.hashesStringToList(summary.get(COMMIT_MANIFEST_HASHES_KEY));
+        for (String hash : hashes) {
+          if (manifestHashes.contains(hash)) {
+            committedHashes.add(hash);
+          }
+        }
+      }
+      final List<FlinkManifestFile> uncommittedManifestFiles = flinkManifestFiles.stream()
+          .filter(f -> !committedHashes.contains(f.hash()))
+          .collect(Collectors.toList());
+      return uncommittedManifestFiles;
+    } catch (Throwable t) {
+      result = "failed";
+      //LOG.error(String.format("Iceberg committer %s.%s failed to check transaction completed", database, tableName),
+      //          t);
+      LOG.error("Iceberg committer {}.{} failed to check transaction completed. Throwable = {}",
+                namespace, tableName, t);
+      throw t;
+    } finally {
+      final long duration = System.currentTimeMillis() - start;
+      LOG.info("Iceberg committer {}.{} {} to check transaction completed" +
+               " after iterating {} snapshots and {} milli-seconds",
+               namespace, tableName, result, snapshotCount, duration);
+    }
+  }
+
+  @Override
+  public void snapshotState(FunctionSnapshotContext context) throws Exception {
+    LOG.info("Iceberg committer {}.{} snapshot state: checkpointId = {}, triggerTime = {}",
+        namespace, tableName, context.getCheckpointId(), context.getCheckpointTimestamp());
+    Preconditions.checkState(null != manifestFilesState,
+        "manifest files state has not been properly initialized.");
+    Preconditions.checkState(null != metadataState,
+        "metadata state has not been properly initialized.");
+
+    // set transaction to null to indicate a start of a new checkpoint/commit/transaction
+    synchronized (this) {
+      snapshot(context, pendingDataFiles);
+      checkpointState(flinkManifestFiles, metadata);
+      postSnapshotSuccess();
+    }
+  }
+
+  @VisibleForTesting
+  void snapshot(FunctionSnapshotContext context, List<FlinkDataFile> pendingDataFiles) throws Exception {
+    FlinkManifestFile flinkManifestFile = null;
+    if (!pendingDataFiles.isEmpty()) {
+      flinkManifestFile = createManifestFile(context, pendingDataFiles);
+      flinkManifestFiles.add(flinkManifestFile);
+    }
+    metadata = updateMetadata(metadata, context, flinkManifestFile);
+  }
+
+  private FlinkManifestFile createManifestFile(
+      FunctionSnapshotContext context, List<FlinkDataFile> pendingDataFiles) throws Exception {
+    LOG.info("Iceberg committer {}.{} checkpointing {} pending data files}",
+        namespace, tableName, pendingDataFiles.size());
+    String result = "succeeded";
+    final long start = System.currentTimeMillis();
+    try {
+      final String manifestFileName = Joiner.on("_")
+          .join(flinkJobId, context.getCheckpointId(), context.getCheckpointTimestamp());
+      // Iceberg requires file format suffix right now
+      final String manifestFileNameWithSuffix = manifestFileName + ".avro";
+      OutputFile outputFile = io.newOutputFile(icebergManifestFileDir + manifestFileNameWithSuffix);
+      ManifestWriter manifestWriter = ManifestWriter.write(spec, outputFile);
+
+      // stats
+      long recordCount = 0;
+      long byteCount = 0;
+      long lowWatermark = Long.MAX_VALUE;
+      long highWatermark = Long.MIN_VALUE;
+      for (FlinkDataFile flinkDataFile : pendingDataFiles) {
+        DataFile dataFile = flinkDataFile.getIcebergDataFile();
+        manifestWriter.add(dataFile);
+        // update stas
+        recordCount += dataFile.recordCount();
+        byteCount += dataFile.fileSizeInBytes();
+        if (flinkDataFile.getLowWatermark() < lowWatermark) {
+          lowWatermark = flinkDataFile.getLowWatermark();
+        }
+        if (flinkDataFile.getHighWatermark() > highWatermark) {
+          highWatermark = flinkDataFile.getHighWatermark();
+        }
+        LOG.debug("Data file with size of {} bytes added to manifest", dataFile.fileSizeInBytes());
+      }
+      manifestWriter.close();
+      ManifestFile manifestFile = manifestWriter.toManifestFile();
+
+      FlinkManifestFile flinkManifestFile = GenericFlinkManifestFile.builder()
+          .setPath(manifestFile.path())
+          .setLength(manifestFile.length())
+          .setSpecId(manifestFile.partitionSpecId())
+          .setCheckpointId(context.getCheckpointId())
+          .setCheckpointTimestamp(context.getCheckpointTimestamp())
+          .setDataFileCount(pendingDataFiles.size())
+          .setRecordCount(recordCount)
+          .setByteCount(byteCount)
+          .setLowWatermark(lowWatermark)
+          .setHighWatermark(highWatermark)
+          .build();
+
+      // don't want to log a giant list at one line.
+      // split the complete list into smaller chunks with 50 files.
+      final AtomicInteger counter = new AtomicInteger(0);
+      Collection<List<String>> listOfFileList = pendingDataFiles.stream()
+          .map(flinkDataFile -> flinkDataFile.toCompactDump())
+          .collect(Collectors.groupingBy(it -> counter.getAndIncrement() / 50))
+          .values();
+      for (List<String> fileList : listOfFileList) {
+        LOG.info("Iceberg committer {}.{} created manifest file {} for {}/{} data files: {}",
+            namespace, tableName, manifestFile.path(), fileList.size(), pendingDataFiles.size(), fileList);
+      }
+      return flinkManifestFile;
+    } catch (Throwable t) {
+      result = "failed";
+      //LOG.error(String.format("Iceberg committer %s.%s failed to create manifest file for %d pending data files",
+      //          database, tableName, pendingDataFiles.size()), t);
+      LOG.error("Iceberg committer {}.{} failed to create manifest file for {} pending data files. Throwable={}",
+          namespace, tableName, pendingDataFiles.size(), t);
+      throw t;
+    } finally {
+      final long duration = System.currentTimeMillis() - start;
+      LOG.info("Iceberg committer {}.{} {} to create manifest file with {} data files after {} milli-seconds",
+          namespace, tableName, result, pendingDataFiles.size(), duration);
+    }
+  }
+
+  private CommitMetadata updateMetadata(
+      CommitMetadata oldMetadata,
+      FunctionSnapshotContext context,
+      @Nullable FlinkManifestFile flinkManifestFile) {
+    LOG.info("Iceberg committer {}.{} updating metadata {} with manifest file {}",
+        namespace, tableName, CommitMetadataUtil.getInstance().encodeAsJson(oldMetadata), flinkManifestFile);
+    CommitMetadata.Builder metadataBuilder = CommitMetadata.newBuilder(oldMetadata)
+        .setLastCheckpointId(context.getCheckpointId())
+        .setLastCheckpointTimestamp(context.getCheckpointTimestamp());
+    if (vttsWatermarkEnabled) {
+      Long vttsWatermark = oldMetadata.getVttsWatermark();
+      if (null == flinkManifestFile) {
+        // DPS-412: when there is no data to be committed
+        // use elapsed wall clock time to move the VTTS watermark forward.
+        if (null != vttsWatermark) {
+          final long elapsedTimeMs = System.currentTimeMillis() - oldMetadata.getLastCommitTimestamp();
+          vttsWatermark += elapsedTimeMs;
+        } else {
+          vttsWatermark = System.currentTimeMillis();
+        }
+      } else {
+        // use lowWatermark.
+        if (null == flinkManifestFile.lowWatermark()) {
+          throw new IllegalArgumentException("VTTS is enabled but lowWatermark is null");
+        }
+        // in case one container/slot is lagging behind,
+        // we want to move watermark forward based on the slowest.
+        final long newWatermark = flinkManifestFile.lowWatermark();
+        // make sure VTTS watermark doesn't go back in time
+        if (null == vttsWatermark || newWatermark > vttsWatermark) {
+          vttsWatermark = newWatermark;
+        }
+      }
+      metadataBuilder.setVttsWatermark(vttsWatermark);
+    }
+    CommitMetadata metadata = metadataBuilder.build();
+    LOG.info("Iceberg committer {}.{} updated metadata {} with manifest file {}",
+        namespace, tableName, CommitMetadataUtil.getInstance().encodeAsJson(metadata), flinkManifestFile);
+    return metadata;
+  }
+
+  private void checkpointState(List<FlinkManifestFile> flinkManifestFiles, CommitMetadata metadata) throws Exception {
+    LOG.info("Iceberg committer {}.{} checkpointing state", namespace, tableName);
+    List<ManifestFileState> manifestFileStates = flinkManifestFiles.stream()
+        .map(f -> f.toState())
+        .collect(Collectors.toList());
+    manifestFilesState.clear();
+    manifestFilesState.addAll(manifestFileStates);
+    metadataState.clear();
+    metadataState.add(metadata);
+    LOG.info("Iceberg committer {}.{} checkpointed state: metadata = {}, flinkManifestFiles({}) = {}",
+        namespace, tableName, CommitMetadataUtil.getInstance().encodeAsJson(metadata),
+        flinkManifestFiles.size(), flinkManifestFiles);
+  }
+
+  private void postSnapshotSuccess() {
+    pendingDataFiles.clear();
+    LOG.debug("Un-committed manifest file count {}, containing data file count {}, record count {} and byte count {}",
+        flinkManifestFiles.size(),
+        FlinkManifestFileUtil.getDataFileCount(flinkManifestFiles),
+        FlinkManifestFileUtil.getRecordCount(flinkManifestFiles),
+        FlinkManifestFileUtil.getByteCount(flinkManifestFiles)
+    );
+  }
+
+  @Override
+  public void notifyCheckpointComplete(long checkpointId) {
+    LOG.info("Iceberg committer {}.{} checkpoint {} completed", namespace, tableName, checkpointId);
+    synchronized (this) {
+      if (checkpointId == metadata.getLastCheckpointId()) {
+        try {
+          commit();
+          postCommitSuccess();
+        } catch (Exception t) {
+          // swallow the exception to avoid job restart in case of commit failure
+          //LOG.error(String.format("Iceberg committer %s.%s failed to do post checkpoint commit", database, tableName),
+          //          t);
+          LOG.error("Iceberg committer {}.{} failed to do post checkpoint commit. Throwable = ",
+              namespace, tableName, t);
+        }
+      } else {
+        // TODO: it would be nice to fix this and allow concurrent checkpoint
+        LOG.info("Iceberg committer {}.{} skip committing transaction: " +

Review comment:
       Actually this is not only about concurrent checkpoint. Let's say if users set a short checkpoint interval, and `IcebergCommitter` may invoke `snapshotState` twice before receiving the notification of the first checkpoint's completion from `JobManager`. In this scenario, I guess the data is discarded because `pendingDataFiles` is cleared in `snapshotState` process if I understand correctly.




----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org



---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org

[GitHub] [incubator-iceberg] JingsongLi commented on a change in pull request #856: [WIP] Flink Iceberg sink

[GitBox <gi...@apache.org>]

JingsongLi commented on a change in pull request #856:
URL: https://github.com/apache/incubator-iceberg/pull/856#discussion_r415601644



##########
File path: flink/src/main/java/org/apache/iceberg/flink/connector/sink/IcebergSinkAppender.java
##########
@@ -0,0 +1,96 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iceberg.flink.connector.sink;
+
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.streaming.api.datastream.DataStream;
+import org.apache.flink.streaming.api.datastream.DataStreamSink;
+import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.data.Record;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Append Iceberg sink to DataStream
+ *
+ * @param <IN> input data type
+ */
+@SuppressWarnings({"checkstyle:ClassTypeParameterName", "checkstyle:HiddenField"})
+public class IcebergSinkAppender<IN> {
+  private static final Logger LOG = LoggerFactory.getLogger(IcebergSinkAppender.class);
+
+  private final Table table;
+  private final Configuration config;
+  private final String sinkName;
+  private RecordSerializer<IN> serializer;
+  private Integer writerParallelism;
+
+  public IcebergSinkAppender(Table table, Configuration config, String sinkName) {
+    this.table = table;
+    this.config = config;
+    this.sinkName = sinkName;
+  }
+
+  /**
+   * Required.
+   *
+   * @param serializer Serialize input data type to Iceberg {@link Record}
+   */
+  public IcebergSinkAppender<IN> withSerializer(RecordSerializer<IN> serializer) {
+    this.serializer = serializer;
+    return this;
+  }
+
+  /**
+   * Optional. Explicitly set the parallelism for Iceberg writer operator.
+   * Otherwise, default job parallelism is used.
+   */
+  public IcebergSinkAppender<IN> withWriterParallelism(Integer writerParallelism) {
+    this.writerParallelism = writerParallelism;
+    return this;
+  }
+
+  /**
+   * @param dataStream append sink to this DataStream
+   */
+  public DataStreamSink<FlinkDataFile> append(DataStream<IN> dataStream) {
+    // TODO: need to return?
+    IcebergWriter<IN> writer = new IcebergWriter<>(table, serializer, config);
+    IcebergCommitter committer = new IcebergCommitter(table, config);
+
+    final String writerId = sinkName + "-writer";
+    SingleOutputStreamOperator<FlinkDataFile> writerStream = dataStream
+        .transform(writerId, TypeInformation.of(FlinkDataFile.class), writer)  // IcebergWriter as stream operator
+        .uid(writerId);
+    if (null != writerParallelism && writerParallelism > 0) {
+      LOG.info("Set Iceberg writer parallelism to {}", writerParallelism);
+      writerStream.setParallelism(writerParallelism);

Review comment:
       Yes




----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org



---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org

[GitHub] [incubator-iceberg] waterlx commented on a change in pull request #856: [WIP] Flink Iceberg sink

[GitBox <gi...@apache.org>]

waterlx commented on a change in pull request #856:
URL: https://github.com/apache/incubator-iceberg/pull/856#discussion_r426275143



##########
File path: flink/src/main/java/org/apache/iceberg/flink/connector/sink/IcebergCommitter.java
##########
@@ -0,0 +1,633 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iceberg.flink.connector.sink;
+
+import com.google.common.base.Joiner;
+import com.google.common.base.Strings;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.stream.Collectors;
+import javax.annotation.Nullable;
+import org.apache.flink.annotation.VisibleForTesting;
+import org.apache.flink.api.common.JobExecutionResult;
+import org.apache.flink.api.common.JobID;
+import org.apache.flink.api.common.state.ListState;
+import org.apache.flink.api.common.state.ListStateDescriptor;
+import org.apache.flink.api.java.ExecutionEnvironment;
+import org.apache.flink.configuration.CheckpointingOptions;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.runtime.state.CheckpointListener;
+import org.apache.flink.runtime.state.FunctionInitializationContext;
+import org.apache.flink.runtime.state.FunctionSnapshotContext;
+import org.apache.flink.streaming.api.checkpoint.CheckpointedFunction;
+import org.apache.flink.streaming.api.environment.CheckpointConfig;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
+import org.apache.flink.util.Preconditions;
+import org.apache.hadoop.hive.conf.HiveConf.ConfVars;
+import org.apache.iceberg.AppendFiles;
+import org.apache.iceberg.DataFile;
+import org.apache.iceberg.ManifestFile;
+import org.apache.iceberg.ManifestFiles;
+import org.apache.iceberg.ManifestWriter;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.Snapshot;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.Transaction;
+import org.apache.iceberg.UpdateProperties;
+import org.apache.iceberg.catalog.Catalog;
+import org.apache.iceberg.catalog.TableIdentifier;
+import org.apache.iceberg.flink.connector.IcebergConnectorConstant;
+import org.apache.iceberg.flink.connector.model.CommitMetadata;
+import org.apache.iceberg.flink.connector.model.CommitMetadataUtil;
+import org.apache.iceberg.flink.connector.model.FlinkManifestFile;
+import org.apache.iceberg.flink.connector.model.FlinkManifestFileUtil;
+import org.apache.iceberg.flink.connector.model.GenericFlinkManifestFile;
+import org.apache.iceberg.flink.connector.model.ManifestFileState;
+import org.apache.iceberg.hadoop.HadoopCatalog;
+import org.apache.iceberg.hive.HiveCatalogs;
+import org.apache.iceberg.io.FileIO;
+import org.apache.iceberg.io.OutputFile;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * This operator commit data files to Iceberg table.
+ * <p>
+ * This operator should always run with parallelism of 1.
+ * Because Iceberg lib perform optimistic concurrency control,
+ * this can help reduce contention and retries
+ * when committing files to Iceberg table.
+ * <p>
+ */
+@SuppressWarnings("checkstyle:HiddenField")
+public class IcebergCommitter extends RichSinkFunction<FlinkDataFile>
+    implements CheckpointedFunction, CheckpointListener {
+  private static final Logger LOG = LoggerFactory.getLogger(IcebergCommitter.class);
+
+  private static final String COMMIT_MANIFEST_HASHES_KEY = "flink.commit.manifest.hashes";
+  private static final String WATERMARK_PROP_KEY_PREFIX = "flink.watermark";
+
+  private Configuration config;
+  private final String namespace;
+  private final String tableName;
+
+  private final boolean watermarkEnabled;
+  private final String watermarkPropKey;
+  private final long snapshotRetentionHours;
+  private final boolean commitRestoredManifestFiles;
+  private final String icebergManifestFileDir;
+  private final PartitionSpec spec;
+  private final FileIO io;
+  private final String flinkJobId;
+
+  private transient Table table;
+  private transient List<FlinkDataFile> pendingDataFiles;
+  private transient List<FlinkManifestFile> flinkManifestFiles;
+  private transient ListState<ManifestFileState> manifestFileState;
+  private transient CommitMetadata metadata;
+  private transient ListState<CommitMetadata> commitMetadataState;
+
+  public IcebergCommitter(Table table, Configuration config) {
+    this.config = config;
+
+    // current Iceberg sink implementation can't work with concurrent checkpoints.
+    // We disable concurrent checkpoints by default as min pause is set to 60s by default.
+    // Add an assertion to fail explicit in case job enables concurrent checkpoints.
+    CheckpointConfig checkpointConfig = StreamExecutionEnvironment.getExecutionEnvironment().getCheckpointConfig();
+    if (checkpointConfig.getMaxConcurrentCheckpoints() > 1) {
+      throw new IllegalArgumentException("Iceberg sink doesn't support concurrent checkpoints");
+    }
+
+    namespace = config.getString(IcebergConnectorConstant.NAMESPACE, "");
+    tableName = config.getString(IcebergConnectorConstant.TABLE, "");
+
+    watermarkEnabled = !Strings.isNullOrEmpty(
+        config.getString(IcebergConnectorConstant.WATERMARK_TIMESTAMP_FIELD, ""));
+    watermarkPropKey = WATERMARK_PROP_KEY_PREFIX;
+    snapshotRetentionHours = config.getLong(IcebergConnectorConstant.SNAPSHOT_RETENTION_HOURS,
+        IcebergConnectorConstant.DEFAULT_SNAPSHOT_RETENTION_HOURS);
+    commitRestoredManifestFiles = config.getBoolean(IcebergConnectorConstant.COMMIT_RESTORED_MANIFEST_FILES,
+        IcebergConnectorConstant.DEFAULT_COMMIT_RESTORED_MANIFEST_FILES);
+    icebergManifestFileDir = getIcebergManifestFileDir(config);
+
+    // The only final fields yielded by table inputted
+    spec = table.spec();
+    io = table.io();
+
+    final JobExecutionResult jobExecutionResult
+        = ExecutionEnvironment.getExecutionEnvironment().getLastJobExecutionResult();
+    if (jobExecutionResult != null) {
+      flinkJobId = jobExecutionResult.getJobID().toString();
+      LOG.info("Get Flink job ID from execution environment: {}", flinkJobId);
+    } else {
+      flinkJobId = new JobID().toString();
+      LOG.info("Execution environment doesn't have executed job. Generate a random job ID : {}", flinkJobId);
+    }
+    LOG.info("Iceberg committer {}.{} created with sink config", namespace, tableName);
+    LOG.info("Iceberg committer {}.{} loaded table partition spec: {}", namespace, tableName, spec);
+  }
+
+  @VisibleForTesting
+  List<FlinkDataFile> getPendingDataFiles() {
+    return pendingDataFiles;
+  }
+
+  @VisibleForTesting
+  List<FlinkManifestFile> getFlinkManifestFiles() {
+    return flinkManifestFiles;
+  }
+
+  @VisibleForTesting
+  CommitMetadata getMetadata() {
+    return metadata;
+  }
+
+  private String getIcebergManifestFileDir(Configuration config) {
+    final String checkpointDir = config.getString(
+        CheckpointingOptions.CHECKPOINTS_DIRECTORY, null);
+    if (null == checkpointDir) {
+      throw new IllegalArgumentException("checkpoint dir is null");
+    }
+
+    return String.format("%s/iceberg/manifest/", checkpointDir);
+  }
+
+  @Override
+  public void close() throws Exception {
+    super.close();
+  }
+
+  void init() {
+    // TODO: duplicate logic, to extract
+    org.apache.hadoop.conf.Configuration hadoopConf = new org.apache.hadoop.conf.Configuration();
+    String catalogType = config.getString(IcebergConnectorConstant.CATALOG_TYPE,
+                                          IcebergConnectorConstant.CATALOG_TYPE_DEFAULT);
+    Catalog catalog = null;
+    switch (catalogType.toUpperCase()) {
+      case IcebergConnectorConstant.HIVE_CATALOG:
+        hadoopConf.set(ConfVars.METASTOREURIS.varname, config.getString(ConfVars.METASTOREURIS.varname, ""));
+        catalog = HiveCatalogs.loadCatalog(hadoopConf);
+        break;
+
+      case IcebergConnectorConstant.HADOOP_CATALOG:
+        catalog = new HadoopCatalog(hadoopConf,
+                                    config.getString(IcebergConnectorConstant.HADOOP_CATALOG_WAREHOUSE_LOCATION, ""));
+        break;
+
+      default:
+        throw new UnsupportedOperationException("Unknown catalog type or not set: " + catalogType);
+    }
+
+    this.table = catalog.loadTable(TableIdentifier.parse(namespace + "." + tableName));
+
+    pendingDataFiles = new ArrayList<>();
+    flinkManifestFiles = new ArrayList<>();
+    metadata = CommitMetadata.newBuilder()
+        .setLastCheckpointId(0)
+        .setLastCheckpointTimestamp(0)
+        .setLastCommitTimestamp(System.currentTimeMillis())

Review comment:
       @Jiayi-Liao, thanks for your comment! I am working on addressing them.




----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org



---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org

[GitHub] [incubator-iceberg] Jiayi-Liao commented on a change in pull request #856: [WIP] Flink Iceberg sink

[GitBox <gi...@apache.org>]

Jiayi-Liao commented on a change in pull request #856:
URL: https://github.com/apache/incubator-iceberg/pull/856#discussion_r426222173



##########
File path: flink/src/main/java/org/apache/iceberg/flink/connector/sink/IcebergCommitter.java
##########
@@ -0,0 +1,633 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iceberg.flink.connector.sink;
+
+import com.google.common.base.Joiner;
+import com.google.common.base.Strings;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.stream.Collectors;
+import javax.annotation.Nullable;
+import org.apache.flink.annotation.VisibleForTesting;
+import org.apache.flink.api.common.JobExecutionResult;
+import org.apache.flink.api.common.JobID;
+import org.apache.flink.api.common.state.ListState;
+import org.apache.flink.api.common.state.ListStateDescriptor;
+import org.apache.flink.api.java.ExecutionEnvironment;
+import org.apache.flink.configuration.CheckpointingOptions;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.runtime.state.CheckpointListener;
+import org.apache.flink.runtime.state.FunctionInitializationContext;
+import org.apache.flink.runtime.state.FunctionSnapshotContext;
+import org.apache.flink.streaming.api.checkpoint.CheckpointedFunction;
+import org.apache.flink.streaming.api.environment.CheckpointConfig;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
+import org.apache.flink.util.Preconditions;
+import org.apache.hadoop.hive.conf.HiveConf.ConfVars;
+import org.apache.iceberg.AppendFiles;
+import org.apache.iceberg.DataFile;
+import org.apache.iceberg.ManifestFile;
+import org.apache.iceberg.ManifestFiles;
+import org.apache.iceberg.ManifestWriter;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.Snapshot;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.Transaction;
+import org.apache.iceberg.UpdateProperties;
+import org.apache.iceberg.catalog.Catalog;
+import org.apache.iceberg.catalog.TableIdentifier;
+import org.apache.iceberg.flink.connector.IcebergConnectorConstant;
+import org.apache.iceberg.flink.connector.model.CommitMetadata;
+import org.apache.iceberg.flink.connector.model.CommitMetadataUtil;
+import org.apache.iceberg.flink.connector.model.FlinkManifestFile;
+import org.apache.iceberg.flink.connector.model.FlinkManifestFileUtil;
+import org.apache.iceberg.flink.connector.model.GenericFlinkManifestFile;
+import org.apache.iceberg.flink.connector.model.ManifestFileState;
+import org.apache.iceberg.hadoop.HadoopCatalog;
+import org.apache.iceberg.hive.HiveCatalogs;
+import org.apache.iceberg.io.FileIO;
+import org.apache.iceberg.io.OutputFile;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * This operator commit data files to Iceberg table.
+ * <p>
+ * This operator should always run with parallelism of 1.
+ * Because Iceberg lib perform optimistic concurrency control,
+ * this can help reduce contention and retries
+ * when committing files to Iceberg table.
+ * <p>
+ */
+@SuppressWarnings("checkstyle:HiddenField")
+public class IcebergCommitter extends RichSinkFunction<FlinkDataFile>
+    implements CheckpointedFunction, CheckpointListener {
+  private static final Logger LOG = LoggerFactory.getLogger(IcebergCommitter.class);
+
+  private static final String COMMIT_MANIFEST_HASHES_KEY = "flink.commit.manifest.hashes";
+  private static final String WATERMARK_PROP_KEY_PREFIX = "flink.watermark";
+
+  private Configuration config;
+  private final String namespace;
+  private final String tableName;
+
+  private final boolean watermarkEnabled;
+  private final String watermarkPropKey;
+  private final long snapshotRetentionHours;
+  private final boolean commitRestoredManifestFiles;
+  private final String icebergManifestFileDir;
+  private final PartitionSpec spec;
+  private final FileIO io;
+  private final String flinkJobId;
+
+  private transient Table table;
+  private transient List<FlinkDataFile> pendingDataFiles;
+  private transient List<FlinkManifestFile> flinkManifestFiles;
+  private transient ListState<ManifestFileState> manifestFileState;
+  private transient CommitMetadata metadata;
+  private transient ListState<CommitMetadata> commitMetadataState;
+
+  public IcebergCommitter(Table table, Configuration config) {
+    this.config = config;
+
+    // current Iceberg sink implementation can't work with concurrent checkpoints.
+    // We disable concurrent checkpoints by default as min pause is set to 60s by default.
+    // Add an assertion to fail explicit in case job enables concurrent checkpoints.
+    CheckpointConfig checkpointConfig = StreamExecutionEnvironment.getExecutionEnvironment().getCheckpointConfig();
+    if (checkpointConfig.getMaxConcurrentCheckpoints() > 1) {
+      throw new IllegalArgumentException("Iceberg sink doesn't support concurrent checkpoints");
+    }
+
+    namespace = config.getString(IcebergConnectorConstant.NAMESPACE, "");
+    tableName = config.getString(IcebergConnectorConstant.TABLE, "");
+
+    watermarkEnabled = !Strings.isNullOrEmpty(
+        config.getString(IcebergConnectorConstant.WATERMARK_TIMESTAMP_FIELD, ""));
+    watermarkPropKey = WATERMARK_PROP_KEY_PREFIX;
+    snapshotRetentionHours = config.getLong(IcebergConnectorConstant.SNAPSHOT_RETENTION_HOURS,
+        IcebergConnectorConstant.DEFAULT_SNAPSHOT_RETENTION_HOURS);
+    commitRestoredManifestFiles = config.getBoolean(IcebergConnectorConstant.COMMIT_RESTORED_MANIFEST_FILES,
+        IcebergConnectorConstant.DEFAULT_COMMIT_RESTORED_MANIFEST_FILES);
+    icebergManifestFileDir = getIcebergManifestFileDir(config);
+
+    // The only final fields yielded by table inputted
+    spec = table.spec();
+    io = table.io();
+
+    final JobExecutionResult jobExecutionResult
+        = ExecutionEnvironment.getExecutionEnvironment().getLastJobExecutionResult();
+    if (jobExecutionResult != null) {
+      flinkJobId = jobExecutionResult.getJobID().toString();
+      LOG.info("Get Flink job ID from execution environment: {}", flinkJobId);
+    } else {
+      flinkJobId = new JobID().toString();
+      LOG.info("Execution environment doesn't have executed job. Generate a random job ID : {}", flinkJobId);
+    }
+    LOG.info("Iceberg committer {}.{} created with sink config", namespace, tableName);
+    LOG.info("Iceberg committer {}.{} loaded table partition spec: {}", namespace, tableName, spec);
+  }
+
+  @VisibleForTesting
+  List<FlinkDataFile> getPendingDataFiles() {
+    return pendingDataFiles;
+  }
+
+  @VisibleForTesting
+  List<FlinkManifestFile> getFlinkManifestFiles() {
+    return flinkManifestFiles;
+  }
+
+  @VisibleForTesting
+  CommitMetadata getMetadata() {
+    return metadata;
+  }
+
+  private String getIcebergManifestFileDir(Configuration config) {
+    final String checkpointDir = config.getString(
+        CheckpointingOptions.CHECKPOINTS_DIRECTORY, null);
+    if (null == checkpointDir) {
+      throw new IllegalArgumentException("checkpoint dir is null");
+    }
+
+    return String.format("%s/iceberg/manifest/", checkpointDir);
+  }
+
+  @Override
+  public void close() throws Exception {
+    super.close();
+  }
+
+  void init() {
+    // TODO: duplicate logic, to extract
+    org.apache.hadoop.conf.Configuration hadoopConf = new org.apache.hadoop.conf.Configuration();
+    String catalogType = config.getString(IcebergConnectorConstant.CATALOG_TYPE,
+                                          IcebergConnectorConstant.CATALOG_TYPE_DEFAULT);
+    Catalog catalog = null;
+    switch (catalogType.toUpperCase()) {
+      case IcebergConnectorConstant.HIVE_CATALOG:
+        hadoopConf.set(ConfVars.METASTOREURIS.varname, config.getString(ConfVars.METASTOREURIS.varname, ""));
+        catalog = HiveCatalogs.loadCatalog(hadoopConf);
+        break;
+
+      case IcebergConnectorConstant.HADOOP_CATALOG:
+        catalog = new HadoopCatalog(hadoopConf,
+                                    config.getString(IcebergConnectorConstant.HADOOP_CATALOG_WAREHOUSE_LOCATION, ""));
+        break;
+
+      default:
+        throw new UnsupportedOperationException("Unknown catalog type or not set: " + catalogType);
+    }
+
+    this.table = catalog.loadTable(TableIdentifier.parse(namespace + "." + tableName));
+
+    pendingDataFiles = new ArrayList<>();
+    flinkManifestFiles = new ArrayList<>();
+    metadata = CommitMetadata.newBuilder()
+        .setLastCheckpointId(0)
+        .setLastCheckpointTimestamp(0)
+        .setLastCommitTimestamp(System.currentTimeMillis())

Review comment:
       How about making these attributes as the default value in CommitMetadata? Otherwise it looks a bit weird because there is no clues that why you set these values manually.

##########
File path: flink/src/main/java/org/apache/iceberg/flink/connector/sink/IcebergCommitter.java
##########
@@ -0,0 +1,642 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iceberg.flink.connector.sink;
+
+import com.google.common.base.Joiner;
+import com.google.common.base.Strings;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.stream.Collectors;
+import javax.annotation.Nullable;
+import org.apache.flink.annotation.VisibleForTesting;
+import org.apache.flink.api.common.JobExecutionResult;
+import org.apache.flink.api.common.JobID;
+import org.apache.flink.api.common.state.ListState;
+import org.apache.flink.api.common.state.ListStateDescriptor;
+import org.apache.flink.api.java.ExecutionEnvironment;
+import org.apache.flink.configuration.CheckpointingOptions;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.runtime.state.CheckpointListener;
+import org.apache.flink.runtime.state.FunctionInitializationContext;
+import org.apache.flink.runtime.state.FunctionSnapshotContext;
+import org.apache.flink.streaming.api.checkpoint.CheckpointedFunction;
+import org.apache.flink.streaming.api.environment.CheckpointConfig;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
+import org.apache.flink.util.Preconditions;
+import org.apache.hadoop.hive.conf.HiveConf.ConfVars;
+import org.apache.iceberg.AppendFiles;
+import org.apache.iceberg.DataFile;
+import org.apache.iceberg.ManifestFile;
+import org.apache.iceberg.ManifestWriter;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.Snapshot;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.Transaction;
+import org.apache.iceberg.UpdateProperties;
+import org.apache.iceberg.catalog.Catalog;
+import org.apache.iceberg.catalog.TableIdentifier;
+import org.apache.iceberg.flink.connector.IcebergConnectorConstant;
+import org.apache.iceberg.flink.connector.model.CommitMetadata;
+import org.apache.iceberg.flink.connector.model.CommitMetadataUtil;
+import org.apache.iceberg.flink.connector.model.FlinkManifestFile;
+import org.apache.iceberg.flink.connector.model.FlinkManifestFileUtil;
+import org.apache.iceberg.flink.connector.model.GenericFlinkManifestFile;
+import org.apache.iceberg.flink.connector.model.ManifestFileState;
+import org.apache.iceberg.hadoop.HadoopCatalog;
+import org.apache.iceberg.hive.HiveCatalogs;
+import org.apache.iceberg.io.FileIO;
+import org.apache.iceberg.io.OutputFile;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * This operator commit data files to Iceberg table.
+ * <p>
+ * This operator should always run with parallelism of 1.
+ * Because Iceberg lib perform optimistic concurrency control,
+ * this can help reduce contention and retries
+ * when committing files to Iceberg table.
+ * <p>
+ * Here are some known contentions
+ * 1) Flink jobs running in multiple regions,
+ * since Iceberg metadata service and commit only happens in us-east-1.
+ * 2) auto tuning and auto lift services may update Iceberg table infrequently.
+ */
+@SuppressWarnings("checkstyle:HiddenField")
+public class IcebergCommitter extends RichSinkFunction<FlinkDataFile>
+    implements CheckpointedFunction, CheckpointListener {
+  private static final Logger LOG = LoggerFactory.getLogger(IcebergCommitter.class);
+
+  private static final String COMMIT_REGION_KEY = "flink.commit.region";
+  private static final String COMMIT_MANIFEST_HASHES_KEY = "flink.commit.manifest.hashes";
+  private static final String VTTS_WATERMARK_PROP_KEY_PREFIX = "flink.watermark.";
+
+  private Configuration config;
+  private final String namespace;
+  private final String tableName;
+
+  private final String region;
+  private final boolean vttsWatermarkEnabled;
+  private final String vttsWatermarkPropKey;
+  private final long snapshotRetentionHours;
+  private final boolean commitRestoredManifestFiles;
+  private final String icebergManifestFileDir;
+  private final PartitionSpec spec;
+  private final FileIO io;
+  private final String flinkJobId;
+
+  private transient Table table;
+  private transient List<FlinkDataFile> pendingDataFiles;
+  private transient List<FlinkManifestFile> flinkManifestFiles;
+  private transient ListState<ManifestFileState> manifestFilesState;
+  private transient CommitMetadata metadata;
+  private transient ListState<CommitMetadata> metadataState;
+
+  public IcebergCommitter(Table table, Configuration config) {
+    this.config = config;
+
+    // current Iceberg sink implementation can't work with concurrent checkpoints.
+    // We disable concurrent checkpoints by default as min pause is set to 60s by default.
+    // Add an assertion to fail explicit in case job enables concurrent checkpoints.
+    CheckpointConfig checkpointConfig = StreamExecutionEnvironment.getExecutionEnvironment().getCheckpointConfig();
+    if (checkpointConfig.getMaxConcurrentCheckpoints() > 1) {
+      throw new IllegalArgumentException("Iceberg sink doesn't support concurrent checkpoints");
+    }
+
+    namespace = config.getString(IcebergConnectorConstant.NAMESPACE, "");
+    tableName = config.getString(IcebergConnectorConstant.TABLE, "");
+
+    String region = System.getenv("EC2_REGION");
+    if (Strings.isNullOrEmpty(region)) {
+      region = "us-east-1";
+      LOG.info("Iceberg committer {}.{} default region to us-east-1", namespace, tableName);
+    }
+    this.region = region;
+
+    vttsWatermarkEnabled = !Strings.isNullOrEmpty(
+        config.getString(IcebergConnectorConstant.VTTS_WATERMARK_TIMESTAMP_FIELD, ""));
+    vttsWatermarkPropKey = VTTS_WATERMARK_PROP_KEY_PREFIX + region;
+    snapshotRetentionHours = config.getLong(IcebergConnectorConstant.SNAPSHOT_RETENTION_HOURS,
+        IcebergConnectorConstant.DEFAULT_SNAPSHOT_RETENTION_HOURS);
+    commitRestoredManifestFiles = config.getBoolean(IcebergConnectorConstant.COMMIT_RESTORED_MANIFEST_FILES,
+        IcebergConnectorConstant.DEFAULT_COMMIT_RESTORED_MANIFEST_FILES);
+    icebergManifestFileDir = getIcebergManifestFileDir(config);
+
+    // The only final fields yielded by table inputted
+    spec = table.spec();
+    io = table.io();
+
+    final JobExecutionResult jobExecutionResult
+        = ExecutionEnvironment.getExecutionEnvironment().getLastJobExecutionResult();
+    if (jobExecutionResult != null) {
+      flinkJobId = jobExecutionResult.getJobID().toString();
+      LOG.info("Get Flink job ID from execution environment: {}", flinkJobId);
+    } else {
+      flinkJobId = new JobID().toString();
+      LOG.info("Execution environment doesn't have executed job. Generate a random job ID : {}", flinkJobId);
+    }
+    LOG.info("Iceberg committer {}.{} created with sink config", namespace, tableName);
+    LOG.info("Iceberg committer {}.{} loaded table partition spec: {}", namespace, tableName, spec);
+  }
+
+  @VisibleForTesting
+  List<FlinkDataFile> getPendingDataFiles() {
+    return pendingDataFiles;
+  }
+
+  @VisibleForTesting
+  List<FlinkManifestFile> getFlinkManifestFiles() {
+    return flinkManifestFiles;
+  }
+
+  @VisibleForTesting
+  CommitMetadata getMetadata() {
+    return metadata;
+  }
+
+  private String getIcebergManifestFileDir(Configuration config) {
+    final String checkpointDir = config.getString(
+        CheckpointingOptions.CHECKPOINTS_DIRECTORY, null);
+    if (null == checkpointDir) {
+      throw new IllegalArgumentException("checkpoint dir is null");
+    }
+
+    return String.format("%s/iceberg/manifest/", checkpointDir);
+  }
+
+  @Override
+  public void close() throws Exception {
+    super.close();
+  }
+
+  void init() {
+    // TODO: duplicate logic, to extract
+    org.apache.hadoop.conf.Configuration hadoopConf = new org.apache.hadoop.conf.Configuration();
+    String catalogType = config.getString(IcebergConnectorConstant.CATALOG_TYPE,
+                                          IcebergConnectorConstant.CATALOG_TYPE_DEFAULT);
+    Catalog catalog = null;
+    switch (catalogType.toUpperCase()) {
+      case IcebergConnectorConstant.HIVE_CATALOG:
+        hadoopConf.set(ConfVars.METASTOREURIS.varname, config.getString(ConfVars.METASTOREURIS.varname, ""));
+        catalog = HiveCatalogs.loadCatalog(hadoopConf);
+        break;
+
+      case IcebergConnectorConstant.HADOOP_CATALOG:
+        catalog = new HadoopCatalog(hadoopConf,
+                                    config.getString(IcebergConnectorConstant.HADOOP_CATALOG_WAREHOUSE_LOCATION, ""));
+        break;
+
+      default:
+        throw new UnsupportedOperationException("Unknown catalog type or not set: " + catalogType);
+    }
+
+    this.table = catalog.loadTable(TableIdentifier.parse(namespace + "." + tableName));
+
+    pendingDataFiles = new ArrayList<>();
+    flinkManifestFiles = new ArrayList<>();
+    metadata = CommitMetadata.newBuilder()
+        .setLastCheckpointId(0)
+        .setLastCheckpointTimestamp(0)
+        .setLastCommitTimestamp(System.currentTimeMillis())
+        .build();
+  }
+
+  @Override
+  public void initializeState(FunctionInitializationContext context) throws Exception {
+    init();
+
+    Preconditions.checkState(manifestFilesState == null,
+        "checkpointedFilesState has already been initialized.");
+    Preconditions.checkState(metadataState == null,
+        "metadataState has already been initialized.");
+    manifestFilesState = context.getOperatorStateStore().getListState(new ListStateDescriptor<>(
+        "iceberg-committer-manifest-files-state", ManifestFileState.class));
+    metadataState = context.getOperatorStateStore().getListState(new ListStateDescriptor<>(
+        "iceberg-committer-metadata-state", CommitMetadata.class));
+
+    if (context.isRestored()) {
+      final Iterable<CommitMetadata> restoredMetadata = metadataState.get();
+      if (null != restoredMetadata) {
+        LOG.info("Iceberg committer {}.{} restoring metadata", namespace, tableName);
+        List<CommitMetadata> metadataList = new ArrayList<>();
+        for (CommitMetadata entry : restoredMetadata) {
+          metadataList.add(entry);
+        }
+        Preconditions.checkState(1 == metadataList.size(),
+            "metadata list size should be 1. got " + metadataList.size());
+        metadata = metadataList.get(0);
+        LOG.info("Iceberg committer {}.{} restored metadata: {}",
+            namespace, tableName, CommitMetadataUtil.getInstance().encodeAsJson(metadata));
+      } else {
+        LOG.info("Iceberg committer {}.{} has nothing to restore for metadata", namespace, tableName);
+      }
+
+      Iterable<ManifestFileState> restoredManifestFileStates = manifestFilesState.get();
+      if (null != restoredManifestFileStates) {
+        LOG.info("Iceberg committer {}.{} restoring manifest files",
+            namespace, tableName);
+        for (ManifestFileState manifestFileState : restoredManifestFileStates) {
+          flinkManifestFiles.add(GenericFlinkManifestFile.fromState(manifestFileState));
+        }
+        LOG.info("Iceberg committer {}.{} restored {} manifest files: {}",
+            namespace, tableName, flinkManifestFiles.size(), flinkManifestFiles);
+        final long now = System.currentTimeMillis();
+        if (now - metadata.getLastCheckpointTimestamp() > TimeUnit.HOURS.toMillis(snapshotRetentionHours)) {
+          flinkManifestFiles.clear();
+          LOG.info("Iceberg committer {}.{} cleared restored manifest files as checkpoint timestamp is too old: " +
+                  "checkpointTimestamp = {}, now = {}, snapshotRetentionHours = {}",
+              namespace, tableName, metadata.getLastCheckpointTimestamp(), now, snapshotRetentionHours);
+        } else {
+          flinkManifestFiles = removeCommittedManifests(flinkManifestFiles);
+          if (flinkManifestFiles.isEmpty()) {
+            LOG.info("Iceberg committer {}.{} has zero uncommitted manifest files from restored state",
+                namespace, tableName);
+          } else {
+            if (commitRestoredManifestFiles) {
+              commitRestoredManifestFiles();
+            } else {
+              LOG.info("skip commit of restored manifest files");
+            }
+          }
+        }
+      } else {
+        LOG.info("Iceberg committer {}.{} has nothing to restore for manifest files", namespace, tableName);
+      }
+    }
+  }
+
+  @VisibleForTesting
+  void commitRestoredManifestFiles() throws Exception {
+    LOG.info("Iceberg committer {}.{} committing last uncompleted transaction upon recovery: " +
+            "metadata = {}, flink manifest files ({}) = {}", namespace, tableName,
+        CommitMetadataUtil.getInstance().encodeAsJson(metadata),
+        flinkManifestFiles.size(), flinkManifestFiles);
+    commit();
+    LOG.info("Iceberg committer {}.{} committed last uncompleted transaction upon recovery: " +
+            "metadata = {}, flink manifest files ({}) = {}", namespace, tableName,
+        CommitMetadataUtil.getInstance().encodeAsJson(metadata),
+        flinkManifestFiles.size(), flinkManifestFiles);
+    postCommitSuccess();
+  }
+
+  private List<FlinkManifestFile> removeCommittedManifests(List<FlinkManifestFile> flinkManifestFiles) {
+    int snapshotCount = 0;
+    String result = "succeeded";
+    final long start = System.currentTimeMillis();
+    try {
+      final Set<String> manifestHashes = flinkManifestFiles.stream()
+          .map(f -> f.hash())
+          .collect(Collectors.toSet());
+      final Set<String> committedHashes = new HashSet<>(flinkManifestFiles.size());
+      final Iterable<Snapshot> snapshots = table.snapshots();
+      for (Snapshot snapshot : snapshots) {
+        ++snapshotCount;
+        final Map<String, String> summary = snapshot.summary();
+        final List<String> hashes = FlinkManifestFileUtil.hashesStringToList(summary.get(COMMIT_MANIFEST_HASHES_KEY));
+        for (String hash : hashes) {
+          if (manifestHashes.contains(hash)) {
+            committedHashes.add(hash);
+          }
+        }
+      }
+      final List<FlinkManifestFile> uncommittedManifestFiles = flinkManifestFiles.stream()
+          .filter(f -> !committedHashes.contains(f.hash()))
+          .collect(Collectors.toList());
+      return uncommittedManifestFiles;
+    } catch (Throwable t) {
+      result = "failed";
+      //LOG.error(String.format("Iceberg committer %s.%s failed to check transaction completed", database, tableName),
+      //          t);
+      LOG.error("Iceberg committer {}.{} failed to check transaction completed. Throwable = {}",
+                namespace, tableName, t);
+      throw t;
+    } finally {
+      final long duration = System.currentTimeMillis() - start;
+      LOG.info("Iceberg committer {}.{} {} to check transaction completed" +
+               " after iterating {} snapshots and {} milli-seconds",
+               namespace, tableName, result, snapshotCount, duration);
+    }
+  }
+
+  @Override
+  public void snapshotState(FunctionSnapshotContext context) throws Exception {
+    LOG.info("Iceberg committer {}.{} snapshot state: checkpointId = {}, triggerTime = {}",
+        namespace, tableName, context.getCheckpointId(), context.getCheckpointTimestamp());
+    Preconditions.checkState(null != manifestFilesState,
+        "manifest files state has not been properly initialized.");
+    Preconditions.checkState(null != metadataState,
+        "metadata state has not been properly initialized.");
+
+    // set transaction to null to indicate a start of a new checkpoint/commit/transaction
+    synchronized (this) {
+      snapshot(context, pendingDataFiles);
+      checkpointState(flinkManifestFiles, metadata);
+      postSnapshotSuccess();
+    }
+  }
+
+  @VisibleForTesting
+  void snapshot(FunctionSnapshotContext context, List<FlinkDataFile> pendingDataFiles) throws Exception {
+    FlinkManifestFile flinkManifestFile = null;
+    if (!pendingDataFiles.isEmpty()) {
+      flinkManifestFile = createManifestFile(context, pendingDataFiles);
+      flinkManifestFiles.add(flinkManifestFile);
+    }
+    metadata = updateMetadata(metadata, context, flinkManifestFile);
+  }
+
+  private FlinkManifestFile createManifestFile(
+      FunctionSnapshotContext context, List<FlinkDataFile> pendingDataFiles) throws Exception {
+    LOG.info("Iceberg committer {}.{} checkpointing {} pending data files}",
+        namespace, tableName, pendingDataFiles.size());
+    String result = "succeeded";
+    final long start = System.currentTimeMillis();
+    try {
+      final String manifestFileName = Joiner.on("_")
+          .join(flinkJobId, context.getCheckpointId(), context.getCheckpointTimestamp());
+      // Iceberg requires file format suffix right now
+      final String manifestFileNameWithSuffix = manifestFileName + ".avro";
+      OutputFile outputFile = io.newOutputFile(icebergManifestFileDir + manifestFileNameWithSuffix);
+      ManifestWriter manifestWriter = ManifestWriter.write(spec, outputFile);
+
+      // stats
+      long recordCount = 0;
+      long byteCount = 0;
+      long lowWatermark = Long.MAX_VALUE;
+      long highWatermark = Long.MIN_VALUE;
+      for (FlinkDataFile flinkDataFile : pendingDataFiles) {
+        DataFile dataFile = flinkDataFile.getIcebergDataFile();
+        manifestWriter.add(dataFile);
+        // update stas
+        recordCount += dataFile.recordCount();
+        byteCount += dataFile.fileSizeInBytes();
+        if (flinkDataFile.getLowWatermark() < lowWatermark) {
+          lowWatermark = flinkDataFile.getLowWatermark();
+        }
+        if (flinkDataFile.getHighWatermark() > highWatermark) {
+          highWatermark = flinkDataFile.getHighWatermark();
+        }
+        LOG.debug("Data file with size of {} bytes added to manifest", dataFile.fileSizeInBytes());
+      }
+      manifestWriter.close();
+      ManifestFile manifestFile = manifestWriter.toManifestFile();
+
+      FlinkManifestFile flinkManifestFile = GenericFlinkManifestFile.builder()
+          .setPath(manifestFile.path())
+          .setLength(manifestFile.length())
+          .setSpecId(manifestFile.partitionSpecId())
+          .setCheckpointId(context.getCheckpointId())
+          .setCheckpointTimestamp(context.getCheckpointTimestamp())
+          .setDataFileCount(pendingDataFiles.size())
+          .setRecordCount(recordCount)
+          .setByteCount(byteCount)
+          .setLowWatermark(lowWatermark)
+          .setHighWatermark(highWatermark)
+          .build();
+
+      // don't want to log a giant list at one line.
+      // split the complete list into smaller chunks with 50 files.
+      final AtomicInteger counter = new AtomicInteger(0);
+      Collection<List<String>> listOfFileList = pendingDataFiles.stream()
+          .map(flinkDataFile -> flinkDataFile.toCompactDump())
+          .collect(Collectors.groupingBy(it -> counter.getAndIncrement() / 50))
+          .values();
+      for (List<String> fileList : listOfFileList) {
+        LOG.info("Iceberg committer {}.{} created manifest file {} for {}/{} data files: {}",
+            namespace, tableName, manifestFile.path(), fileList.size(), pendingDataFiles.size(), fileList);
+      }
+      return flinkManifestFile;
+    } catch (Throwable t) {
+      result = "failed";
+      //LOG.error(String.format("Iceberg committer %s.%s failed to create manifest file for %d pending data files",
+      //          database, tableName, pendingDataFiles.size()), t);
+      LOG.error("Iceberg committer {}.{} failed to create manifest file for {} pending data files. Throwable={}",
+          namespace, tableName, pendingDataFiles.size(), t);
+      throw t;
+    } finally {
+      final long duration = System.currentTimeMillis() - start;
+      LOG.info("Iceberg committer {}.{} {} to create manifest file with {} data files after {} milli-seconds",
+          namespace, tableName, result, pendingDataFiles.size(), duration);
+    }
+  }
+
+  private CommitMetadata updateMetadata(
+      CommitMetadata oldMetadata,
+      FunctionSnapshotContext context,
+      @Nullable FlinkManifestFile flinkManifestFile) {
+    LOG.info("Iceberg committer {}.{} updating metadata {} with manifest file {}",
+        namespace, tableName, CommitMetadataUtil.getInstance().encodeAsJson(oldMetadata), flinkManifestFile);
+    CommitMetadata.Builder metadataBuilder = CommitMetadata.newBuilder(oldMetadata)
+        .setLastCheckpointId(context.getCheckpointId())
+        .setLastCheckpointTimestamp(context.getCheckpointTimestamp());
+    if (vttsWatermarkEnabled) {
+      Long vttsWatermark = oldMetadata.getVttsWatermark();
+      if (null == flinkManifestFile) {
+        // DPS-412: when there is no data to be committed
+        // use elapsed wall clock time to move the VTTS watermark forward.
+        if (null != vttsWatermark) {
+          final long elapsedTimeMs = System.currentTimeMillis() - oldMetadata.getLastCommitTimestamp();
+          vttsWatermark += elapsedTimeMs;
+        } else {
+          vttsWatermark = System.currentTimeMillis();
+        }
+      } else {
+        // use lowWatermark.
+        if (null == flinkManifestFile.lowWatermark()) {
+          throw new IllegalArgumentException("VTTS is enabled but lowWatermark is null");
+        }
+        // in case one container/slot is lagging behind,
+        // we want to move watermark forward based on the slowest.
+        final long newWatermark = flinkManifestFile.lowWatermark();
+        // make sure VTTS watermark doesn't go back in time
+        if (null == vttsWatermark || newWatermark > vttsWatermark) {
+          vttsWatermark = newWatermark;
+        }
+      }
+      metadataBuilder.setVttsWatermark(vttsWatermark);
+    }
+    CommitMetadata metadata = metadataBuilder.build();
+    LOG.info("Iceberg committer {}.{} updated metadata {} with manifest file {}",
+        namespace, tableName, CommitMetadataUtil.getInstance().encodeAsJson(metadata), flinkManifestFile);
+    return metadata;
+  }
+
+  private void checkpointState(List<FlinkManifestFile> flinkManifestFiles, CommitMetadata metadata) throws Exception {
+    LOG.info("Iceberg committer {}.{} checkpointing state", namespace, tableName);
+    List<ManifestFileState> manifestFileStates = flinkManifestFiles.stream()
+        .map(f -> f.toState())
+        .collect(Collectors.toList());
+    manifestFilesState.clear();
+    manifestFilesState.addAll(manifestFileStates);
+    metadataState.clear();
+    metadataState.add(metadata);
+    LOG.info("Iceberg committer {}.{} checkpointed state: metadata = {}, flinkManifestFiles({}) = {}",
+        namespace, tableName, CommitMetadataUtil.getInstance().encodeAsJson(metadata),
+        flinkManifestFiles.size(), flinkManifestFiles);
+  }
+
+  private void postSnapshotSuccess() {
+    pendingDataFiles.clear();
+    LOG.debug("Un-committed manifest file count {}, containing data file count {}, record count {} and byte count {}",
+        flinkManifestFiles.size(),
+        FlinkManifestFileUtil.getDataFileCount(flinkManifestFiles),
+        FlinkManifestFileUtil.getRecordCount(flinkManifestFiles),
+        FlinkManifestFileUtil.getByteCount(flinkManifestFiles)
+    );
+  }
+
+  @Override
+  public void notifyCheckpointComplete(long checkpointId) {
+    LOG.info("Iceberg committer {}.{} checkpoint {} completed", namespace, tableName, checkpointId);
+    synchronized (this) {
+      if (checkpointId == metadata.getLastCheckpointId()) {
+        try {
+          commit();
+          postCommitSuccess();
+        } catch (Exception t) {
+          // swallow the exception to avoid job restart in case of commit failure
+          //LOG.error(String.format("Iceberg committer %s.%s failed to do post checkpoint commit", database, tableName),
+          //          t);
+          LOG.error("Iceberg committer {}.{} failed to do post checkpoint commit. Throwable = ",
+              namespace, tableName, t);
+        }
+      } else {
+        // TODO: it would be nice to fix this and allow concurrent checkpoint
+        LOG.info("Iceberg committer {}.{} skip committing transaction: " +

Review comment:
       Actually this is not only about concurrent checkpoint. Let's say if users set a short checkpoint interval, and `IcebergCommitter` may invoke `snapshotState` twice before receiving the notification of the first checkpoint's completion from `JobManager`. In this scenario, I guess the data is discarded because `pendingDataFiles` is cleared in `snapshotState` process if I understand correctly.

##########
File path: flink/src/main/java/org/apache/iceberg/flink/connector/sink/IcebergWriter.java
##########
@@ -0,0 +1,330 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iceberg.flink.connector.sink;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Locale;
+import java.util.Map;
+import java.util.UUID;
+import java.util.concurrent.TimeUnit;
+import javax.annotation.Nullable;
+import org.apache.flink.annotation.VisibleForTesting;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.runtime.state.StateInitializationContext;
+import org.apache.flink.streaming.api.operators.AbstractStreamOperator;
+import org.apache.flink.streaming.api.operators.ChainingStrategy;
+import org.apache.flink.streaming.api.operators.OneInputStreamOperator;
+import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
+import org.apache.flink.streaming.runtime.tasks.ProcessingTimeService;
+import org.apache.hadoop.fs.Path;
+import org.apache.iceberg.FileFormat;
+import org.apache.iceberg.MetricsConfig;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.data.Record;
+import org.apache.iceberg.data.parquet.GenericParquetWriter;
+import org.apache.iceberg.exceptions.RuntimeIOException;
+import org.apache.iceberg.flink.connector.IcebergConnectorConstant;
+import org.apache.iceberg.io.FileAppender;
+import org.apache.iceberg.io.FileIO;
+import org.apache.iceberg.io.LocationProvider;
+import org.apache.iceberg.io.OutputFile;
+import org.apache.iceberg.parquet.Parquet;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import static org.apache.iceberg.TableProperties.DEFAULT_FILE_FORMAT;
+import static org.apache.iceberg.TableProperties.DEFAULT_FILE_FORMAT_DEFAULT;
+
+@SuppressWarnings("checkstyle:ClassTypeParameterName")
+public class IcebergWriter<IN> extends AbstractStreamOperator<FlinkDataFile>
+    implements OneInputStreamOperator<IN, FlinkDataFile> {
+
+  private static final Logger LOG = LoggerFactory.getLogger(IcebergWriter.class);
+
+  private final RecordSerializer<IN> serializer;
+  private final String namespace;
+  private final String tableName;
+  private final FileFormat format;
+  private final boolean skipIncompatibleRecord;
+  private final Schema schema;
+  private final PartitionSpec spec;
+  private final LocationProvider locations;
+  private final FileIO io;
+  private final Map<String, String> properties;
+  private final String timestampFeild;
+  private final TimeUnit timestampUnit;
+  private final long maxFileSize;
+
+  private transient org.apache.hadoop.conf.Configuration hadoopConf;
+  private transient Map<String, FileWriter> openPartitionFiles;
+  private transient int subtaskId;
+  private transient ProcessingTimeService timerService;
+  private transient Partitioner<Record> partitioner;
+
+  public IcebergWriter(Table table, @Nullable RecordSerializer<IN> serializer, Configuration config) {
+    this.serializer = serializer;
+    skipIncompatibleRecord = config.getBoolean(IcebergConnectorConstant.SKIP_INCOMPATIBLE_RECORD,
+        IcebergConnectorConstant.DEFAULT_SKIP_INCOMPATIBLE_RECORD);
+    timestampFeild = config.getString(IcebergConnectorConstant.WATERMARK_TIMESTAMP_FIELD, "");
+    timestampUnit = TimeUnit.valueOf(config.getString(IcebergConnectorConstant.WATERMARK_TIMESTAMP_UNIT,
+        IcebergConnectorConstant.DEFAULT_WATERMARK_TIMESTAMP_UNIT));
+    maxFileSize = config.getLong(IcebergConnectorConstant.MAX_FILE_SIZE,
+        IcebergConnectorConstant.DEFAULT_MAX_FILE_SIZE);
+
+    namespace = config.getString(IcebergConnectorConstant.NAMESPACE, "");
+    tableName = config.getString(IcebergConnectorConstant.TABLE, "");
+
+    schema = table.schema();
+    spec = table.spec();
+    locations = table.locationProvider();
+    io = table.io();
+    properties = table.properties();
+    format = FileFormat.valueOf(
+        properties.getOrDefault(DEFAULT_FILE_FORMAT, DEFAULT_FILE_FORMAT_DEFAULT).toUpperCase(Locale.ENGLISH));
+
+    LOG.info("Iceberg writer {}.{} data file location: {}",
+        namespace, tableName, locations.newDataLocation(""));
+    LOG.info("Iceberg writer {}.{} created with sink config", namespace, tableName);
+    LOG.info("Iceberg writer {}.{} loaded table: schema = {}\npartition spec = {}",
+        namespace, tableName, schema, spec);
+
+    // default ChainingStrategy is set to HEAD
+    // we prefer chaining to avoid the huge serialization and deserializatoin overhead.
+    super.setChainingStrategy(ChainingStrategy.ALWAYS);
+  }
+
+  @Override
+  public void open() throws Exception {
+    super.open();
+
+    hadoopConf = new org.apache.hadoop.conf.Configuration();
+    subtaskId = getRuntimeContext().getIndexOfThisSubtask();
+    timerService = getProcessingTimeService();
+    openPartitionFiles = new HashMap<>();
+  }
+
+  @Override
+  public void prepareSnapshotPreBarrier(long checkpointId) throws Exception {
+    LOG.info("Iceberg writer {}.{} subtask {} begin preparing for checkpoint {}",
+        namespace, tableName, subtaskId, checkpointId);
+    // close all open files and emit files to downstream committer operator
+    flush(true);
+    LOG.info("Iceberg writer {}.{} subtask {} completed preparing for checkpoint {}",
+        namespace, tableName, subtaskId, checkpointId);
+  }
+
+  @Override
+  public void initializeState(StateInitializationContext context) throws Exception {
+    super.initializeState(context);
+  }
+
+  @VisibleForTesting
+  List<FlinkDataFile> flush(boolean emit) throws IOException {
+    List<FlinkDataFile> dataFiles = new ArrayList<>(openPartitionFiles.size());
+    for (Map.Entry<String, FileWriter> entry : openPartitionFiles.entrySet()) {
+      FileWriter writer = entry.getValue();
+      FlinkDataFile flinkDataFile = closeWriter(writer);
+      dataFiles.add(flinkDataFile);
+      if (emit) {
+        emit(flinkDataFile);
+      }
+    }
+    LOG.info("Iceberg writer {}.{} subtask {} flushed {} open files",
+        namespace, tableName, subtaskId, openPartitionFiles.size());
+    openPartitionFiles.clear();
+    return dataFiles;
+  }
+
+  FlinkDataFile closeWriter(FileWriter writer) throws IOException {
+    FlinkDataFile flinkDataFile = writer.close();
+    LOG.info(
+        "Iceberg writer {}.{} subtask {} uploaded to Iceberg table {}.{} with {} records and {} bytes on this path: {}",
+        namespace, tableName, subtaskId, namespace, tableName, flinkDataFile.getIcebergDataFile().recordCount(),
+        flinkDataFile.getIcebergDataFile().fileSizeInBytes(), flinkDataFile.getIcebergDataFile().path());
+    return flinkDataFile;
+  }
+
+  void emit(FlinkDataFile flinkDataFile) {
+    output.collect(new StreamRecord<>(flinkDataFile));
+    LOG.debug("Iceberg writer {}.{} subtask {} emitted uploaded file to committer for Iceberg table {}.{}" +
+        " with {} records and {} bytes on this path: {}",
+        namespace, tableName, subtaskId, namespace, tableName, flinkDataFile.getIcebergDataFile().recordCount(),
+        flinkDataFile.getIcebergDataFile().fileSizeInBytes(), flinkDataFile.getIcebergDataFile().path());
+  }
+
+  @Override
+  public void close() throws Exception {
+    super.close();
+
+    LOG.info("Iceberg writer {}.{} subtask {} begin close", namespace, tableName, subtaskId);
+    // close all open files without emitting to downstream committer
+    flush(false);
+    LOG.info("Iceberg writer {}.{} subtask {} completed close", namespace, tableName, subtaskId);
+  }
+
+  @Override
+  public void dispose() throws Exception {
+    super.dispose();
+
+    LOG.info("Iceberg writer {}.{} subtask {} begin dispose", namespace, tableName, subtaskId);
+    abort();
+    LOG.info("Iceberg writer {}.{} subtask {} completed dispose", namespace, tableName, subtaskId);
+  }
+
+  private void abort() {
+    LOG.info("Iceberg writer {}.{} subtask {} has {} open files to abort",
+        namespace, tableName, subtaskId, openPartitionFiles.size());
+    // close all open files without sending DataFile list to downstream committer operator.
+    // because there are not checkpointed,
+    // we don't want to commit these files.
+    for (Map.Entry<String, FileWriter> entry : openPartitionFiles.entrySet()) {
+      final FileWriter writer = entry.getValue();
+      final Path path = writer.getPath();
+      try {
+        LOG.debug("Iceberg writer {}.{} subtask {} start to abort file: {}",
+            namespace, tableName, subtaskId, path);
+        writer.abort();
+        LOG.info("Iceberg writer {}.{} subtask {} completed aborting file: {}",
+            namespace, tableName, subtaskId, path);
+      } catch (Throwable t) {
+//                LOG.error(String.format("Iceberg writer %s.%s subtask %d failed to abort open file: %s",
+//                        namespace, tableName, subtaskId, path.toString()), t);
+        LOG.error("Iceberg writer {}.{} subtask {} failed to abort open file: {}. Throwable = {}",
+            namespace, tableName, subtaskId, path.toString(), t);
+        continue;
+      }
+
+      try {
+        LOG.debug("Iceberg writer {}.{} subtask {} deleting aborted file: {}",
+            namespace, tableName, subtaskId, path);
+        io.deleteFile(path.toString());
+        LOG.info("Iceberg writer {}.{} subtask {} deleted aborted file: {}",
+            namespace, tableName, subtaskId, path);
+      } catch (Throwable t) {
+//                LOG.error(String.format(
+//                        "Iceberg writer %s.%s subtask %d failed to delete aborted file: %s",
+//                        namespace, tableName, subtaskId, path.toString()), t);
+        LOG.error("Iceberg writer {}.{} subtask {} failed to delete aborted file: {}. Throwable = {}",
+            namespace, tableName, subtaskId, path.toString(), t);
+      }
+    }
+    LOG.info("Iceberg writer {}.{} subtask {} aborted {} open files",
+        namespace, tableName, subtaskId, openPartitionFiles.size());
+    openPartitionFiles.clear();
+  }
+
+  @Override
+  public void processElement(StreamRecord<IN> element) throws Exception {
+    IN value = element.getValue();
+    try {
+      processInternal(value);
+    } catch (Exception t) {
+      if (!skipIncompatibleRecord) {
+        throw t;
+      }
+    }
+  }
+
+  @VisibleForTesting
+  void processInternal(IN value) throws Exception {
+    Record record = serializer.serialize(value, schema);
+    processRecord(record);
+  }
+
+  /**
+   * process element as {@link Record}
+   */
+  private void processRecord(Record record) throws Exception {
+    if (partitioner == null) {
+      partitioner = new RecordPartitioner(spec);
+    }
+    partitioner.partition(record);
+    final String partitionPath = locations.newDataLocation(spec, partitioner, "");
+    if (!openPartitionFiles.containsKey(partitionPath)) {
+      final Path path = new Path(partitionPath, generateFileName());

Review comment:
       I wonder whether we have a max number of `openPartitionFiles` here, because this may introduce performance regression when too many files are procesed. (random reads/writes)




----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org



---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org

[GitHub] [incubator-iceberg] waterlx commented on a change in pull request #856: [WIP] Flink Iceberg sink

[GitBox <gi...@apache.org>]

waterlx commented on a change in pull request #856:
URL: https://github.com/apache/incubator-iceberg/pull/856#discussion_r411995834



##########
File path: build.gradle
##########
@@ -406,7 +407,91 @@ project(':iceberg-spark-runtime') {
     relocate 'io.netty.buffer', 'org.apache.iceberg.shaded.io.netty.buffer'
     relocate 'org.apache.arrow', 'org.apache.iceberg.shaded.org.apache.arrow'
 
-    archiveName = "iceberg-spark-runtime-${version}.${extension}"
+    archiveName = "iceberg-spark-runtime.${extension}"
+  }
+}
+
+project(':iceberg-flink') {
+  apply plugin: 'com.commercehub.gradle.plugin.avro'
+
+  dependencies {
+    compile project(':iceberg-api')
+    compile project(':iceberg-core')
+    compile project(':iceberg-hive')
+    compile project(':iceberg-parquet')
+
+    compileOnly("com.amazonaws:aws-java-sdk-s3")

Review comment:
       Addreseed.




----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org



---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org

[GitHub] [incubator-iceberg] waterlx commented on issue #856: [WIP] Flink Iceberg sink

[GitBox <gi...@apache.org>]

waterlx commented on issue #856:
URL: https://github.com/apache/incubator-iceberg/pull/856#issuecomment-617259227


   Closed by mistake and re-open it


----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org



---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org

[GitHub] [incubator-iceberg] waterlx edited a comment on pull request #856: [WIP] Flink Iceberg sink

[GitBox <gi...@apache.org>]

waterlx edited a comment on pull request #856:
URL: https://github.com/apache/incubator-iceberg/pull/856#issuecomment-624800409


   @stevenzwu @JingsongLi Really appreciate your effort to review this PR.  I am working on the comments you provided.
   Meanwhile, our team are focusing on verifying/testing the sink and I am in process of uploading the bug fixes/enhancements into this PR one after another.
   
   **Progress update:**
   The comments by @rdblue @openinx and @jerryshao regarding the initial version of this PR have been almost all addressed. 
   Begin to work on the following items:
   - Addressing the comments added recnetly by @stevenzwu and @JingsongLi 
   - Testing and bug fixing
   - Adding UT
   - Polishing the code and working on check-style issues
   
   The PR is being updated and always a workable version (against the master branch) with functions verified. So if you are interested, feel free to download the PR and get it compiled with master branch. [This design doc](https://docs.google.com/document/d/19M-sP6FlTVm7BV7MM4Om1n_MVo1xCy7GyDl_9ZAjVNQ/edit) includes the sample Java code to give it a try  8-)
   
   Drop me a mail or add any comments if it does not work.
   
   @jerryshao @chenjunjiedada @openinx @aokolnychyi @ @bowenli86 @stevenzwu @JingsongLi @rdblue FYI


----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org



---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org

[GitHub] [incubator-iceberg] stevenzwu commented on a change in pull request #856: [WIP] Flink Iceberg sink

[GitBox <gi...@apache.org>]

stevenzwu commented on a change in pull request #856:
URL: https://github.com/apache/incubator-iceberg/pull/856#discussion_r426845330



##########
File path: flink/src/main/java/org/apache/iceberg/flink/connector/sink/IcebergWriter.java
##########
@@ -0,0 +1,330 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iceberg.flink.connector.sink;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Locale;
+import java.util.Map;
+import java.util.UUID;
+import java.util.concurrent.TimeUnit;
+import javax.annotation.Nullable;
+import org.apache.flink.annotation.VisibleForTesting;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.runtime.state.StateInitializationContext;
+import org.apache.flink.streaming.api.operators.AbstractStreamOperator;
+import org.apache.flink.streaming.api.operators.ChainingStrategy;
+import org.apache.flink.streaming.api.operators.OneInputStreamOperator;
+import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
+import org.apache.flink.streaming.runtime.tasks.ProcessingTimeService;
+import org.apache.hadoop.fs.Path;
+import org.apache.iceberg.FileFormat;
+import org.apache.iceberg.MetricsConfig;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.data.Record;
+import org.apache.iceberg.data.parquet.GenericParquetWriter;
+import org.apache.iceberg.exceptions.RuntimeIOException;
+import org.apache.iceberg.flink.connector.IcebergConnectorConstant;
+import org.apache.iceberg.io.FileAppender;
+import org.apache.iceberg.io.FileIO;
+import org.apache.iceberg.io.LocationProvider;
+import org.apache.iceberg.io.OutputFile;
+import org.apache.iceberg.parquet.Parquet;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import static org.apache.iceberg.TableProperties.DEFAULT_FILE_FORMAT;
+import static org.apache.iceberg.TableProperties.DEFAULT_FILE_FORMAT_DEFAULT;
+
+@SuppressWarnings("checkstyle:ClassTypeParameterName")
+public class IcebergWriter<IN> extends AbstractStreamOperator<FlinkDataFile>
+    implements OneInputStreamOperator<IN, FlinkDataFile> {
+
+  private static final Logger LOG = LoggerFactory.getLogger(IcebergWriter.class);
+
+  private final RecordSerializer<IN> serializer;
+  private final String namespace;
+  private final String tableName;
+  private final FileFormat format;
+  private final boolean skipIncompatibleRecord;
+  private final Schema schema;
+  private final PartitionSpec spec;
+  private final LocationProvider locations;
+  private final FileIO io;
+  private final Map<String, String> properties;
+  private final String timestampFeild;
+  private final TimeUnit timestampUnit;
+  private final long maxFileSize;
+
+  private transient org.apache.hadoop.conf.Configuration hadoopConf;
+  private transient Map<String, FileWriter> openPartitionFiles;
+  private transient int subtaskId;
+  private transient ProcessingTimeService timerService;
+  private transient Partitioner<Record> partitioner;
+
+  public IcebergWriter(Table table, @Nullable RecordSerializer<IN> serializer, Configuration config) {
+    this.serializer = serializer;
+    skipIncompatibleRecord = config.getBoolean(IcebergConnectorConstant.SKIP_INCOMPATIBLE_RECORD,
+        IcebergConnectorConstant.DEFAULT_SKIP_INCOMPATIBLE_RECORD);
+    timestampFeild = config.getString(IcebergConnectorConstant.WATERMARK_TIMESTAMP_FIELD, "");
+    timestampUnit = TimeUnit.valueOf(config.getString(IcebergConnectorConstant.WATERMARK_TIMESTAMP_UNIT,
+        IcebergConnectorConstant.DEFAULT_WATERMARK_TIMESTAMP_UNIT));
+    maxFileSize = config.getLong(IcebergConnectorConstant.MAX_FILE_SIZE,
+        IcebergConnectorConstant.DEFAULT_MAX_FILE_SIZE);
+
+    namespace = config.getString(IcebergConnectorConstant.NAMESPACE, "");
+    tableName = config.getString(IcebergConnectorConstant.TABLE, "");
+
+    schema = table.schema();
+    spec = table.spec();
+    locations = table.locationProvider();
+    io = table.io();
+    properties = table.properties();
+    format = FileFormat.valueOf(
+        properties.getOrDefault(DEFAULT_FILE_FORMAT, DEFAULT_FILE_FORMAT_DEFAULT).toUpperCase(Locale.ENGLISH));
+
+    LOG.info("Iceberg writer {}.{} data file location: {}",
+        namespace, tableName, locations.newDataLocation(""));
+    LOG.info("Iceberg writer {}.{} created with sink config", namespace, tableName);
+    LOG.info("Iceberg writer {}.{} loaded table: schema = {}\npartition spec = {}",
+        namespace, tableName, schema, spec);
+
+    // default ChainingStrategy is set to HEAD
+    // we prefer chaining to avoid the huge serialization and deserializatoin overhead.
+    super.setChainingStrategy(ChainingStrategy.ALWAYS);
+  }
+
+  @Override
+  public void open() throws Exception {
+    super.open();
+
+    hadoopConf = new org.apache.hadoop.conf.Configuration();
+    subtaskId = getRuntimeContext().getIndexOfThisSubtask();
+    timerService = getProcessingTimeService();
+    openPartitionFiles = new HashMap<>();
+  }
+
+  @Override
+  public void prepareSnapshotPreBarrier(long checkpointId) throws Exception {
+    LOG.info("Iceberg writer {}.{} subtask {} begin preparing for checkpoint {}",
+        namespace, tableName, subtaskId, checkpointId);
+    // close all open files and emit files to downstream committer operator
+    flush(true);
+    LOG.info("Iceberg writer {}.{} subtask {} completed preparing for checkpoint {}",
+        namespace, tableName, subtaskId, checkpointId);
+  }
+
+  @Override
+  public void initializeState(StateInitializationContext context) throws Exception {
+    super.initializeState(context);
+  }
+
+  @VisibleForTesting
+  List<FlinkDataFile> flush(boolean emit) throws IOException {
+    List<FlinkDataFile> dataFiles = new ArrayList<>(openPartitionFiles.size());
+    for (Map.Entry<String, FileWriter> entry : openPartitionFiles.entrySet()) {
+      FileWriter writer = entry.getValue();
+      FlinkDataFile flinkDataFile = closeWriter(writer);
+      dataFiles.add(flinkDataFile);
+      if (emit) {
+        emit(flinkDataFile);
+      }
+    }
+    LOG.info("Iceberg writer {}.{} subtask {} flushed {} open files",
+        namespace, tableName, subtaskId, openPartitionFiles.size());
+    openPartitionFiles.clear();
+    return dataFiles;
+  }
+
+  FlinkDataFile closeWriter(FileWriter writer) throws IOException {
+    FlinkDataFile flinkDataFile = writer.close();
+    LOG.info(
+        "Iceberg writer {}.{} subtask {} uploaded to Iceberg table {}.{} with {} records and {} bytes on this path: {}",
+        namespace, tableName, subtaskId, namespace, tableName, flinkDataFile.getIcebergDataFile().recordCount(),
+        flinkDataFile.getIcebergDataFile().fileSizeInBytes(), flinkDataFile.getIcebergDataFile().path());
+    return flinkDataFile;
+  }
+
+  void emit(FlinkDataFile flinkDataFile) {
+    output.collect(new StreamRecord<>(flinkDataFile));
+    LOG.debug("Iceberg writer {}.{} subtask {} emitted uploaded file to committer for Iceberg table {}.{}" +
+        " with {} records and {} bytes on this path: {}",
+        namespace, tableName, subtaskId, namespace, tableName, flinkDataFile.getIcebergDataFile().recordCount(),
+        flinkDataFile.getIcebergDataFile().fileSizeInBytes(), flinkDataFile.getIcebergDataFile().path());
+  }
+
+  @Override
+  public void close() throws Exception {
+    super.close();
+
+    LOG.info("Iceberg writer {}.{} subtask {} begin close", namespace, tableName, subtaskId);
+    // close all open files without emitting to downstream committer
+    flush(false);
+    LOG.info("Iceberg writer {}.{} subtask {} completed close", namespace, tableName, subtaskId);
+  }
+
+  @Override
+  public void dispose() throws Exception {
+    super.dispose();
+
+    LOG.info("Iceberg writer {}.{} subtask {} begin dispose", namespace, tableName, subtaskId);
+    abort();
+    LOG.info("Iceberg writer {}.{} subtask {} completed dispose", namespace, tableName, subtaskId);
+  }
+
+  private void abort() {
+    LOG.info("Iceberg writer {}.{} subtask {} has {} open files to abort",
+        namespace, tableName, subtaskId, openPartitionFiles.size());
+    // close all open files without sending DataFile list to downstream committer operator.
+    // because there are not checkpointed,
+    // we don't want to commit these files.
+    for (Map.Entry<String, FileWriter> entry : openPartitionFiles.entrySet()) {
+      final FileWriter writer = entry.getValue();
+      final Path path = writer.getPath();
+      try {
+        LOG.debug("Iceberg writer {}.{} subtask {} start to abort file: {}",
+            namespace, tableName, subtaskId, path);
+        writer.abort();
+        LOG.info("Iceberg writer {}.{} subtask {} completed aborting file: {}",
+            namespace, tableName, subtaskId, path);
+      } catch (Throwable t) {
+//                LOG.error(String.format("Iceberg writer %s.%s subtask %d failed to abort open file: %s",
+//                        namespace, tableName, subtaskId, path.toString()), t);
+        LOG.error("Iceberg writer {}.{} subtask {} failed to abort open file: {}. Throwable = {}",
+            namespace, tableName, subtaskId, path.toString(), t);
+        continue;
+      }
+
+      try {
+        LOG.debug("Iceberg writer {}.{} subtask {} deleting aborted file: {}",
+            namespace, tableName, subtaskId, path);
+        io.deleteFile(path.toString());
+        LOG.info("Iceberg writer {}.{} subtask {} deleted aborted file: {}",
+            namespace, tableName, subtaskId, path);
+      } catch (Throwable t) {
+//                LOG.error(String.format(
+//                        "Iceberg writer %s.%s subtask %d failed to delete aborted file: %s",
+//                        namespace, tableName, subtaskId, path.toString()), t);
+        LOG.error("Iceberg writer {}.{} subtask {} failed to delete aborted file: {}. Throwable = {}",
+            namespace, tableName, subtaskId, path.toString(), t);
+      }
+    }
+    LOG.info("Iceberg writer {}.{} subtask {} aborted {} open files",
+        namespace, tableName, subtaskId, openPartitionFiles.size());
+    openPartitionFiles.clear();
+  }
+
+  @Override
+  public void processElement(StreamRecord<IN> element) throws Exception {
+    IN value = element.getValue();
+    try {
+      processInternal(value);
+    } catch (Exception t) {
+      if (!skipIncompatibleRecord) {
+        throw t;
+      }
+    }
+  }
+
+  @VisibleForTesting
+  void processInternal(IN value) throws Exception {
+    Record record = serializer.serialize(value, schema);
+    processRecord(record);
+  }
+
+  /**
+   * process element as {@link Record}
+   */
+  private void processRecord(Record record) throws Exception {
+    if (partitioner == null) {
+      partitioner = new RecordPartitioner(spec);
+    }
+    partitioner.partition(record);
+    final String partitionPath = locations.newDataLocation(spec, partitioner, "");
+    if (!openPartitionFiles.containsKey(partitionPath)) {
+      final Path path = new Path(partitionPath, generateFileName());

Review comment:
       there is no max limit. here the number of open files is largely determined by table partitioning spec. there are a few examples
   * partition by event time and there is long tail problem for event time
   * partition by some high cardinality columns, like country.
   
   Main concern for large number of open files (per subtask) is the memory footprint, especially when writing as Parquet format. There is also concern of many small files.




----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org



---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org

[GitHub] [incubator-iceberg] waterlx commented on a change in pull request #856: [WIP] Flink Iceberg sink

[GitBox <gi...@apache.org>]

waterlx commented on a change in pull request #856:
URL: https://github.com/apache/incubator-iceberg/pull/856#discussion_r415655236



##########
File path: build.gradle
##########
@@ -406,7 +407,91 @@ project(':iceberg-spark-runtime') {
     relocate 'io.netty.buffer', 'org.apache.iceberg.shaded.io.netty.buffer'
     relocate 'org.apache.arrow', 'org.apache.iceberg.shaded.org.apache.arrow'
 
-    archiveName = "iceberg-spark-runtime-${version}.${extension}"
+    archiveName = "iceberg-spark-runtime.${extension}"

Review comment:
       Addressed.




----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org



---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org

[GitHub] [incubator-iceberg] stevenzwu commented on a change in pull request #856: [WIP] Flink Iceberg sink

[GitBox <gi...@apache.org>]

stevenzwu commented on a change in pull request #856:
URL: https://github.com/apache/incubator-iceberg/pull/856#discussion_r420916720



##########
File path: flink/src/main/java/org/apache/iceberg/flink/connector/sink/IcebergSinkAppender.java
##########
@@ -0,0 +1,96 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iceberg.flink.connector.sink;
+
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.streaming.api.datastream.DataStream;
+import org.apache.flink.streaming.api.datastream.DataStreamSink;
+import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.data.Record;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Append Iceberg sink to DataStream
+ *
+ * @param <IN> input data type
+ */
+@SuppressWarnings({"checkstyle:ClassTypeParameterName", "checkstyle:HiddenField"})
+public class IcebergSinkAppender<IN> {
+  private static final Logger LOG = LoggerFactory.getLogger(IcebergSinkAppender.class);
+
+  private final Table table;
+  private final Configuration config;
+  private final String sinkName;
+  private RecordSerializer<IN> serializer;
+  private Integer writerParallelism;
+
+  public IcebergSinkAppender(Table table, Configuration config, String sinkName) {
+    this.table = table;
+    this.config = config;
+    this.sinkName = sinkName;
+  }
+
+  /**
+   * Required.
+   *
+   * @param serializer Serialize input data type to Iceberg {@link Record}
+   */
+  public IcebergSinkAppender<IN> withSerializer(RecordSerializer<IN> serializer) {
+    this.serializer = serializer;
+    return this;
+  }
+
+  /**
+   * Optional. Explicitly set the parallelism for Iceberg writer operator.
+   * Otherwise, default job parallelism is used.
+   */
+  public IcebergSinkAppender<IN> withWriterParallelism(Integer writerParallelism) {
+    this.writerParallelism = writerParallelism;
+    return this;
+  }
+
+  /**
+   * @param dataStream append sink to this DataStream
+   */
+  public DataStreamSink<FlinkDataFile> append(DataStream<IN> dataStream) {
+    // TODO: need to return?
+    IcebergWriter<IN> writer = new IcebergWriter<>(table, serializer, config);
+    IcebergCommitter committer = new IcebergCommitter(table, config);
+
+    final String writerId = sinkName + "-writer";
+    SingleOutputStreamOperator<FlinkDataFile> writerStream = dataStream
+        .transform(writerId, TypeInformation.of(FlinkDataFile.class), writer)  // IcebergWriter as stream operator
+        .uid(writerId);
+    if (null != writerParallelism && writerParallelism > 0) {
+      LOG.info("Set Iceberg writer parallelism to {}", writerParallelism);
+      writerStream.setParallelism(writerParallelism);

Review comment:
       if user doesn't set the parallelism of writer explicitly, default job parallelism is used. curious about how to get the parallelism of the upstream operator




----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org



---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org

[GitHub] [incubator-iceberg] waterlx commented on a change in pull request #856: [WIP] Flink Iceberg sink

[GitBox <gi...@apache.org>]

waterlx commented on a change in pull request #856:
URL: https://github.com/apache/incubator-iceberg/pull/856#discussion_r415588989



##########
File path: flink/src/main/java/org/apache/iceberg/flink/connector/sink/IcebergSinkAppender.java
##########
@@ -0,0 +1,96 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iceberg.flink.connector.sink;
+
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.streaming.api.datastream.DataStream;
+import org.apache.flink.streaming.api.datastream.DataStreamSink;
+import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.data.Record;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Append Iceberg sink to DataStream
+ *
+ * @param <IN> input data type
+ */
+@SuppressWarnings({"checkstyle:ClassTypeParameterName", "checkstyle:HiddenField"})
+public class IcebergSinkAppender<IN> {
+  private static final Logger LOG = LoggerFactory.getLogger(IcebergSinkAppender.class);
+
+  private final Table table;
+  private final Configuration config;
+  private final String sinkName;
+  private RecordSerializer<IN> serializer;
+  private Integer writerParallelism;
+
+  public IcebergSinkAppender(Table table, Configuration config, String sinkName) {
+    this.table = table;
+    this.config = config;
+    this.sinkName = sinkName;
+  }
+
+  /**
+   * Required.
+   *
+   * @param serializer Serialize input data type to Iceberg {@link Record}
+   */
+  public IcebergSinkAppender<IN> withSerializer(RecordSerializer<IN> serializer) {
+    this.serializer = serializer;
+    return this;
+  }
+
+  /**
+   * Optional. Explicitly set the parallelism for Iceberg writer operator.
+   * Otherwise, default job parallelism is used.
+   */
+  public IcebergSinkAppender<IN> withWriterParallelism(Integer writerParallelism) {
+    this.writerParallelism = writerParallelism;
+    return this;
+  }
+
+  /**
+   * @param dataStream append sink to this DataStream
+   */
+  public DataStreamSink<FlinkDataFile> append(DataStream<IN> dataStream) {
+    // TODO: need to return?
+    IcebergWriter<IN> writer = new IcebergWriter<>(table, serializer, config);
+    IcebergCommitter committer = new IcebergCommitter(table, config);
+
+    final String writerId = sinkName + "-writer";
+    SingleOutputStreamOperator<FlinkDataFile> writerStream = dataStream
+        .transform(writerId, TypeInformation.of(FlinkDataFile.class), writer)  // IcebergWriter as stream operator
+        .uid(writerId);
+    if (null != writerParallelism && writerParallelism > 0) {
+      LOG.info("Set Iceberg writer parallelism to {}", writerParallelism);
+      writerStream.setParallelism(writerParallelism);

Review comment:
       @JingsongLi Got you idea. Do you mean that if user does not set parallelism of writer explicitly, the logic could set it to the parallelism of the input stream of writer?




----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org



---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org

[GitHub] [incubator-iceberg] rdblue commented on pull request #856: [WIP] Flink Iceberg sink

[GitBox <gi...@apache.org>]

rdblue commented on pull request #856:
URL: https://github.com/apache/incubator-iceberg/pull/856#issuecomment-624805772


   For table serialization, I think the right thing to do is to serialize what is needed to load the table on the node that needs to commit. I'd prefer not to make that a single string if the user passes the namespace and table separately. We should serialize those instead of trying to parse and recover them.


----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org



---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org

[GitHub] [incubator-iceberg] waterlx commented on a change in pull request #856: [WIP] Flink Iceberg sink

[GitBox <gi...@apache.org>]

waterlx commented on a change in pull request #856:
URL: https://github.com/apache/incubator-iceberg/pull/856#discussion_r415621717



##########
File path: flink/src/main/java/org/apache/iceberg/flink/connector/IcebergConnectorConstant.java
##########
@@ -0,0 +1,62 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iceberg.flink.connector;
+
+import java.util.concurrent.TimeUnit;
+import org.apache.hadoop.hive.conf.HiveConf.ConfVars;
+
+public class IcebergConnectorConstant {
+  private IcebergConnectorConstant() {}
+
+  public static final String TYPE = "iceberg";
+
+  public static final String ICEBERG_APP_TYPE = "flink";
+
+  //public static final String CATALOG = "catalog";
+  public static final String DATABASE = "database";
+  public static final String TABLE = "table";
+
+  public static final String FORMAT = "format";
+  public static final String CODEC = "codec";
+
+  public static final String METACAT_HOST_HADOOP_CONF_KEY = ConfVars.METASTOREURIS.varname;  // For HiveCatalog
+  public static final String METACAT_HOST = "metacat_host";
+  public static final String DEFAULT_METACAT_HOST = "???";
+
+  public static final String SKIP_INCOMPATIBLE_RECORD = "skip_incompatible_record";
+  public static final String VTTS_WATERMARK_TIMESTAMP_FIELD = "vtts_watermark_timestamp_field";
+  public static final String VTTS_WATERMARK_TIMESTAMP_UNIT = "vtts_watermark_timestamp_unit";

Review comment:
       Addressed. VTTS is removed literally, but all "watermark" related logic is kept as it is.




----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org



---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org

[GitHub] [incubator-iceberg] waterlx commented on a change in pull request #856: [WIP] Flink Iceberg sink

[GitBox <gi...@apache.org>]

waterlx commented on a change in pull request #856:
URL: https://github.com/apache/incubator-iceberg/pull/856#discussion_r415607130



##########
File path: flink/src/main/java/org/apache/iceberg/flink/connector/model/CommitMetadataUtil.java
##########
@@ -0,0 +1,59 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iceberg.flink.connector.model;
+
+import java.io.ByteArrayOutputStream;
+import org.apache.avro.io.DatumWriter;
+import org.apache.avro.io.Encoder;
+import org.apache.avro.io.EncoderFactory;
+import org.apache.avro.specific.SpecificDatumWriter;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Companion class with CommitMetadata (Avro generated SpecificRecord)
+ */
+public class CommitMetadataUtil {

Review comment:
       Addressed.




----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org



---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org

[GitHub] [incubator-iceberg] waterlx commented on a change in pull request #856: [WIP] Flink Iceberg sink

[GitBox <gi...@apache.org>]

waterlx commented on a change in pull request #856:
URL: https://github.com/apache/incubator-iceberg/pull/856#discussion_r411996047



##########
File path: flink/src/main/java/org/apache/iceberg/flink/connector/sink/IcebergWriter.java
##########
@@ -0,0 +1,480 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iceberg.flink.connector.sink;
+
+import com.google.common.base.Strings;
+import com.google.common.collect.ImmutableMap;
+import com.netflix.spectator.api.DefaultRegistry;
+import com.netflix.spectator.api.Registry;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.UUID;
+import java.util.concurrent.TimeUnit;
+import javax.annotation.Nullable;
+import org.apache.avro.AvroTypeException;
+import org.apache.avro.Schema;
+import org.apache.avro.generic.IndexedRecord;
+import org.apache.flink.annotation.VisibleForTesting;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.runtime.state.StateInitializationContext;
+import org.apache.flink.streaming.api.operators.AbstractStreamOperator;
+import org.apache.flink.streaming.api.operators.ChainingStrategy;
+import org.apache.flink.streaming.api.operators.OneInputStreamOperator;
+import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
+import org.apache.flink.streaming.runtime.tasks.ProcessingTimeService;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.hive.conf.HiveConf.ConfVars;
+import org.apache.iceberg.FileFormat;
+import org.apache.iceberg.MetricsConfig;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.avro.AvroSchemaUtil;
+import org.apache.iceberg.catalog.Catalog;
+import org.apache.iceberg.catalog.TableIdentifier;
+import org.apache.iceberg.flink.connector.IcebergConnectorConstant;
+import org.apache.iceberg.hadoop.HadoopOutputFile;
+import org.apache.iceberg.hive.HiveCatalogs;
+import org.apache.iceberg.io.FileAppender;
+import org.apache.iceberg.io.OutputFile;
+import org.apache.iceberg.parquet.Parquet;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import static org.apache.iceberg.TableProperties.PARQUET_COMPRESSION;
+import static org.apache.iceberg.TableProperties.PARQUET_COMPRESSION_DEFAULT;
+
+public class IcebergWriter<T> extends AbstractStreamOperator<FlinkDataFile>
+    implements OneInputStreamOperator<T, FlinkDataFile> {
+
+  private static final Logger LOG = LoggerFactory.getLogger(IcebergWriter.class);
+  private static final String FILE_NAME_SEPARATOR = "_";
+
+  private final AvroSerializer serializer;
+  private final Configuration config;
+  private final String metacatHost;
+  //private final String jobName;
+  //private final String catalog;
+  private final String database;
+  private final String tableName;
+  private final FileFormat format;
+  private final boolean skipIncompatibleRecord;
+  private final org.apache.iceberg.Schema icebergSchema;
+  private final PartitionSpec spec;
+  private final String s3BasePath;
+  private final Map<String, String> tableProperties;
+  private final String timestampFeild;
+  private final TimeUnit timestampUnit;
+  private final long maxFileSize;
+
+  private transient String instanceId;
+  private transient String titusTaskId;
+  private transient Schema avroSchema;
+  private transient org.apache.hadoop.conf.Configuration hadoopConfig;
+  private transient Map<String, FileWriter> openPartitionFiles;
+  private transient int subtaskId;
+  private transient IcebergWriterSubtaskMetrics subtaskMetrics;
+  private transient IcebergWriterTaskMetrics taskMetrics;
+  private transient ProcessingTimeService timerService;
+  private transient Partitioner partitioner;
+  private transient FileSystem fs;
+
+  public IcebergWriter(@Nullable AvroSerializer<T> serializer,
+                       Configuration config) {
+    this.serializer = serializer;
+    this.config = config;
+    metacatHost = config.getString(IcebergConnectorConstant.METACAT_HOST,
+        IcebergConnectorConstant.DEFAULT_METACAT_HOST);
+    //jobName = config.getString(System.getenv("JOB_CLUSTER_NAME"), "");
+    //catalog = config.getString(IcebergConnectorConstant.CATALOG, "");
+    database = config.getString(IcebergConnectorConstant.DATABASE, "");
+    tableName = config.getString(IcebergConnectorConstant.TABLE, "");
+    format = FileFormat.valueOf(config.getString(IcebergConnectorConstant.FORMAT,
+        FileFormat.PARQUET.name()));
+    skipIncompatibleRecord = config.getBoolean(IcebergConnectorConstant.SKIP_INCOMPATIBLE_RECORD,
+        IcebergConnectorConstant.DEFAULT_SKIP_INCOMPATIBLE_RECORD);
+    // TODO: different from IcebergCommitter, line 147, in which, "" is taken as default
+    timestampFeild = config.getString(IcebergConnectorConstant.VTTS_WATERMARK_TIMESTAMP_FIELD,
+        IcebergConnectorConstant.DEFAULT_VTTS_WATERMARK_TIMESTAMP_UNIT);
+    timestampUnit = TimeUnit.valueOf(config.getString(IcebergConnectorConstant.VTTS_WATERMARK_TIMESTAMP_UNIT,
+        IcebergConnectorConstant.DEFAULT_VTTS_WATERMARK_TIMESTAMP_UNIT));
+    maxFileSize = config.getLong(IcebergConnectorConstant.MAX_FILE_SIZE,
+        IcebergConnectorConstant.DEFAULT_MAX_FILE_SIZE);
+
+//    org.apache.hadoop.conf.Configuration hadoopConfig = new org.apache.hadoop.conf.Configuration();
+//    hadoopConfig.set(IcebergConnectorConstant.METACAT_HOST_HADOOP_CONF_KEY, metacatHost);
+    // Avoid Netflix code just to make it compile
+    //final MetacatIcebergCatalog icebergCatalog
+    // = new MetacatIcebergCatalog(hadoopConfig, jobName, IcebergConnectorConstant.ICEBERG_APP_TYPE);
+    //final BaseMetastoreCatalog icebergCatalog = null;
+    //final TableIdentifier tableIdentifier = TableIdentifier.of(catalog, database, tableName);
+    // final Table table = icebergCatalog.loadTable(tableIdentifier);
+    org.apache.hadoop.conf.Configuration hadoopConf = new org.apache.hadoop.conf.Configuration();
+    //hadoopConf.set(ConfVars.METASTOREURIS.varname, config.getString(ConfVars.METASTOREURIS.varname, ""));
+    hadoopConf.set(IcebergConnectorConstant.METACAT_HOST_HADOOP_CONF_KEY, metacatHost);
+    hadoopConf.set(ConfVars.METASTOREWAREHOUSE.varname, config.getString(ConfVars.METASTOREWAREHOUSE.varname, ""));
+
+    Catalog icebergCatalog = HiveCatalogs.loadCatalog(hadoopConf);
+    final Table table = icebergCatalog.loadTable(TableIdentifier.of(database, tableName));
+    ImmutableMap.Builder<String, String> tablePropsBuilder = ImmutableMap.<String, String>builder()
+        .putAll(table.properties());
+    if (!table.properties().containsKey(PARQUET_COMPRESSION)) {
+      // if compression is not set in table properties,
+      // Flink writer defaults it to BROTLI
+      //TODO: org.apache.hadoop.io.compress.BrotliCodec, class not found
+      //tablePropsBuilder.put(PARQUET_COMPRESSION, CompressionCodecName.BROTLI.name());
+      tablePropsBuilder.put(PARQUET_COMPRESSION, PARQUET_COMPRESSION_DEFAULT);
+    }
+    tableProperties = tablePropsBuilder.build();
+    icebergSchema = table.schema();
+    spec = table.spec();
+    //s3BasePath = getS3BasePath(table.location());
+    s3BasePath = table.locationProvider().newDataLocation("");  // data location of the Iceberg table
+    LOG.info("Iceberg writer {}.{} has S3 base path: {}", database, tableName, s3BasePath);
+    LOG.info("Iceberg writer {}.{} created with sink config", database, tableName);
+    LOG.info("Iceberg writer {}.{} loaded table: schema = {}\npartition spec = {}",
+        database, tableName, icebergSchema, spec);
+
+    // default ChainingStrategy is set to HEAD
+    // we prefer chaining to avoid the huge serialization and deserializatoin overhead.
+    super.setChainingStrategy(ChainingStrategy.ALWAYS);
+  }
+
+  /**
+   * @param location location from table metadata
+   *                 e.g. s3n://bucket/hive/warehouse/database_name.db/table_name
+   */
+  private String getS3BasePath(final String location) {

Review comment:
       Removed and addressed




----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org



---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org

[GitHub] [incubator-iceberg] stevenzwu commented on a change in pull request #856: [WIP] Flink Iceberg sink

[GitBox <gi...@apache.org>]

stevenzwu commented on a change in pull request #856:
URL: https://github.com/apache/incubator-iceberg/pull/856#discussion_r420914501



##########
File path: flink/src/main/java/org/apache/iceberg/flink/connector/sink/IcebergCommitter.java
##########
@@ -0,0 +1,642 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iceberg.flink.connector.sink;
+
+import com.google.common.base.Joiner;
+import com.google.common.base.Strings;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.stream.Collectors;
+import javax.annotation.Nullable;
+import org.apache.flink.annotation.VisibleForTesting;
+import org.apache.flink.api.common.JobExecutionResult;
+import org.apache.flink.api.common.JobID;
+import org.apache.flink.api.common.state.ListState;
+import org.apache.flink.api.common.state.ListStateDescriptor;
+import org.apache.flink.api.java.ExecutionEnvironment;
+import org.apache.flink.configuration.CheckpointingOptions;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.runtime.state.CheckpointListener;
+import org.apache.flink.runtime.state.FunctionInitializationContext;
+import org.apache.flink.runtime.state.FunctionSnapshotContext;
+import org.apache.flink.streaming.api.checkpoint.CheckpointedFunction;
+import org.apache.flink.streaming.api.environment.CheckpointConfig;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
+import org.apache.flink.util.Preconditions;
+import org.apache.hadoop.hive.conf.HiveConf.ConfVars;
+import org.apache.iceberg.AppendFiles;
+import org.apache.iceberg.DataFile;
+import org.apache.iceberg.ManifestFile;
+import org.apache.iceberg.ManifestWriter;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.Snapshot;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.Transaction;
+import org.apache.iceberg.UpdateProperties;
+import org.apache.iceberg.catalog.Catalog;
+import org.apache.iceberg.catalog.TableIdentifier;
+import org.apache.iceberg.flink.connector.IcebergConnectorConstant;
+import org.apache.iceberg.flink.connector.model.CommitMetadata;
+import org.apache.iceberg.flink.connector.model.CommitMetadataUtil;
+import org.apache.iceberg.flink.connector.model.FlinkManifestFile;
+import org.apache.iceberg.flink.connector.model.FlinkManifestFileUtil;
+import org.apache.iceberg.flink.connector.model.GenericFlinkManifestFile;
+import org.apache.iceberg.flink.connector.model.ManifestFileState;
+import org.apache.iceberg.hadoop.HadoopCatalog;
+import org.apache.iceberg.hive.HiveCatalogs;
+import org.apache.iceberg.io.FileIO;
+import org.apache.iceberg.io.OutputFile;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * This operator commit data files to Iceberg table.
+ * <p>
+ * This operator should always run with parallelism of 1.
+ * Because Iceberg lib perform optimistic concurrency control,
+ * this can help reduce contention and retries
+ * when committing files to Iceberg table.
+ * <p>
+ * Here are some known contentions
+ * 1) Flink jobs running in multiple regions,
+ * since Iceberg metadata service and commit only happens in us-east-1.
+ * 2) auto tuning and auto lift services may update Iceberg table infrequently.
+ */
+@SuppressWarnings("checkstyle:HiddenField")
+public class IcebergCommitter extends RichSinkFunction<FlinkDataFile>

Review comment:
       yeah. OperatorCoordinator seems like the better way to go. One drawback of the old two-stage operators model is that it can invalidate an embarrassingly parallel DAG. I brought this up to Becket Qin. This is the reverse/mirroring problem of FLIP-27 (source interface redesign)




----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org



---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org

[GitHub] [incubator-iceberg] waterlx commented on a change in pull request #856: [WIP] Flink Iceberg sink

[GitBox <gi...@apache.org>]

waterlx commented on a change in pull request #856:
URL: https://github.com/apache/incubator-iceberg/pull/856#discussion_r415622385



##########
File path: flink/src/main/java/org/apache/iceberg/flink/connector/sink/AvroUtils.java
##########
@@ -0,0 +1,68 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iceberg.flink.connector.sink;
+
+import java.util.List;
+import org.apache.avro.AvroTypeException;
+import org.apache.avro.Schema;
+
+public class AvroUtils {

Review comment:
       Addressed




----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org



---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org

[GitHub] [incubator-iceberg] stevenzwu commented on a change in pull request #856: [WIP] Flink Iceberg sink

[GitBox <gi...@apache.org>]

stevenzwu commented on a change in pull request #856:
URL: https://github.com/apache/incubator-iceberg/pull/856#discussion_r420932982



##########
File path: build.gradle
##########
@@ -442,6 +443,31 @@ project(':iceberg-spark-runtime') {
   }
 }
 
+project(':iceberg-flink') {
+  apply plugin: 'com.commercehub.gradle.plugin.avro'
+
+  dependencies {
+    compile project(':iceberg-api')
+    compile project(':iceberg-core')
+    compile project(':iceberg-data')
+    compile project(path: ':iceberg-data', configuration: 'testArtifacts')  // TODO: to remove, for RandomGenericData
+    compile project(':iceberg-hive')
+    compile project(':iceberg-parquet')
+
+    compileOnly("org.apache.flink:flink-streaming-java_2.11")

Review comment:
       maybe make the Scala version a gradle property. unfortunately, Flink doesn't have a clean Java API. even `flink-streaming-java` has a Scala version suffix in the jar name.
   
   @rdblue what is your take on Scala cross compile? I saw the `iceberg-spark`  module also has hard-coded `2.11` Scala suffix in the spark deps




----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org



---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org

[GitHub] [incubator-iceberg] JingsongLi commented on a change in pull request #856: [WIP] Flink Iceberg sink

[GitBox <gi...@apache.org>]

JingsongLi commented on a change in pull request #856:
URL: https://github.com/apache/incubator-iceberg/pull/856#discussion_r413681096



##########
File path: flink/src/main/java/org/apache/iceberg/flink/connector/sink/IcebergCommitter.java
##########
@@ -0,0 +1,642 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iceberg.flink.connector.sink;
+
+import com.google.common.base.Joiner;
+import com.google.common.base.Strings;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.stream.Collectors;
+import javax.annotation.Nullable;
+import org.apache.flink.annotation.VisibleForTesting;
+import org.apache.flink.api.common.JobExecutionResult;
+import org.apache.flink.api.common.JobID;
+import org.apache.flink.api.common.state.ListState;
+import org.apache.flink.api.common.state.ListStateDescriptor;
+import org.apache.flink.api.java.ExecutionEnvironment;
+import org.apache.flink.configuration.CheckpointingOptions;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.runtime.state.CheckpointListener;
+import org.apache.flink.runtime.state.FunctionInitializationContext;
+import org.apache.flink.runtime.state.FunctionSnapshotContext;
+import org.apache.flink.streaming.api.checkpoint.CheckpointedFunction;
+import org.apache.flink.streaming.api.environment.CheckpointConfig;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
+import org.apache.flink.util.Preconditions;
+import org.apache.hadoop.hive.conf.HiveConf.ConfVars;
+import org.apache.iceberg.AppendFiles;
+import org.apache.iceberg.DataFile;
+import org.apache.iceberg.ManifestFile;
+import org.apache.iceberg.ManifestWriter;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.Snapshot;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.Transaction;
+import org.apache.iceberg.UpdateProperties;
+import org.apache.iceberg.catalog.Catalog;
+import org.apache.iceberg.catalog.TableIdentifier;
+import org.apache.iceberg.flink.connector.IcebergConnectorConstant;
+import org.apache.iceberg.flink.connector.model.CommitMetadata;
+import org.apache.iceberg.flink.connector.model.CommitMetadataUtil;
+import org.apache.iceberg.flink.connector.model.FlinkManifestFile;
+import org.apache.iceberg.flink.connector.model.FlinkManifestFileUtil;
+import org.apache.iceberg.flink.connector.model.GenericFlinkManifestFile;
+import org.apache.iceberg.flink.connector.model.ManifestFileState;
+import org.apache.iceberg.hadoop.HadoopCatalog;
+import org.apache.iceberg.hive.HiveCatalogs;
+import org.apache.iceberg.io.FileIO;
+import org.apache.iceberg.io.OutputFile;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * This operator commit data files to Iceberg table.
+ * <p>
+ * This operator should always run with parallelism of 1.
+ * Because Iceberg lib perform optimistic concurrency control,
+ * this can help reduce contention and retries
+ * when committing files to Iceberg table.
+ * <p>
+ * Here are some known contentions
+ * 1) Flink jobs running in multiple regions,
+ * since Iceberg metadata service and commit only happens in us-east-1.
+ * 2) auto tuning and auto lift services may update Iceberg table infrequently.
+ */
+@SuppressWarnings("checkstyle:HiddenField")
+public class IcebergCommitter extends RichSinkFunction<FlinkDataFile>
+    implements CheckpointedFunction, CheckpointListener {
+  private static final Logger LOG = LoggerFactory.getLogger(IcebergCommitter.class);
+
+  private static final String COMMIT_REGION_KEY = "flink.commit.region";
+  private static final String COMMIT_MANIFEST_HASHES_KEY = "flink.commit.manifest.hashes";
+  private static final String VTTS_WATERMARK_PROP_KEY_PREFIX = "flink.watermark.";
+
+  private Configuration config;
+  private final String namespace;
+  private final String tableName;
+
+  private final String region;
+  private final boolean vttsWatermarkEnabled;
+  private final String vttsWatermarkPropKey;
+  private final long snapshotRetentionHours;
+  private final boolean commitRestoredManifestFiles;
+  private final String icebergManifestFileDir;
+  private final PartitionSpec spec;
+  private final FileIO io;
+  private final String flinkJobId;
+
+  private transient Table table;
+  private transient List<FlinkDataFile> pendingDataFiles;
+  private transient List<FlinkManifestFile> flinkManifestFiles;
+  private transient ListState<ManifestFileState> manifestFilesState;
+  private transient CommitMetadata metadata;
+  private transient ListState<CommitMetadata> metadataState;
+
+  public IcebergCommitter(Table table, Configuration config) {
+    this.config = config;
+
+    // current Iceberg sink implementation can't work with concurrent checkpoints.
+    // We disable concurrent checkpoints by default as min pause is set to 60s by default.
+    // Add an assertion to fail explicit in case job enables concurrent checkpoints.
+    CheckpointConfig checkpointConfig = StreamExecutionEnvironment.getExecutionEnvironment().getCheckpointConfig();
+    if (checkpointConfig.getMaxConcurrentCheckpoints() > 1) {
+      throw new IllegalArgumentException("Iceberg sink doesn't support concurrent checkpoints");
+    }
+
+    namespace = config.getString(IcebergConnectorConstant.NAMESPACE, "");
+    tableName = config.getString(IcebergConnectorConstant.TABLE, "");
+
+    String region = System.getenv("EC2_REGION");
+    if (Strings.isNullOrEmpty(region)) {
+      region = "us-east-1";
+      LOG.info("Iceberg committer {}.{} default region to us-east-1", namespace, tableName);
+    }
+    this.region = region;
+
+    vttsWatermarkEnabled = !Strings.isNullOrEmpty(
+        config.getString(IcebergConnectorConstant.VTTS_WATERMARK_TIMESTAMP_FIELD, ""));
+    vttsWatermarkPropKey = VTTS_WATERMARK_PROP_KEY_PREFIX + region;
+    snapshotRetentionHours = config.getLong(IcebergConnectorConstant.SNAPSHOT_RETENTION_HOURS,
+        IcebergConnectorConstant.DEFAULT_SNAPSHOT_RETENTION_HOURS);
+    commitRestoredManifestFiles = config.getBoolean(IcebergConnectorConstant.COMMIT_RESTORED_MANIFEST_FILES,
+        IcebergConnectorConstant.DEFAULT_COMMIT_RESTORED_MANIFEST_FILES);
+    icebergManifestFileDir = getIcebergManifestFileDir(config);
+
+    // The only final fields yielded by table inputted
+    spec = table.spec();
+    io = table.io();
+
+    final JobExecutionResult jobExecutionResult
+        = ExecutionEnvironment.getExecutionEnvironment().getLastJobExecutionResult();
+    if (jobExecutionResult != null) {
+      flinkJobId = jobExecutionResult.getJobID().toString();
+      LOG.info("Get Flink job ID from execution environment: {}", flinkJobId);
+    } else {
+      flinkJobId = new JobID().toString();
+      LOG.info("Execution environment doesn't have executed job. Generate a random job ID : {}", flinkJobId);
+    }
+    LOG.info("Iceberg committer {}.{} created with sink config", namespace, tableName);
+    LOG.info("Iceberg committer {}.{} loaded table partition spec: {}", namespace, tableName, spec);
+  }
+
+  @VisibleForTesting
+  List<FlinkDataFile> getPendingDataFiles() {
+    return pendingDataFiles;
+  }
+
+  @VisibleForTesting
+  List<FlinkManifestFile> getFlinkManifestFiles() {
+    return flinkManifestFiles;
+  }
+
+  @VisibleForTesting
+  CommitMetadata getMetadata() {
+    return metadata;
+  }
+
+  private String getIcebergManifestFileDir(Configuration config) {
+    final String checkpointDir = config.getString(
+        CheckpointingOptions.CHECKPOINTS_DIRECTORY, null);
+    if (null == checkpointDir) {
+      throw new IllegalArgumentException("checkpoint dir is null");
+    }
+
+    return String.format("%s/iceberg/manifest/", checkpointDir);
+  }
+
+  @Override
+  public void close() throws Exception {
+    super.close();
+  }
+
+  void init() {
+    // TODO: duplicate logic, to extract
+    org.apache.hadoop.conf.Configuration hadoopConf = new org.apache.hadoop.conf.Configuration();
+    String catalogType = config.getString(IcebergConnectorConstant.CATALOG_TYPE,
+                                          IcebergConnectorConstant.CATALOG_TYPE_DEFAULT);
+    Catalog catalog = null;
+    switch (catalogType.toUpperCase()) {
+      case IcebergConnectorConstant.HIVE_CATALOG:
+        hadoopConf.set(ConfVars.METASTOREURIS.varname, config.getString(ConfVars.METASTOREURIS.varname, ""));
+        catalog = HiveCatalogs.loadCatalog(hadoopConf);
+        break;
+
+      case IcebergConnectorConstant.HADOOP_CATALOG:
+        catalog = new HadoopCatalog(hadoopConf,
+                                    config.getString(IcebergConnectorConstant.HADOOP_CATALOG_WAREHOUSE_LOCATION, ""));
+        break;
+
+      default:
+        throw new UnsupportedOperationException("Unknown catalog type or not set: " + catalogType);
+    }
+
+    this.table = catalog.loadTable(TableIdentifier.parse(namespace + "." + tableName));
+
+    pendingDataFiles = new ArrayList<>();
+    flinkManifestFiles = new ArrayList<>();
+    metadata = CommitMetadata.newBuilder()
+        .setLastCheckpointId(0)
+        .setLastCheckpointTimestamp(0)
+        .setLastCommitTimestamp(System.currentTimeMillis())
+        .build();
+  }
+
+  @Override
+  public void initializeState(FunctionInitializationContext context) throws Exception {
+    init();
+
+    Preconditions.checkState(manifestFilesState == null,
+        "checkpointedFilesState has already been initialized.");
+    Preconditions.checkState(metadataState == null,
+        "metadataState has already been initialized.");
+    manifestFilesState = context.getOperatorStateStore().getListState(new ListStateDescriptor<>(
+        "iceberg-committer-manifest-files-state", ManifestFileState.class));
+    metadataState = context.getOperatorStateStore().getListState(new ListStateDescriptor<>(
+        "iceberg-committer-metadata-state", CommitMetadata.class));
+
+    if (context.isRestored()) {
+      final Iterable<CommitMetadata> restoredMetadata = metadataState.get();
+      if (null != restoredMetadata) {
+        LOG.info("Iceberg committer {}.{} restoring metadata", namespace, tableName);
+        List<CommitMetadata> metadataList = new ArrayList<>();
+        for (CommitMetadata entry : restoredMetadata) {
+          metadataList.add(entry);
+        }
+        Preconditions.checkState(1 == metadataList.size(),
+            "metadata list size should be 1. got " + metadataList.size());
+        metadata = metadataList.get(0);
+        LOG.info("Iceberg committer {}.{} restored metadata: {}",
+            namespace, tableName, CommitMetadataUtil.getInstance().encodeAsJson(metadata));
+      } else {
+        LOG.info("Iceberg committer {}.{} has nothing to restore for metadata", namespace, tableName);
+      }
+
+      Iterable<ManifestFileState> restoredManifestFileStates = manifestFilesState.get();
+      if (null != restoredManifestFileStates) {
+        LOG.info("Iceberg committer {}.{} restoring manifest files",
+            namespace, tableName);
+        for (ManifestFileState manifestFileState : restoredManifestFileStates) {
+          flinkManifestFiles.add(GenericFlinkManifestFile.fromState(manifestFileState));
+        }
+        LOG.info("Iceberg committer {}.{} restored {} manifest files: {}",
+            namespace, tableName, flinkManifestFiles.size(), flinkManifestFiles);
+        final long now = System.currentTimeMillis();
+        if (now - metadata.getLastCheckpointTimestamp() > TimeUnit.HOURS.toMillis(snapshotRetentionHours)) {
+          flinkManifestFiles.clear();
+          LOG.info("Iceberg committer {}.{} cleared restored manifest files as checkpoint timestamp is too old: " +
+                  "checkpointTimestamp = {}, now = {}, snapshotRetentionHours = {}",
+              namespace, tableName, metadata.getLastCheckpointTimestamp(), now, snapshotRetentionHours);
+        } else {
+          flinkManifestFiles = removeCommittedManifests(flinkManifestFiles);
+          if (flinkManifestFiles.isEmpty()) {
+            LOG.info("Iceberg committer {}.{} has zero uncommitted manifest files from restored state",
+                namespace, tableName);
+          } else {
+            if (commitRestoredManifestFiles) {
+              commitRestoredManifestFiles();
+            } else {
+              LOG.info("skip commit of restored manifest files");
+            }
+          }
+        }
+      } else {
+        LOG.info("Iceberg committer {}.{} has nothing to restore for manifest files", namespace, tableName);
+      }
+    }
+  }
+
+  @VisibleForTesting
+  void commitRestoredManifestFiles() throws Exception {
+    LOG.info("Iceberg committer {}.{} committing last uncompleted transaction upon recovery: " +
+            "metadata = {}, flink manifest files ({}) = {}", namespace, tableName,
+        CommitMetadataUtil.getInstance().encodeAsJson(metadata),
+        flinkManifestFiles.size(), flinkManifestFiles);
+    commit();
+    LOG.info("Iceberg committer {}.{} committed last uncompleted transaction upon recovery: " +
+            "metadata = {}, flink manifest files ({}) = {}", namespace, tableName,
+        CommitMetadataUtil.getInstance().encodeAsJson(metadata),
+        flinkManifestFiles.size(), flinkManifestFiles);
+    postCommitSuccess();
+  }
+
+  private List<FlinkManifestFile> removeCommittedManifests(List<FlinkManifestFile> flinkManifestFiles) {
+    int snapshotCount = 0;
+    String result = "succeeded";
+    final long start = System.currentTimeMillis();
+    try {
+      final Set<String> manifestHashes = flinkManifestFiles.stream()
+          .map(f -> f.hash())
+          .collect(Collectors.toSet());
+      final Set<String> committedHashes = new HashSet<>(flinkManifestFiles.size());
+      final Iterable<Snapshot> snapshots = table.snapshots();
+      for (Snapshot snapshot : snapshots) {
+        ++snapshotCount;
+        final Map<String, String> summary = snapshot.summary();
+        final List<String> hashes = FlinkManifestFileUtil.hashesStringToList(summary.get(COMMIT_MANIFEST_HASHES_KEY));
+        for (String hash : hashes) {
+          if (manifestHashes.contains(hash)) {
+            committedHashes.add(hash);
+          }
+        }
+      }
+      final List<FlinkManifestFile> uncommittedManifestFiles = flinkManifestFiles.stream()
+          .filter(f -> !committedHashes.contains(f.hash()))
+          .collect(Collectors.toList());
+      return uncommittedManifestFiles;
+    } catch (Throwable t) {
+      result = "failed";
+      //LOG.error(String.format("Iceberg committer %s.%s failed to check transaction completed", database, tableName),
+      //          t);
+      LOG.error("Iceberg committer {}.{} failed to check transaction completed. Throwable = {}",
+                namespace, tableName, t);
+      throw t;
+    } finally {
+      final long duration = System.currentTimeMillis() - start;
+      LOG.info("Iceberg committer {}.{} {} to check transaction completed" +
+               " after iterating {} snapshots and {} milli-seconds",
+               namespace, tableName, result, snapshotCount, duration);
+    }
+  }
+
+  @Override
+  public void snapshotState(FunctionSnapshotContext context) throws Exception {
+    LOG.info("Iceberg committer {}.{} snapshot state: checkpointId = {}, triggerTime = {}",
+        namespace, tableName, context.getCheckpointId(), context.getCheckpointTimestamp());
+    Preconditions.checkState(null != manifestFilesState,
+        "manifest files state has not been properly initialized.");
+    Preconditions.checkState(null != metadataState,
+        "metadata state has not been properly initialized.");
+
+    // set transaction to null to indicate a start of a new checkpoint/commit/transaction
+    synchronized (this) {
+      snapshot(context, pendingDataFiles);
+      checkpointState(flinkManifestFiles, metadata);
+      postSnapshotSuccess();
+    }
+  }
+
+  @VisibleForTesting
+  void snapshot(FunctionSnapshotContext context, List<FlinkDataFile> pendingDataFiles) throws Exception {
+    FlinkManifestFile flinkManifestFile = null;
+    if (!pendingDataFiles.isEmpty()) {
+      flinkManifestFile = createManifestFile(context, pendingDataFiles);
+      flinkManifestFiles.add(flinkManifestFile);
+    }
+    metadata = updateMetadata(metadata, context, flinkManifestFile);
+  }
+
+  private FlinkManifestFile createManifestFile(
+      FunctionSnapshotContext context, List<FlinkDataFile> pendingDataFiles) throws Exception {
+    LOG.info("Iceberg committer {}.{} checkpointing {} pending data files}",
+        namespace, tableName, pendingDataFiles.size());
+    String result = "succeeded";
+    final long start = System.currentTimeMillis();
+    try {
+      final String manifestFileName = Joiner.on("_")
+          .join(flinkJobId, context.getCheckpointId(), context.getCheckpointTimestamp());
+      // Iceberg requires file format suffix right now
+      final String manifestFileNameWithSuffix = manifestFileName + ".avro";
+      OutputFile outputFile = io.newOutputFile(icebergManifestFileDir + manifestFileNameWithSuffix);
+      ManifestWriter manifestWriter = ManifestWriter.write(spec, outputFile);
+
+      // stats
+      long recordCount = 0;
+      long byteCount = 0;
+      long lowWatermark = Long.MAX_VALUE;
+      long highWatermark = Long.MIN_VALUE;
+      for (FlinkDataFile flinkDataFile : pendingDataFiles) {
+        DataFile dataFile = flinkDataFile.getIcebergDataFile();
+        manifestWriter.add(dataFile);
+        // update stas
+        recordCount += dataFile.recordCount();
+        byteCount += dataFile.fileSizeInBytes();
+        if (flinkDataFile.getLowWatermark() < lowWatermark) {
+          lowWatermark = flinkDataFile.getLowWatermark();
+        }
+        if (flinkDataFile.getHighWatermark() > highWatermark) {
+          highWatermark = flinkDataFile.getHighWatermark();
+        }
+        LOG.debug("Data file with size of {} bytes added to manifest", dataFile.fileSizeInBytes());
+      }
+      manifestWriter.close();
+      ManifestFile manifestFile = manifestWriter.toManifestFile();
+
+      FlinkManifestFile flinkManifestFile = GenericFlinkManifestFile.builder()
+          .setPath(manifestFile.path())
+          .setLength(manifestFile.length())
+          .setSpecId(manifestFile.partitionSpecId())
+          .setCheckpointId(context.getCheckpointId())
+          .setCheckpointTimestamp(context.getCheckpointTimestamp())
+          .setDataFileCount(pendingDataFiles.size())
+          .setRecordCount(recordCount)
+          .setByteCount(byteCount)
+          .setLowWatermark(lowWatermark)
+          .setHighWatermark(highWatermark)
+          .build();
+
+      // don't want to log a giant list at one line.
+      // split the complete list into smaller chunks with 50 files.
+      final AtomicInteger counter = new AtomicInteger(0);
+      Collection<List<String>> listOfFileList = pendingDataFiles.stream()
+          .map(flinkDataFile -> flinkDataFile.toCompactDump())
+          .collect(Collectors.groupingBy(it -> counter.getAndIncrement() / 50))
+          .values();
+      for (List<String> fileList : listOfFileList) {
+        LOG.info("Iceberg committer {}.{} created manifest file {} for {}/{} data files: {}",
+            namespace, tableName, manifestFile.path(), fileList.size(), pendingDataFiles.size(), fileList);
+      }
+      return flinkManifestFile;
+    } catch (Throwable t) {
+      result = "failed";
+      //LOG.error(String.format("Iceberg committer %s.%s failed to create manifest file for %d pending data files",
+      //          database, tableName, pendingDataFiles.size()), t);
+      LOG.error("Iceberg committer {}.{} failed to create manifest file for {} pending data files. Throwable={}",
+          namespace, tableName, pendingDataFiles.size(), t);
+      throw t;
+    } finally {
+      final long duration = System.currentTimeMillis() - start;
+      LOG.info("Iceberg committer {}.{} {} to create manifest file with {} data files after {} milli-seconds",
+          namespace, tableName, result, pendingDataFiles.size(), duration);
+    }
+  }
+
+  private CommitMetadata updateMetadata(
+      CommitMetadata oldMetadata,
+      FunctionSnapshotContext context,
+      @Nullable FlinkManifestFile flinkManifestFile) {
+    LOG.info("Iceberg committer {}.{} updating metadata {} with manifest file {}",
+        namespace, tableName, CommitMetadataUtil.getInstance().encodeAsJson(oldMetadata), flinkManifestFile);
+    CommitMetadata.Builder metadataBuilder = CommitMetadata.newBuilder(oldMetadata)
+        .setLastCheckpointId(context.getCheckpointId())
+        .setLastCheckpointTimestamp(context.getCheckpointTimestamp());
+    if (vttsWatermarkEnabled) {
+      Long vttsWatermark = oldMetadata.getVttsWatermark();
+      if (null == flinkManifestFile) {
+        // DPS-412: when there is no data to be committed
+        // use elapsed wall clock time to move the VTTS watermark forward.
+        if (null != vttsWatermark) {
+          final long elapsedTimeMs = System.currentTimeMillis() - oldMetadata.getLastCommitTimestamp();
+          vttsWatermark += elapsedTimeMs;
+        } else {
+          vttsWatermark = System.currentTimeMillis();
+        }
+      } else {
+        // use lowWatermark.
+        if (null == flinkManifestFile.lowWatermark()) {
+          throw new IllegalArgumentException("VTTS is enabled but lowWatermark is null");
+        }
+        // in case one container/slot is lagging behind,
+        // we want to move watermark forward based on the slowest.
+        final long newWatermark = flinkManifestFile.lowWatermark();
+        // make sure VTTS watermark doesn't go back in time
+        if (null == vttsWatermark || newWatermark > vttsWatermark) {
+          vttsWatermark = newWatermark;
+        }
+      }
+      metadataBuilder.setVttsWatermark(vttsWatermark);
+    }
+    CommitMetadata metadata = metadataBuilder.build();
+    LOG.info("Iceberg committer {}.{} updated metadata {} with manifest file {}",
+        namespace, tableName, CommitMetadataUtil.getInstance().encodeAsJson(metadata), flinkManifestFile);
+    return metadata;
+  }
+
+  private void checkpointState(List<FlinkManifestFile> flinkManifestFiles, CommitMetadata metadata) throws Exception {
+    LOG.info("Iceberg committer {}.{} checkpointing state", namespace, tableName);
+    List<ManifestFileState> manifestFileStates = flinkManifestFiles.stream()
+        .map(f -> f.toState())
+        .collect(Collectors.toList());
+    manifestFilesState.clear();
+    manifestFilesState.addAll(manifestFileStates);
+    metadataState.clear();
+    metadataState.add(metadata);
+    LOG.info("Iceberg committer {}.{} checkpointed state: metadata = {}, flinkManifestFiles({}) = {}",
+        namespace, tableName, CommitMetadataUtil.getInstance().encodeAsJson(metadata),
+        flinkManifestFiles.size(), flinkManifestFiles);
+  }
+
+  private void postSnapshotSuccess() {
+    pendingDataFiles.clear();
+    LOG.debug("Un-committed manifest file count {}, containing data file count {}, record count {} and byte count {}",
+        flinkManifestFiles.size(),
+        FlinkManifestFileUtil.getDataFileCount(flinkManifestFiles),
+        FlinkManifestFileUtil.getRecordCount(flinkManifestFiles),
+        FlinkManifestFileUtil.getByteCount(flinkManifestFiles)
+    );
+  }
+
+  @Override
+  public void notifyCheckpointComplete(long checkpointId) {
+    LOG.info("Iceberg committer {}.{} checkpoint {} completed", namespace, tableName, checkpointId);
+    synchronized (this) {
+      if (checkpointId == metadata.getLastCheckpointId()) {
+        try {
+          commit();
+          postCommitSuccess();
+        } catch (Exception t) {
+          // swallow the exception to avoid job restart in case of commit failure
+          //LOG.error(String.format("Iceberg committer %s.%s failed to do post checkpoint commit", database, tableName),
+          //          t);
+          LOG.error("Iceberg committer {}.{} failed to do post checkpoint commit. Throwable = ",
+              namespace, tableName, t);
+        }
+      } else {
+        // TODO: it would be nice to fix this and allow concurrent checkpoint
+        LOG.info("Iceberg committer {}.{} skip committing transaction: " +

Review comment:
       If there are some checkpoint failures. Flink will continue running.
   So actually it will be `commitUpToCheckpoint`, it is better to support this now? Otherwise will never commit after this situation?

##########
File path: flink/src/main/java/org/apache/iceberg/flink/connector/sink/IcebergCommitter.java
##########
@@ -0,0 +1,642 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iceberg.flink.connector.sink;
+
+import com.google.common.base.Joiner;
+import com.google.common.base.Strings;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.stream.Collectors;
+import javax.annotation.Nullable;
+import org.apache.flink.annotation.VisibleForTesting;
+import org.apache.flink.api.common.JobExecutionResult;
+import org.apache.flink.api.common.JobID;
+import org.apache.flink.api.common.state.ListState;
+import org.apache.flink.api.common.state.ListStateDescriptor;
+import org.apache.flink.api.java.ExecutionEnvironment;
+import org.apache.flink.configuration.CheckpointingOptions;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.runtime.state.CheckpointListener;
+import org.apache.flink.runtime.state.FunctionInitializationContext;
+import org.apache.flink.runtime.state.FunctionSnapshotContext;
+import org.apache.flink.streaming.api.checkpoint.CheckpointedFunction;
+import org.apache.flink.streaming.api.environment.CheckpointConfig;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
+import org.apache.flink.util.Preconditions;
+import org.apache.hadoop.hive.conf.HiveConf.ConfVars;
+import org.apache.iceberg.AppendFiles;
+import org.apache.iceberg.DataFile;
+import org.apache.iceberg.ManifestFile;
+import org.apache.iceberg.ManifestWriter;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.Snapshot;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.Transaction;
+import org.apache.iceberg.UpdateProperties;
+import org.apache.iceberg.catalog.Catalog;
+import org.apache.iceberg.catalog.TableIdentifier;
+import org.apache.iceberg.flink.connector.IcebergConnectorConstant;
+import org.apache.iceberg.flink.connector.model.CommitMetadata;
+import org.apache.iceberg.flink.connector.model.CommitMetadataUtil;
+import org.apache.iceberg.flink.connector.model.FlinkManifestFile;
+import org.apache.iceberg.flink.connector.model.FlinkManifestFileUtil;
+import org.apache.iceberg.flink.connector.model.GenericFlinkManifestFile;
+import org.apache.iceberg.flink.connector.model.ManifestFileState;
+import org.apache.iceberg.hadoop.HadoopCatalog;
+import org.apache.iceberg.hive.HiveCatalogs;
+import org.apache.iceberg.io.FileIO;
+import org.apache.iceberg.io.OutputFile;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * This operator commit data files to Iceberg table.
+ * <p>
+ * This operator should always run with parallelism of 1.
+ * Because Iceberg lib perform optimistic concurrency control,
+ * this can help reduce contention and retries
+ * when committing files to Iceberg table.
+ * <p>
+ * Here are some known contentions
+ * 1) Flink jobs running in multiple regions,
+ * since Iceberg metadata service and commit only happens in us-east-1.
+ * 2) auto tuning and auto lift services may update Iceberg table infrequently.
+ */
+@SuppressWarnings("checkstyle:HiddenField")
+public class IcebergCommitter extends RichSinkFunction<FlinkDataFile>
+    implements CheckpointedFunction, CheckpointListener {
+  private static final Logger LOG = LoggerFactory.getLogger(IcebergCommitter.class);
+
+  private static final String COMMIT_REGION_KEY = "flink.commit.region";
+  private static final String COMMIT_MANIFEST_HASHES_KEY = "flink.commit.manifest.hashes";
+  private static final String VTTS_WATERMARK_PROP_KEY_PREFIX = "flink.watermark.";
+
+  private Configuration config;
+  private final String namespace;
+  private final String tableName;
+
+  private final String region;
+  private final boolean vttsWatermarkEnabled;
+  private final String vttsWatermarkPropKey;
+  private final long snapshotRetentionHours;
+  private final boolean commitRestoredManifestFiles;
+  private final String icebergManifestFileDir;
+  private final PartitionSpec spec;
+  private final FileIO io;
+  private final String flinkJobId;
+
+  private transient Table table;
+  private transient List<FlinkDataFile> pendingDataFiles;
+  private transient List<FlinkManifestFile> flinkManifestFiles;
+  private transient ListState<ManifestFileState> manifestFilesState;
+  private transient CommitMetadata metadata;
+  private transient ListState<CommitMetadata> metadataState;
+
+  public IcebergCommitter(Table table, Configuration config) {
+    this.config = config;
+
+    // current Iceberg sink implementation can't work with concurrent checkpoints.
+    // We disable concurrent checkpoints by default as min pause is set to 60s by default.
+    // Add an assertion to fail explicit in case job enables concurrent checkpoints.
+    CheckpointConfig checkpointConfig = StreamExecutionEnvironment.getExecutionEnvironment().getCheckpointConfig();
+    if (checkpointConfig.getMaxConcurrentCheckpoints() > 1) {
+      throw new IllegalArgumentException("Iceberg sink doesn't support concurrent checkpoints");
+    }
+
+    namespace = config.getString(IcebergConnectorConstant.NAMESPACE, "");
+    tableName = config.getString(IcebergConnectorConstant.TABLE, "");
+
+    String region = System.getenv("EC2_REGION");
+    if (Strings.isNullOrEmpty(region)) {
+      region = "us-east-1";
+      LOG.info("Iceberg committer {}.{} default region to us-east-1", namespace, tableName);
+    }
+    this.region = region;
+
+    vttsWatermarkEnabled = !Strings.isNullOrEmpty(
+        config.getString(IcebergConnectorConstant.VTTS_WATERMARK_TIMESTAMP_FIELD, ""));
+    vttsWatermarkPropKey = VTTS_WATERMARK_PROP_KEY_PREFIX + region;
+    snapshotRetentionHours = config.getLong(IcebergConnectorConstant.SNAPSHOT_RETENTION_HOURS,
+        IcebergConnectorConstant.DEFAULT_SNAPSHOT_RETENTION_HOURS);
+    commitRestoredManifestFiles = config.getBoolean(IcebergConnectorConstant.COMMIT_RESTORED_MANIFEST_FILES,
+        IcebergConnectorConstant.DEFAULT_COMMIT_RESTORED_MANIFEST_FILES);
+    icebergManifestFileDir = getIcebergManifestFileDir(config);
+
+    // The only final fields yielded by table inputted
+    spec = table.spec();
+    io = table.io();
+
+    final JobExecutionResult jobExecutionResult
+        = ExecutionEnvironment.getExecutionEnvironment().getLastJobExecutionResult();
+    if (jobExecutionResult != null) {
+      flinkJobId = jobExecutionResult.getJobID().toString();
+      LOG.info("Get Flink job ID from execution environment: {}", flinkJobId);
+    } else {
+      flinkJobId = new JobID().toString();
+      LOG.info("Execution environment doesn't have executed job. Generate a random job ID : {}", flinkJobId);
+    }
+    LOG.info("Iceberg committer {}.{} created with sink config", namespace, tableName);
+    LOG.info("Iceberg committer {}.{} loaded table partition spec: {}", namespace, tableName, spec);
+  }
+
+  @VisibleForTesting
+  List<FlinkDataFile> getPendingDataFiles() {
+    return pendingDataFiles;
+  }
+
+  @VisibleForTesting
+  List<FlinkManifestFile> getFlinkManifestFiles() {
+    return flinkManifestFiles;
+  }
+
+  @VisibleForTesting
+  CommitMetadata getMetadata() {
+    return metadata;
+  }
+
+  private String getIcebergManifestFileDir(Configuration config) {
+    final String checkpointDir = config.getString(
+        CheckpointingOptions.CHECKPOINTS_DIRECTORY, null);
+    if (null == checkpointDir) {
+      throw new IllegalArgumentException("checkpoint dir is null");
+    }
+
+    return String.format("%s/iceberg/manifest/", checkpointDir);
+  }
+
+  @Override
+  public void close() throws Exception {
+    super.close();
+  }
+
+  void init() {
+    // TODO: duplicate logic, to extract
+    org.apache.hadoop.conf.Configuration hadoopConf = new org.apache.hadoop.conf.Configuration();
+    String catalogType = config.getString(IcebergConnectorConstant.CATALOG_TYPE,
+                                          IcebergConnectorConstant.CATALOG_TYPE_DEFAULT);
+    Catalog catalog = null;
+    switch (catalogType.toUpperCase()) {
+      case IcebergConnectorConstant.HIVE_CATALOG:
+        hadoopConf.set(ConfVars.METASTOREURIS.varname, config.getString(ConfVars.METASTOREURIS.varname, ""));
+        catalog = HiveCatalogs.loadCatalog(hadoopConf);
+        break;
+
+      case IcebergConnectorConstant.HADOOP_CATALOG:
+        catalog = new HadoopCatalog(hadoopConf,
+                                    config.getString(IcebergConnectorConstant.HADOOP_CATALOG_WAREHOUSE_LOCATION, ""));
+        break;
+
+      default:
+        throw new UnsupportedOperationException("Unknown catalog type or not set: " + catalogType);
+    }
+
+    this.table = catalog.loadTable(TableIdentifier.parse(namespace + "." + tableName));
+
+    pendingDataFiles = new ArrayList<>();
+    flinkManifestFiles = new ArrayList<>();
+    metadata = CommitMetadata.newBuilder()
+        .setLastCheckpointId(0)
+        .setLastCheckpointTimestamp(0)
+        .setLastCommitTimestamp(System.currentTimeMillis())
+        .build();
+  }
+
+  @Override
+  public void initializeState(FunctionInitializationContext context) throws Exception {
+    init();
+
+    Preconditions.checkState(manifestFilesState == null,
+        "checkpointedFilesState has already been initialized.");
+    Preconditions.checkState(metadataState == null,
+        "metadataState has already been initialized.");
+    manifestFilesState = context.getOperatorStateStore().getListState(new ListStateDescriptor<>(
+        "iceberg-committer-manifest-files-state", ManifestFileState.class));
+    metadataState = context.getOperatorStateStore().getListState(new ListStateDescriptor<>(
+        "iceberg-committer-metadata-state", CommitMetadata.class));
+
+    if (context.isRestored()) {
+      final Iterable<CommitMetadata> restoredMetadata = metadataState.get();
+      if (null != restoredMetadata) {
+        LOG.info("Iceberg committer {}.{} restoring metadata", namespace, tableName);
+        List<CommitMetadata> metadataList = new ArrayList<>();
+        for (CommitMetadata entry : restoredMetadata) {
+          metadataList.add(entry);
+        }
+        Preconditions.checkState(1 == metadataList.size(),
+            "metadata list size should be 1. got " + metadataList.size());
+        metadata = metadataList.get(0);
+        LOG.info("Iceberg committer {}.{} restored metadata: {}",
+            namespace, tableName, CommitMetadataUtil.getInstance().encodeAsJson(metadata));
+      } else {
+        LOG.info("Iceberg committer {}.{} has nothing to restore for metadata", namespace, tableName);
+      }
+
+      Iterable<ManifestFileState> restoredManifestFileStates = manifestFilesState.get();
+      if (null != restoredManifestFileStates) {
+        LOG.info("Iceberg committer {}.{} restoring manifest files",
+            namespace, tableName);
+        for (ManifestFileState manifestFileState : restoredManifestFileStates) {
+          flinkManifestFiles.add(GenericFlinkManifestFile.fromState(manifestFileState));
+        }
+        LOG.info("Iceberg committer {}.{} restored {} manifest files: {}",
+            namespace, tableName, flinkManifestFiles.size(), flinkManifestFiles);
+        final long now = System.currentTimeMillis();
+        if (now - metadata.getLastCheckpointTimestamp() > TimeUnit.HOURS.toMillis(snapshotRetentionHours)) {
+          flinkManifestFiles.clear();
+          LOG.info("Iceberg committer {}.{} cleared restored manifest files as checkpoint timestamp is too old: " +
+                  "checkpointTimestamp = {}, now = {}, snapshotRetentionHours = {}",
+              namespace, tableName, metadata.getLastCheckpointTimestamp(), now, snapshotRetentionHours);
+        } else {
+          flinkManifestFiles = removeCommittedManifests(flinkManifestFiles);
+          if (flinkManifestFiles.isEmpty()) {
+            LOG.info("Iceberg committer {}.{} has zero uncommitted manifest files from restored state",
+                namespace, tableName);
+          } else {
+            if (commitRestoredManifestFiles) {
+              commitRestoredManifestFiles();
+            } else {
+              LOG.info("skip commit of restored manifest files");
+            }
+          }
+        }
+      } else {
+        LOG.info("Iceberg committer {}.{} has nothing to restore for manifest files", namespace, tableName);
+      }
+    }
+  }
+
+  @VisibleForTesting
+  void commitRestoredManifestFiles() throws Exception {
+    LOG.info("Iceberg committer {}.{} committing last uncompleted transaction upon recovery: " +
+            "metadata = {}, flink manifest files ({}) = {}", namespace, tableName,
+        CommitMetadataUtil.getInstance().encodeAsJson(metadata),
+        flinkManifestFiles.size(), flinkManifestFiles);
+    commit();
+    LOG.info("Iceberg committer {}.{} committed last uncompleted transaction upon recovery: " +
+            "metadata = {}, flink manifest files ({}) = {}", namespace, tableName,
+        CommitMetadataUtil.getInstance().encodeAsJson(metadata),
+        flinkManifestFiles.size(), flinkManifestFiles);
+    postCommitSuccess();
+  }
+
+  private List<FlinkManifestFile> removeCommittedManifests(List<FlinkManifestFile> flinkManifestFiles) {
+    int snapshotCount = 0;
+    String result = "succeeded";
+    final long start = System.currentTimeMillis();
+    try {
+      final Set<String> manifestHashes = flinkManifestFiles.stream()
+          .map(f -> f.hash())
+          .collect(Collectors.toSet());
+      final Set<String> committedHashes = new HashSet<>(flinkManifestFiles.size());
+      final Iterable<Snapshot> snapshots = table.snapshots();
+      for (Snapshot snapshot : snapshots) {
+        ++snapshotCount;
+        final Map<String, String> summary = snapshot.summary();
+        final List<String> hashes = FlinkManifestFileUtil.hashesStringToList(summary.get(COMMIT_MANIFEST_HASHES_KEY));
+        for (String hash : hashes) {
+          if (manifestHashes.contains(hash)) {
+            committedHashes.add(hash);
+          }
+        }
+      }
+      final List<FlinkManifestFile> uncommittedManifestFiles = flinkManifestFiles.stream()
+          .filter(f -> !committedHashes.contains(f.hash()))
+          .collect(Collectors.toList());
+      return uncommittedManifestFiles;
+    } catch (Throwable t) {
+      result = "failed";
+      //LOG.error(String.format("Iceberg committer %s.%s failed to check transaction completed", database, tableName),
+      //          t);
+      LOG.error("Iceberg committer {}.{} failed to check transaction completed. Throwable = {}",
+                namespace, tableName, t);
+      throw t;
+    } finally {
+      final long duration = System.currentTimeMillis() - start;
+      LOG.info("Iceberg committer {}.{} {} to check transaction completed" +
+               " after iterating {} snapshots and {} milli-seconds",
+               namespace, tableName, result, snapshotCount, duration);
+    }
+  }
+
+  @Override
+  public void snapshotState(FunctionSnapshotContext context) throws Exception {
+    LOG.info("Iceberg committer {}.{} snapshot state: checkpointId = {}, triggerTime = {}",
+        namespace, tableName, context.getCheckpointId(), context.getCheckpointTimestamp());
+    Preconditions.checkState(null != manifestFilesState,
+        "manifest files state has not been properly initialized.");
+    Preconditions.checkState(null != metadataState,
+        "metadata state has not been properly initialized.");
+
+    // set transaction to null to indicate a start of a new checkpoint/commit/transaction
+    synchronized (this) {

Review comment:
       Flink mail-box mechanism ensure there is only one thread to invoke the methods of operator/function, so don't need to sync?

##########
File path: flink/src/main/java/org/apache/iceberg/flink/connector/sink/IcebergCommitter.java
##########
@@ -0,0 +1,642 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iceberg.flink.connector.sink;
+
+import com.google.common.base.Joiner;
+import com.google.common.base.Strings;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.stream.Collectors;
+import javax.annotation.Nullable;
+import org.apache.flink.annotation.VisibleForTesting;
+import org.apache.flink.api.common.JobExecutionResult;
+import org.apache.flink.api.common.JobID;
+import org.apache.flink.api.common.state.ListState;
+import org.apache.flink.api.common.state.ListStateDescriptor;
+import org.apache.flink.api.java.ExecutionEnvironment;
+import org.apache.flink.configuration.CheckpointingOptions;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.runtime.state.CheckpointListener;
+import org.apache.flink.runtime.state.FunctionInitializationContext;
+import org.apache.flink.runtime.state.FunctionSnapshotContext;
+import org.apache.flink.streaming.api.checkpoint.CheckpointedFunction;
+import org.apache.flink.streaming.api.environment.CheckpointConfig;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
+import org.apache.flink.util.Preconditions;
+import org.apache.hadoop.hive.conf.HiveConf.ConfVars;
+import org.apache.iceberg.AppendFiles;
+import org.apache.iceberg.DataFile;
+import org.apache.iceberg.ManifestFile;
+import org.apache.iceberg.ManifestWriter;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.Snapshot;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.Transaction;
+import org.apache.iceberg.UpdateProperties;
+import org.apache.iceberg.catalog.Catalog;
+import org.apache.iceberg.catalog.TableIdentifier;
+import org.apache.iceberg.flink.connector.IcebergConnectorConstant;
+import org.apache.iceberg.flink.connector.model.CommitMetadata;
+import org.apache.iceberg.flink.connector.model.CommitMetadataUtil;
+import org.apache.iceberg.flink.connector.model.FlinkManifestFile;
+import org.apache.iceberg.flink.connector.model.FlinkManifestFileUtil;
+import org.apache.iceberg.flink.connector.model.GenericFlinkManifestFile;
+import org.apache.iceberg.flink.connector.model.ManifestFileState;
+import org.apache.iceberg.hadoop.HadoopCatalog;
+import org.apache.iceberg.hive.HiveCatalogs;
+import org.apache.iceberg.io.FileIO;
+import org.apache.iceberg.io.OutputFile;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * This operator commit data files to Iceberg table.
+ * <p>
+ * This operator should always run with parallelism of 1.
+ * Because Iceberg lib perform optimistic concurrency control,
+ * this can help reduce contention and retries
+ * when committing files to Iceberg table.
+ * <p>
+ * Here are some known contentions
+ * 1) Flink jobs running in multiple regions,
+ * since Iceberg metadata service and commit only happens in us-east-1.
+ * 2) auto tuning and auto lift services may update Iceberg table infrequently.
+ */
+@SuppressWarnings("checkstyle:HiddenField")
+public class IcebergCommitter extends RichSinkFunction<FlinkDataFile>

Review comment:
       It is better to decouple the commit logic and Flink sink function?
   Maybe in future we can support commit in `OperatorCoordinator`[1]. Now this feature is designing to be exposed to sink function too.
   
   [1]https://github.com/apache/flink/blob/4f6efedd31d0e7705a094725824ddec8940efb3b/flink-runtime/src/main/java/org/apache/flink/runtime/operators/coordination/OperatorCoordinator.java
   

##########
File path: flink/src/main/java/org/apache/iceberg/flink/connector/sink/IcebergSinkAppender.java
##########
@@ -0,0 +1,96 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iceberg.flink.connector.sink;
+
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.streaming.api.datastream.DataStream;
+import org.apache.flink.streaming.api.datastream.DataStreamSink;
+import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.data.Record;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Append Iceberg sink to DataStream
+ *
+ * @param <IN> input data type
+ */
+@SuppressWarnings({"checkstyle:ClassTypeParameterName", "checkstyle:HiddenField"})
+public class IcebergSinkAppender<IN> {
+  private static final Logger LOG = LoggerFactory.getLogger(IcebergSinkAppender.class);
+
+  private final Table table;
+  private final Configuration config;
+  private final String sinkName;
+  private RecordSerializer<IN> serializer;
+  private Integer writerParallelism;
+
+  public IcebergSinkAppender(Table table, Configuration config, String sinkName) {
+    this.table = table;
+    this.config = config;
+    this.sinkName = sinkName;
+  }
+
+  /**
+   * Required.
+   *
+   * @param serializer Serialize input data type to Iceberg {@link Record}
+   */
+  public IcebergSinkAppender<IN> withSerializer(RecordSerializer<IN> serializer) {
+    this.serializer = serializer;
+    return this;
+  }
+
+  /**
+   * Optional. Explicitly set the parallelism for Iceberg writer operator.
+   * Otherwise, default job parallelism is used.
+   */
+  public IcebergSinkAppender<IN> withWriterParallelism(Integer writerParallelism) {
+    this.writerParallelism = writerParallelism;
+    return this;
+  }
+
+  /**
+   * @param dataStream append sink to this DataStream
+   */
+  public DataStreamSink<FlinkDataFile> append(DataStream<IN> dataStream) {
+    // TODO: need to return?
+    IcebergWriter<IN> writer = new IcebergWriter<>(table, serializer, config);
+    IcebergCommitter committer = new IcebergCommitter(table, config);
+
+    final String writerId = sinkName + "-writer";
+    SingleOutputStreamOperator<FlinkDataFile> writerStream = dataStream
+        .transform(writerId, TypeInformation.of(FlinkDataFile.class), writer)  // IcebergWriter as stream operator
+        .uid(writerId);
+    if (null != writerParallelism && writerParallelism > 0) {
+      LOG.info("Set Iceberg writer parallelism to {}", writerParallelism);
+      writerStream.setParallelism(writerParallelism);

Review comment:
       Maybe we can have a strategy to set to input parallelism? (can be chained with the upstream operator)




----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org



---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org

[GitHub] [incubator-iceberg] stevenzwu commented on a change in pull request #856: [WIP] Flink Iceberg sink

[GitBox <gi...@apache.org>]

stevenzwu commented on a change in pull request #856:
URL: https://github.com/apache/incubator-iceberg/pull/856#discussion_r420932982



##########
File path: build.gradle
##########
@@ -442,6 +443,31 @@ project(':iceberg-spark-runtime') {
   }
 }
 
+project(':iceberg-flink') {
+  apply plugin: 'com.commercehub.gradle.plugin.avro'
+
+  dependencies {
+    compile project(':iceberg-api')
+    compile project(':iceberg-core')
+    compile project(':iceberg-data')
+    compile project(path: ':iceberg-data', configuration: 'testArtifacts')  // TODO: to remove, for RandomGenericData
+    compile project(':iceberg-hive')
+    compile project(':iceberg-parquet')
+
+    compileOnly("org.apache.flink:flink-streaming-java_2.11")

Review comment:
       maybe make the Scala version a gradle property. unfortunately, Flink doesn't have a clean Java API. even `flink-streaming-java` has a Scala version suffix in the jar name.
   
   @rdblue what is your take on Scala cross compile? I saw the `iceberg-spark`  module also has hard-coded `2.11` Scala suffix in the spark deps. 
   
   For the context, we need to support both Scala 2.11 and 2.12 internally. We can create a separate jira to discuss that.




----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org



---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org

[GitHub] [incubator-iceberg] stevenzwu commented on pull request #856: [WIP] Flink Iceberg sink

[GitBox <gi...@apache.org>]

stevenzwu commented on pull request #856:
URL: https://github.com/apache/incubator-iceberg/pull/856#issuecomment-624766145


   @waterlx  I would +1 on your proposal of some serializable table interface. I understand @rdblue argument on not making `Table` interface serializable. 
   
   We defined a holder interface that holds the serializable components/fields/properties from the `Table` interface in our internal PoC implementation of Flink Iceberg source. However, I don't like the interface name `TableProperties` as it can easily get confused with the table properties concept in Iceberg.
   ```
   @Value
   public class TableProperties implements Serializable {
   
     private final Schema schema;
     private final FileIO fileIO;
     private final EncryptionManager encryptionManager;
   
     public TableProperties(Table table) {
       this.schema = table.schema();
       this.fileIO = table.io();
       this.encryptionManager = table.encryption();
     }
   }
   ```


----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org



---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org

[GitHub] [incubator-iceberg] waterlx commented on a change in pull request #856: [WIP] Flink Iceberg sink

[GitBox <gi...@apache.org>]

waterlx commented on a change in pull request #856:
URL: https://github.com/apache/incubator-iceberg/pull/856#discussion_r411993357



##########
File path: flink/src/main/java/org/apache/iceberg/flink/connector/sink/FileWriter.java
##########
@@ -0,0 +1,241 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iceberg.flink.connector.sink;
+
+import com.google.common.base.Preconditions;
+import java.io.IOException;
+import java.io.Serializable;
+import java.util.concurrent.TimeUnit;
+import org.apache.avro.Schema;
+import org.apache.avro.generic.IndexedRecord;
+import org.apache.flink.streaming.runtime.tasks.ProcessingTimeService;
+import org.apache.hadoop.fs.Path;
+import org.apache.iceberg.DataFile;
+import org.apache.iceberg.DataFiles;
+import org.apache.iceberg.FileFormat;
+import org.apache.iceberg.Metrics;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.exceptions.RuntimeIOException;
+import org.apache.iceberg.hadoop.HadoopInputFile;
+import org.apache.iceberg.io.FileAppender;
+import org.apache.iceberg.io.InputFile;
+
+public final class FileWriter implements Serializable {
+  private static final long serialVersionUID = 1L;
+
+  private final FileFormat format;
+  private final Path path;
+  private final ProcessingTimeService timeService;
+  private final long creationTime;
+  private final Partitioner<IndexedRecord> partitioner;
+  private final FileAppender<IndexedRecord> appender;
+  private final org.apache.hadoop.conf.Configuration hadoopConfig;
+  private final PartitionSpec spec;
+  private final IcebergWriterTaskMetrics icebergWriterTaskMetrics;
+  private final IcebergWriterSubtaskMetrics icebergWriterSubtaskMetrics;
+  private final WatermarkTimeExtractor watermarkTimeExtractor;
+
+  private long lastWrittenToTime;
+  private long lowWatermark = Long.MAX_VALUE;
+  private long highWatermark = Long.MIN_VALUE;
+
+  @SuppressWarnings("checkstyle:HiddenField")
+  public static class Builder<T> {
+    private FileFormat format;
+    private Path path;
+    private ProcessingTimeService timeService;
+    private Partitioner<T> partitioner;
+    private FileAppender<T> appender;
+    private org.apache.hadoop.conf.Configuration hadoopConfig;
+    private PartitionSpec spec;
+    private IcebergWriterTaskMetrics icebergWriterTaskMetrics;
+    private IcebergWriterSubtaskMetrics icebergWriterSubtaskMetrics;
+    private Schema avroSchema;
+    private String vttsTimestampField;
+    private TimeUnit vttsTimestampUnit;
+
+    private Builder() {
+    }
+
+    public Builder withFileFormat(final FileFormat format) {
+      this.format = format;
+      return this;
+    }
+
+    public Builder withPath(final Path path) {
+      this.path = path;
+      return this;
+    }
+
+    public Builder withProcessingTimeService(final ProcessingTimeService timeService) {
+      this.timeService = timeService;
+      return this;
+    }
+
+    public Builder withPartitioner(final Partitioner<T> partitioner) {
+      this.partitioner = partitioner;
+      return this;
+    }
+
+    public Builder withAppender(final FileAppender appender) {
+      this.appender = appender;
+      return this;
+    }
+
+    public Builder withHadooopConfig(final org.apache.hadoop.conf.Configuration hadoopConfig) {
+      this.hadoopConfig = hadoopConfig;
+      return this;
+    }
+
+    public Builder withSpec(final PartitionSpec spec) {
+      this.spec = spec;
+      return this;
+    }
+
+    public Builder withMetrics(final IcebergWriterTaskMetrics icebergWriterTaskMetrics) {
+      this.icebergWriterTaskMetrics = icebergWriterTaskMetrics;
+      return this;
+    }
+
+    public Builder withSubtaskMetrics(final IcebergWriterSubtaskMetrics icebergWriterSubtaskMetrics) {
+      this.icebergWriterSubtaskMetrics = icebergWriterSubtaskMetrics;
+      return this;
+    }
+
+    public Builder withSchema(final Schema avroSchema) {
+      this.avroSchema = avroSchema;
+      return this;
+    }
+
+    public Builder withVttsTimestampField(final String vttsTimestampField) {
+      this.vttsTimestampField = vttsTimestampField;
+      return this;
+    }
+
+    public Builder withVttsTimestampUnit(final TimeUnit vttsTimestampUnit) {
+      this.vttsTimestampUnit = vttsTimestampUnit;
+      return this;
+    }
+
+    public FileWriter build() {
+      Preconditions.checkArgument(this.format != null, "File format is required");
+      Preconditions.checkArgument(this.path != null, "File path is required");
+      Preconditions.checkArgument(this.timeService != null, "ProcessingTimeService is required");
+      Preconditions.checkArgument(this.partitioner != null, "Partitioner is required");
+      Preconditions.checkArgument(this.appender != null, "File appender is required");
+      Preconditions.checkArgument(this.hadoopConfig != null, "Hadoop config is required");
+      Preconditions.checkArgument(this.spec != null, "Partition spec is required");
+      Preconditions.checkArgument(this.icebergWriterTaskMetrics != null, "icebergWriterTaskMetrics is required");
+      Preconditions.checkArgument(this.avroSchema != null, "avroSchema is required");
+      Preconditions.checkArgument(this.vttsTimestampField != null, "vttsTimestampField is required");
+      Preconditions.checkArgument(this.vttsTimestampUnit != null, "vttsTimestampUnit is required");
+      return new FileWriter(this);
+    }
+  }
+
+  private FileWriter(Builder builder) {
+    format = builder.format;
+    path = builder.path;
+    timeService = builder.timeService;
+    creationTime = timeService.getCurrentProcessingTime();
+    partitioner = builder.partitioner;
+    appender = builder.appender;
+    hadoopConfig = builder.hadoopConfig;
+    spec = builder.spec;
+    icebergWriterTaskMetrics = builder.icebergWriterTaskMetrics;
+    icebergWriterSubtaskMetrics = builder.icebergWriterSubtaskMetrics;
+    watermarkTimeExtractor = new WatermarkTimeExtractor(
+        builder.avroSchema, builder.vttsTimestampField, builder.vttsTimestampUnit);
+  }
+
+  public static Builder builder() {
+    return new Builder();
+  }
+
+  public long write(IndexedRecord record) {

Review comment:
       Addressed. Use Record (within Iceberg) as the input of write().
   Will put the alternative solution as you sugguested into future enhancement.




----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org



---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org

[GitHub] [incubator-iceberg] waterlx commented on issue #856: [WIP] Flink Iceberg sink

[GitBox <gi...@apache.org>]

waterlx commented on issue #856:
URL: https://github.com/apache/incubator-iceberg/pull/856#issuecomment-617074928


   @rdblue Thanks for sharing your thoughts on the reasons why Table/BaseTable is not serializable. Totally agree. But I am currently in the dilemma where there might be a need to call high-level operations in Flink tasks,  like table.newTransaction() when trying to cimmit DataFiles accumulated from the streaming inputs. Currently code limits the parallelism to 1 so that the commit won't be performed in parallel.  
   
   For now, it is not a blocker because I could pass the namespace and table name by config and call loadTable() of Catalog to build the table when there is a need. But the implementation is not that good as the table informations(like namespace, table name, it is HiveCatalog or HadoopCatalog) passes eveywhere, while some of them are not needed.
   
   I am also thinking about passing the path as a string (db.table for Hive Catalog while full qualified path for HadoopTables) instead of passing table instance, but the purpose is also to re-build the table instance so as to call some high-level operations.


----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org



---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org

[GitHub] [incubator-iceberg] waterlx commented on pull request #856: [WIP] Flink Iceberg sink

[GitBox <gi...@apache.org>]

waterlx commented on pull request #856:
URL: https://github.com/apache/incubator-iceberg/pull/856#issuecomment-624800409


   @stevenzwu @JingsongLi Really appreciate your effort to review this PR.  I am working on the comments you provided.
   Meanwhile, our team are focusing on verifying/testing the sink and I am in process of uploading the bug fixes/enhancements into this PR one after another.
   
   **Progress update:**
   The comments by @rdblue @openinx and @jerryshao regarding the initial version of this PR have almost been all addressed. 
   Begin to work on the following items:
   - Addressing the comments added recnetly by @stevenzwu and @JingsongLi 
   - Testing and bug fixing
   - Adding UT
   - Polishing the code and working on check-style issues
   
   The PR is being updated and always a workable version (against the master branch) with functions verified. So if you are interested, feel free to download the PR and get it compiled with master branch. [This design doc](https://docs.google.com/document/d/19M-sP6FlTVm7BV7MM4Om1n_MVo1xCy7GyDl_9ZAjVNQ/edit) includes the sample Java code to give it a try  8-)
   
   Drop me a mail or add any comments if it does not work.


----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org



---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org

[GitHub] [incubator-iceberg] waterlx edited a comment on pull request #856: [WIP] Flink Iceberg sink

[GitBox <gi...@apache.org>]

waterlx edited a comment on pull request #856:
URL: https://github.com/apache/incubator-iceberg/pull/856#issuecomment-624800409


   @stevenzwu @JingsongLi Really appreciate your effort to review this PR.  I am working on the comments you provided.
   Meanwhile, our team are focusing on verifying/testing the sink and I am in process of uploading the bug fixes/enhancements into this PR one after another.
   
   **Progress update:**
   The comments by @rdblue @openinx and @jerryshao regarding the initial version of this PR have almost been all addressed. 
   Begin to work on the following items:
   - Addressing the comments added recnetly by @stevenzwu and @JingsongLi 
   - Testing and bug fixing
   - Adding UT
   - Polishing the code and working on check-style issues
   
   The PR is being updated and always a workable version (against the master branch) with functions verified. So if you are interested, feel free to download the PR and get it compiled with master branch. [This design doc](https://docs.google.com/document/d/19M-sP6FlTVm7BV7MM4Om1n_MVo1xCy7GyDl_9ZAjVNQ/edit) includes the sample Java code to give it a try  8-)
   
   Drop me a mail or add any comments if it does not work.
   
   @jerryshao @chenjunjiedada @openinx @aokolnychyi @ @bowenli86 @stevenzwu @JingsongLi @rdblue FYI


----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org



---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org

[GitHub] [incubator-iceberg] waterlx commented on issue #856: [WIP] Flink Iceberg sink

[GitBox <gi...@apache.org>]

waterlx commented on issue #856:
URL: https://github.com/apache/incubator-iceberg/pull/856#issuecomment-617082172


   Some progress of Flink Iceberg sink in case you are interested:
   1. S3 and AWS SDK are removed, while FileIO is used instead. So https://github.com/apache/incubator-iceberg/issues/872 is closed.
   2. The usage of Avro and IndexedRecord is almost removed. The input of sink accepted is Record/GenericRecord within Iceberg; For other inputs, RecordSerializer is supposed to be implemented accordingly.
   3. Working on clearing up VTTS and watermark related code. It has something to do with Avro and is the last dependent on Avro.
   
   PR is updated with basic functions verified, but still not for detailed review yet. Hopefully I will get  a clean version by the end of this week.
   @jerryshao @chenjunjiedada @openinx @aokolnychyi @bowenli86 @stevenzwu @rdblue FYI


----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org



---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org

[GitHub] [incubator-iceberg] stevenzwu edited a comment on pull request #856: [WIP] Flink Iceberg sink

[GitBox <gi...@apache.org>]

stevenzwu edited a comment on pull request #856:
URL: https://github.com/apache/incubator-iceberg/pull/856#issuecomment-624766145


   @waterlx  I would +1 on your proposal of some serializable table interface. I understand @rdblue argument on not making `Table` interface serializable. 
   
   We defined a holder interface that holds the serializable components/fields/properties from the `Table` interface in our internal PoC implementation of Flink Iceberg source. However, I don't like the interface name `TableProperties` as it can easily get confused with the table properties concept in Iceberg.
   ```
   import lombok.Value;
   
   @Value
   public class TableProperties implements Serializable {
   
     private final Schema schema;
     private final FileIO fileIO;
     private final EncryptionManager encryptionManager;
   
     public TableProperties(Table table) {
       this.schema = table.schema();
       this.fileIO = table.io();
       this.encryptionManager = table.encryption();
     }
   }
   ```


----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org



---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org

[GitHub] [incubator-iceberg] stevenzwu commented on a change in pull request #856: [WIP] Flink Iceberg sink

[GitBox <gi...@apache.org>]

stevenzwu commented on a change in pull request #856:
URL: https://github.com/apache/incubator-iceberg/pull/856#discussion_r420932982



##########
File path: build.gradle
##########
@@ -442,6 +443,31 @@ project(':iceberg-spark-runtime') {
   }
 }
 
+project(':iceberg-flink') {
+  apply plugin: 'com.commercehub.gradle.plugin.avro'
+
+  dependencies {
+    compile project(':iceberg-api')
+    compile project(':iceberg-core')
+    compile project(':iceberg-data')
+    compile project(path: ':iceberg-data', configuration: 'testArtifacts')  // TODO: to remove, for RandomGenericData
+    compile project(':iceberg-hive')
+    compile project(':iceberg-parquet')
+
+    compileOnly("org.apache.flink:flink-streaming-java_2.11")

Review comment:
       maybe make the Scala version a gradle property. unfortunately, Flink doesn't have a clean Java API. even `flink-streaming-java` has a Scala version suffix in the jar name.
   
   @rdblue what is your take on Scala cross compile? I saw the `iceberg-spark`  module also has hard-coded `2.11` Scala suffix in the spark deps. 
   
   For the context, we need to support both Scala 2.11 and 2.12 internally.




----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org



---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org

[GitHub] [incubator-iceberg] waterlx edited a comment on pull request #856: [WIP] Flink Iceberg sink

[GitBox <gi...@apache.org>]

waterlx edited a comment on pull request #856:
URL: https://github.com/apache/incubator-iceberg/pull/856#issuecomment-624800409


   @stevenzwu @JingsongLi Really appreciate your effort to review this PR.  I am working on the comments you provided.
   Meanwhile, our team are focusing on verifying/testing the sink and I am in process of uploading the bug fixes/enhancements into this PR one after another.
   
   **Progress update:**
   The comments by @rdblue @openinx and @jerryshao regarding the initial version of this PR have been almost all addressed. 
   Begin to work on the following items:
   - Addressing the comments added recently by @stevenzwu and @JingsongLi 
   - Testing and bug fixing
   - Adding UT
   - Polishing the code and working on check-style issues
   
   The PR is being updated and always a workable version (against the master branch) with functions verified. So if you are interested, feel free to download the PR and get it compiled with master branch. [This design doc](https://docs.google.com/document/d/19M-sP6FlTVm7BV7MM4Om1n_MVo1xCy7GyDl_9ZAjVNQ/edit) includes the sample Java code to give it a try  8-)
   
   Drop me a mail or add any comments if it does not work.
   
   @jerryshao @chenjunjiedada @openinx @aokolnychyi @ @bowenli86 @stevenzwu @JingsongLi @rdblue FYI


----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org



---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org

[GitHub] [incubator-iceberg] waterlx commented on a change in pull request #856: [WIP] Flink Iceberg sink

[GitBox <gi...@apache.org>]

waterlx commented on a change in pull request #856:
URL: https://github.com/apache/incubator-iceberg/pull/856#discussion_r411987951



##########
File path: flink/src/main/java/org/apache/iceberg/flink/connector/sink/AvroSerializer.java
##########
@@ -0,0 +1,32 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iceberg.flink.connector.sink;
+
+import java.io.Serializable;
+import org.apache.avro.Schema;
+import org.apache.avro.generic.IndexedRecord;
+
+/**
+ * Serialize input data type to Avro IndexedRecord
+ */
+@FunctionalInterface
+public interface AvroSerializer<IN> extends Serializable {

Review comment:
       Addressed. Avro IndexedRecord is removed and [Record](https://github.com/apache/incubator-iceberg/blob/master/data/src/main/java/org/apache/iceberg/data/Record.java) is used instead.
   Regarding the translation from/to Flink format, will open a new issue to address that. 




----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org



---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org

[GitHub] [incubator-iceberg] waterlx commented on a change in pull request #856: [WIP] Flink Iceberg sink

[GitBox <gi...@apache.org>]

waterlx commented on a change in pull request #856:
URL: https://github.com/apache/incubator-iceberg/pull/856#discussion_r415622644



##########
File path: flink/src/main/java/org/apache/iceberg/flink/connector/sink/IcebergCommitter.java
##########
@@ -0,0 +1,693 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iceberg.flink.connector.sink;
+
+import com.google.common.base.Joiner;
+import com.google.common.base.Strings;
+import com.netflix.spectator.api.DefaultRegistry;
+import com.netflix.spectator.api.Registry;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.stream.Collectors;
+import javax.annotation.Nullable;
+import org.apache.flink.annotation.VisibleForTesting;
+import org.apache.flink.api.common.JobExecutionResult;
+import org.apache.flink.api.common.JobID;
+import org.apache.flink.api.common.state.ListState;
+import org.apache.flink.api.common.state.ListStateDescriptor;
+import org.apache.flink.api.java.ExecutionEnvironment;
+import org.apache.flink.configuration.CheckpointingOptions;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.runtime.state.CheckpointListener;
+import org.apache.flink.runtime.state.FunctionInitializationContext;
+import org.apache.flink.runtime.state.FunctionSnapshotContext;
+import org.apache.flink.streaming.api.checkpoint.CheckpointedFunction;
+import org.apache.flink.streaming.api.environment.CheckpointConfig;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
+import org.apache.flink.util.Preconditions;
+import org.apache.hadoop.hive.conf.HiveConf;
+import org.apache.iceberg.AppendFiles;
+import org.apache.iceberg.DataFile;
+import org.apache.iceberg.ManifestFile;
+import org.apache.iceberg.ManifestWriter;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.Snapshot;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.Transaction;
+import org.apache.iceberg.UpdateProperties;
+import org.apache.iceberg.catalog.Catalog;
+import org.apache.iceberg.catalog.TableIdentifier;
+import org.apache.iceberg.flink.connector.IcebergConnectorConstant;
+import org.apache.iceberg.flink.connector.model.CommitMetadata;
+import org.apache.iceberg.flink.connector.model.CommitMetadataUtil;
+import org.apache.iceberg.flink.connector.model.FlinkManifestFile;
+import org.apache.iceberg.flink.connector.model.FlinkManifestFileUtil;
+import org.apache.iceberg.flink.connector.model.GenericFlinkManifestFile;
+import org.apache.iceberg.flink.connector.model.ManifestFileState;
+import org.apache.iceberg.hadoop.HadoopFileIO;
+import org.apache.iceberg.hive.HiveCatalogs;
+import org.apache.iceberg.io.OutputFile;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * This operator commit data files to Iceberg table.
+ * <p>
+ * This operator should always run with parallelism of 1.
+ * Because Iceberg lib perform optimistic concurrency control,
+ * this can help reduce contention and retries
+ * when committing files to Iceberg table.
+ * <p>
+ * Here are some known contentions
+ * 1) Flink jobs running in multiple regions,
+ * since Iceberg metadata service and commit only happens in us-east-1.
+ * 2) auto tuning and auto lift services may update Iceberg table infrequently.
+ */
+@SuppressWarnings("checkstyle:HiddenField")
+public class IcebergCommitter extends RichSinkFunction<FlinkDataFile>
+    implements CheckpointedFunction, CheckpointListener {
+  private static final Logger LOG = LoggerFactory.getLogger(IcebergCommitter.class);
+
+  private static final String COMMIT_REGION_KEY = "flink.commit.region";
+  private static final String COMMIT_MANIFEST_HASHES_KEY = "flink.commit.manifest.hashes";
+  private static final String VTTS_WATERMARK_PROP_KEY_PREFIX = "flink.watermark.";
+
+  private final String metacatHost;
+  //private final String jobName;
+  //private final String catalog;
+  private final String database;
+  private final String tableName;
+  private final String region;
+  private final boolean vttsWatermarkEnabled;
+  private final String vttsWatermarkPropKey;
+  private final long snapshotRetentionHours;
+  private final boolean commitRestoredManifestFiles;
+  private final String icebergManifestFileDir;
+  private final PartitionSpec spec;
+  private final HadoopFileIO hadoopFileIO;
+  private final String flinkJobId;
+
+  private transient Registry registry;
+  private transient Table table;
+  private transient List<FlinkDataFile> pendingDataFiles;
+  private transient List<FlinkManifestFile> flinkManifestFiles;
+  private transient ListState<ManifestFileState> manifestFilesState;
+  private transient CommitMetadata metadata;
+  private transient ListState<CommitMetadata> metadataState;
+  private transient IcebergCommitterMetrics metrics;
+
+  public IcebergCommitter(Configuration config) {
+    // current Iceberg sink implementation can't work with concurrent checkpoints.
+    // We disable concurrent checkpoints by default as min pause is set to 60s by default.
+    // Add an assertion to fail explicit in case job enables concurrent checkpoints.
+    CheckpointConfig checkpointConfig = StreamExecutionEnvironment.getExecutionEnvironment().getCheckpointConfig();
+    if (checkpointConfig.getMaxConcurrentCheckpoints() > 1) {
+      throw new IllegalArgumentException("Iceberg sink doesn't support concurrent checkpoints");
+    }
+
+    metacatHost = config.getString(IcebergConnectorConstant.METACAT_HOST,

Review comment:
       Addressed




----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org



---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org