apache · danny0405 · Sep 9, 2022 · Sep 8, 2022
diff --git a/hudi-client/hudi-flink-client/pom.xml b/hudi-client/hudi-flink-client/pom.xml
diff --git a/...ient/hudi-flink-client/src/main/java/org/apache/hudi/client/FlinkTaskContextSupplier.java b/...ient/hudi-flink-client/src/main/java/org/apache/hudi/client/FlinkTaskContextSupplier.java
@@ -61,5 +61,5 @@ public Option<String> getProperty(EngineProperty prop) {
     // no operation for now
     return Option.empty();
   }
-  
+
 }
diff --git a/...client/hudi-flink-client/src/main/java/org/apache/hudi/client/HoodieFlinkWriteClient.java b/...client/hudi-flink-client/src/main/java/org/apache/hudi/client/HoodieFlinkWriteClient.java
@@ -81,6 +81,14 @@
 import java.util.Map;
 import java.util.stream.Collectors;
 
+/**
+ * Flink hoodie write client.
+ *
+ * <p>The client is used both on driver (for starting/committing transactions)
+ * and executor (for writing dataset).
+ *
+ * @param <T> type of the payload
+ */
 @SuppressWarnings("checkstyle:LineLength")
 public class HoodieFlinkWriteClient<T extends HoodieRecordPayload> extends
     BaseHoodieWriteClient<T, List<HoodieRecord<T>>, List<HoodieKey>, List<WriteStatus>> {

diff --git a/...di-flink-client/src/main/java/org/apache/hudi/client/common/HoodieFlinkEngineContext.java b/...di-flink-client/src/main/java/org/apache/hudi/client/common/HoodieFlinkEngineContext.java
@@ -33,6 +33,9 @@
 import org.apache.hudi.common.function.SerializablePairFlatMapFunction;
 import org.apache.hudi.common.function.SerializablePairFunction;
 import org.apache.hudi.common.util.Option;
+import org.apache.hudi.common.util.collection.ImmutablePair;
+import org.apache.hudi.common.util.collection.Pair;
+import org.apache.hudi.util.FlinkClientUtil;
 
 import org.apache.flink.api.common.functions.RuntimeContext;
 
@@ -45,10 +48,6 @@
 import java.util.stream.Collectors;
 import java.util.stream.Stream;
 
-import org.apache.hudi.common.util.collection.ImmutablePair;
-import org.apache.hudi.common.util.collection.Pair;
-import org.apache.hudi.util.FlinkClientUtil;
-
 import static org.apache.hudi.common.function.FunctionWrapper.throwingFlatMapToPairWrapper;
 import static org.apache.hudi.common.function.FunctionWrapper.throwingFlatMapWrapper;
 import static org.apache.hudi.common.function.FunctionWrapper.throwingForeachWrapper;

diff --git a/...nt/hudi-flink-client/src/main/java/org/apache/hudi/execution/FlinkLazyInsertIterable.java b/...nt/hudi-flink-client/src/main/java/org/apache/hudi/execution/FlinkLazyInsertIterable.java
@@ -36,6 +36,11 @@
 import java.util.Iterator;
 import java.util.List;
 
+/**
+ * Flink lazy iterable that supports explicit write handler.
+ *
+ * @param <T> type of the payload
+ */
 public class FlinkLazyInsertIterable<T extends HoodieRecordPayload> extends HoodieLazyInsertIterable<T> {
 
   public FlinkLazyInsertIterable(Iterator<HoodieRecord<T>> recordItr,

diff --git a/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/io/MiniBatchHandle.java b/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/io/MiniBatchHandle.java
@@ -30,7 +30,8 @@ public interface MiniBatchHandle {
    * come from one checkpoint interval. The file handle may roll over to new name
    * if the name conflicts, give a chance to clean the intermediate file.
    */
-  default void finalizeWrite() {}
+  default void finalizeWrite() {
+  }
 
   /**
    * Close the file handle gracefully, if any error happens during the file handle close,

diff --git a/...flink-client/src/main/java/org/apache/hudi/io/storage/row/HoodieRowDataParquetWriter.java b/...flink-client/src/main/java/org/apache/hudi/io/storage/row/HoodieRowDataParquetWriter.java
@@ -20,10 +20,10 @@
 
 import org.apache.hudi.common.fs.FSUtils;
 import org.apache.hudi.common.fs.HoodieWrapperFileSystem;
+import org.apache.hudi.io.storage.HoodieParquetConfig;
 
 import org.apache.flink.table.data.RowData;
 import org.apache.hadoop.fs.Path;
-import org.apache.hudi.io.storage.HoodieParquetConfig;
 import org.apache.parquet.hadoop.ParquetFileWriter;
 import org.apache.parquet.hadoop.ParquetWriter;
 

diff --git a/...k-client/src/main/java/org/apache/hudi/io/storage/row/parquet/ParquetSchemaConverter.java b/...k-client/src/main/java/org/apache/hudi/io/storage/row/parquet/ParquetSchemaConverter.java
@@ -31,7 +31,6 @@
 import org.apache.flink.table.types.logical.LogicalType;
 import org.apache.flink.table.types.logical.MapType;
 import org.apache.flink.table.types.logical.RowType;
-
 import org.apache.flink.table.types.logical.TimestampType;
 import org.apache.parquet.schema.GroupType;
 import org.apache.parquet.schema.LogicalTypeAnnotation;
@@ -76,7 +75,7 @@ public static TypeInformation<?> fromParquetType(MessageType type) {
    * Converts Flink Internal Type to Parquet schema.
    *
    * @param typeInformation Flink type information
-   * @param legacyMode is standard LIST and MAP schema or back-compatible schema
+   * @param legacyMode      is standard LIST and MAP schema or back-compatible schema
    * @return Parquet schema
    */
   public static MessageType toParquetType(
@@ -569,7 +568,7 @@ private static Type convertToParquetType(
         int scale = ((DecimalType) type).getScale();
         int numBytes = computeMinBytesForDecimalPrecision(precision);
         return Types.primitive(
-            PrimitiveType.PrimitiveTypeName.FIXED_LEN_BYTE_ARRAY, repetition)
+                PrimitiveType.PrimitiveTypeName.FIXED_LEN_BYTE_ARRAY, repetition)
             .as(LogicalTypeAnnotation.decimalType(scale, precision))
             .length(numBytes)
             .named(name);

diff --git a/...k-client/src/main/java/org/apache/hudi/metadata/FlinkHoodieBackedTableMetadataWriter.java b/...k-client/src/main/java/org/apache/hudi/metadata/FlinkHoodieBackedTableMetadataWriter.java
@@ -42,6 +42,9 @@
 import java.util.List;
 import java.util.Map;
 
+/**
+ * Flink hoodie backed table metadata writer.
+ */
 public class FlinkHoodieBackedTableMetadataWriter extends HoodieBackedTableMetadataWriter {
 
   private static final Logger LOG = LogManager.getLogger(FlinkHoodieBackedTableMetadataWriter.class);

diff --git a/...lient/hudi-flink-client/src/main/java/org/apache/hudi/table/ExplicitWriteHandleTable.java b/...lient/hudi-flink-client/src/main/java/org/apache/hudi/table/ExplicitWriteHandleTable.java
@@ -79,7 +79,7 @@ HoodieWriteMetadata<List<WriteStatus>> insert(
    * @param context     HoodieEngineContext
    * @param writeHandle The write handle
    * @param instantTime Instant Time for the action
-   * @param keys   {@link List} of {@link HoodieKey}s to be deleted
+   * @param keys        {@link List} of {@link HoodieKey}s to be deleted
    * @return HoodieWriteMetadata
    */
   HoodieWriteMetadata<List<WriteStatus>> delete(
@@ -96,9 +96,9 @@ HoodieWriteMetadata<List<WriteStatus>> delete(
    * <p>Specifies the write handle explicitly in order to have fine grained control with
    * the underneath file.
    *
-   * @param context    HoodieEngineContext
-   * @param instantTime Instant Time for the action
-   * @param preppedRecords  hoodieRecords to upsert
+   * @param context        HoodieEngineContext
+   * @param instantTime    Instant Time for the action
+   * @param preppedRecords hoodieRecords to upsert
    * @return HoodieWriteMetadata
    */
   HoodieWriteMetadata<List<WriteStatus>> upsertPrepped(
@@ -115,9 +115,9 @@ HoodieWriteMetadata<List<WriteStatus>> upsertPrepped(
    * <p>Specifies the write handle explicitly in order to have fine grained control with
    * the underneath file.
    *
-   * @param context    HoodieEngineContext
-   * @param instantTime Instant Time for the action
-   * @param preppedRecords  hoodieRecords to upsert
+   * @param context        HoodieEngineContext
+   * @param instantTime    Instant Time for the action
+   * @param preppedRecords hoodieRecords to upsert
    * @return HoodieWriteMetadata
    */
   HoodieWriteMetadata<List<WriteStatus>> insertPrepped(
@@ -130,10 +130,10 @@ HoodieWriteMetadata<List<WriteStatus>> insertPrepped(
    * Replaces all the existing records and inserts the specified new records into Hoodie table at the supplied instantTime,
    * for the partition paths contained in input records.
    *
-   * @param context HoodieEngineContext
+   * @param context     HoodieEngineContext
    * @param writeHandle The write handle
    * @param instantTime Instant time for the replace action
-   * @param records input records
+   * @param records     input records
    * @return HoodieWriteMetadata
    */
   HoodieWriteMetadata<List<WriteStatus>> insertOverwrite(
@@ -146,10 +146,10 @@ HoodieWriteMetadata<List<WriteStatus>> insertOverwrite(
    * Deletes all the existing records of the Hoodie table and inserts the specified new records into Hoodie table at the supplied instantTime,
    * for the partition paths contained in input records.
    *
-   * @param context HoodieEngineContext
+   * @param context     HoodieEngineContext
    * @param writeHandle The write handle
    * @param instantTime Instant time for the replace action
-   * @param records input records
+   * @param records     input records
    * @return HoodieWriteMetadata
    */
   HoodieWriteMetadata<List<WriteStatus>> insertOverwriteTable(

diff --git a/...nt/hudi-flink-client/src/main/java/org/apache/hudi/table/HoodieFlinkCopyOnWriteTable.java b/...nt/hudi-flink-client/src/main/java/org/apache/hudi/table/HoodieFlinkCopyOnWriteTable.java
@@ -66,6 +66,7 @@
 import org.apache.hudi.table.action.commit.FlinkUpsertPreppedCommitActionExecutor;
 import org.apache.hudi.table.action.rollback.BaseRollbackPlanActionExecutor;
 import org.apache.hudi.table.action.rollback.CopyOnWriteRollbackActionExecutor;
+
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -147,7 +148,7 @@ public HoodieWriteMetadata<List<WriteStatus>> insert(
    * @param context     HoodieEngineContext
    * @param writeHandle The write handle
    * @param instantTime Instant Time for the action
-   * @param keys   {@link List} of {@link HoodieKey}s to be deleted
+   * @param keys        {@link List} of {@link HoodieKey}s to be deleted
    * @return HoodieWriteMetadata
    */
   public HoodieWriteMetadata<List<WriteStatus>> delete(
@@ -166,9 +167,9 @@ public HoodieWriteMetadata<List<WriteStatus>> delete(
    * <p>Specifies the write handle explicitly in order to have fine grained control with
    * the underneath file.
    *
-   * @param context    HoodieEngineContext
-   * @param instantTime Instant Time for the action
-   * @param preppedRecords  hoodieRecords to upsert
+   * @param context        HoodieEngineContext
+   * @param instantTime    Instant Time for the action
+   * @param preppedRecords hoodieRecords to upsert
    * @return HoodieWriteMetadata
    */
   public HoodieWriteMetadata<List<WriteStatus>> upsertPrepped(
@@ -187,9 +188,9 @@ public HoodieWriteMetadata<List<WriteStatus>> upsertPrepped(
    * <p>Specifies the write handle explicitly in order to have fine grained control with
    * the underneath file.
    *
-   * @param context    HoodieEngineContext
-   * @param instantTime Instant Time for the action
-   * @param preppedRecords  hoodieRecords to upsert
+   * @param context        HoodieEngineContext
+   * @param instantTime    Instant Time for the action
+   * @param preppedRecords hoodieRecords to upsert
    * @return HoodieWriteMetadata
    */
   public HoodieWriteMetadata<List<WriteStatus>> insertPrepped(
@@ -287,7 +288,7 @@ public HoodieWriteMetadata<List<WriteStatus>> compact(
 
   @Override
   public Option<HoodieClusteringPlan> scheduleClustering(final HoodieEngineContext context, final String instantTime, final Option<Map<String, String>> extraMetadata) {
-    return new ClusteringPlanActionExecutor<>(context, config,this, instantTime, extraMetadata).execute();
+    return new ClusteringPlanActionExecutor<>(context, config, this, instantTime, extraMetadata).execute();
   }
 
   @Override
@@ -306,8 +307,8 @@ public void rollbackBootstrap(HoodieEngineContext context, String instantTime) {
   }
 
   /**
-   * @param context HoodieEngineContext
-   * @param instantTime Instant Time for scheduling cleaning
+   * @param context       HoodieEngineContext
+   * @param instantTime   Instant Time for scheduling cleaning
    * @param extraMetadata additional metadata to write into plan
    * @return
    */
@@ -371,7 +372,7 @@ public Iterator<List<WriteStatus>> handleUpdate(
     return handleUpdateInternal(upsertHandle, instantTime, fileId);
   }
 
-  protected Iterator<List<WriteStatus>> handleUpdateInternal(HoodieMergeHandle<?,?,?,?> upsertHandle, String instantTime,
+  protected Iterator<List<WriteStatus>> handleUpdateInternal(HoodieMergeHandle<?, ?, ?, ?> upsertHandle, String instantTime,
                                                              String fileId) throws IOException {
     if (upsertHandle.getOldFilePath() == null) {
       throw new HoodieUpsertException(

diff --git a/...nt/hudi-flink-client/src/main/java/org/apache/hudi/table/HoodieFlinkMergeOnReadTable.java b/...nt/hudi-flink-client/src/main/java/org/apache/hudi/table/HoodieFlinkMergeOnReadTable.java
@@ -45,6 +45,9 @@
 import java.util.List;
 import java.util.Map;
 
+/**
+ * Flink MERGE_ON_READ table.
+ */
 public class HoodieFlinkMergeOnReadTable<T extends HoodieRecordPayload>
     extends HoodieFlinkCopyOnWriteTable<T> {
 

diff --git a/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/table/HoodieFlinkTable.java b/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/table/HoodieFlinkTable.java
@@ -40,6 +40,9 @@
 
 import java.util.List;
 
+/**
+ * Impl of a flink hoodie table.
+ */
 public abstract class HoodieFlinkTable<T extends HoodieRecordPayload>
     extends HoodieTable<T, List<HoodieRecord<T>>, List<HoodieKey>, List<WriteStatus>>
     implements ExplicitWriteHandleTable<T> {

diff --git a/...nt/src/main/java/org/apache/hudi/table/action/commit/FlinkDeleteCommitActionExecutor.java b/...nt/src/main/java/org/apache/hudi/table/action/commit/FlinkDeleteCommitActionExecutor.java
@@ -30,6 +30,9 @@
 
 import java.util.List;
 
+/**
+ * Flink delete commit action executor.
+ */
 public class FlinkDeleteCommitActionExecutor<T extends HoodieRecordPayload<T>> extends BaseFlinkCommitActionExecutor<T> {
   private final List<HoodieKey> keys;
 

diff --git a/...udi-flink-client/src/main/java/org/apache/hudi/table/action/commit/FlinkDeleteHelper.java b/...udi-flink-client/src/main/java/org/apache/hudi/table/action/commit/FlinkDeleteHelper.java
@@ -42,6 +42,9 @@
 import java.util.List;
 import java.util.stream.Collectors;
 
+/**
+ * Flink delete helper.
+ */
 @SuppressWarnings("checkstyle:LineLength")
 public class FlinkDeleteHelper<R> extends
     BaseDeleteHelper<EmptyHoodieRecordPayload, List<HoodieRecord<EmptyHoodieRecordPayload>>, List<HoodieKey>, List<WriteStatus>, R> {

diff --git a/...nt/src/main/java/org/apache/hudi/table/action/commit/FlinkInsertCommitActionExecutor.java b/...nt/src/main/java/org/apache/hudi/table/action/commit/FlinkInsertCommitActionExecutor.java
@@ -30,6 +30,9 @@
 
 import java.util.List;
 
+/**
+ * Flink insert commit action executor.
+ */
 public class FlinkInsertCommitActionExecutor<T extends HoodieRecordPayload<T>> extends BaseFlinkCommitActionExecutor<T> {
 
   private List<HoodieRecord<T>> inputRecords;

diff --git a/...in/java/org/apache/hudi/table/action/commit/FlinkInsertOverwriteCommitActionExecutor.java b/...in/java/org/apache/hudi/table/action/commit/FlinkInsertOverwriteCommitActionExecutor.java
@@ -31,6 +31,9 @@
 
 import java.util.List;
 
+/**
+ * Flink INSERT OVERWRITE commit action executor.
+ */
 public class FlinkInsertOverwriteCommitActionExecutor<T extends HoodieRecordPayload<T>>
     extends BaseFlinkCommitActionExecutor<T> {
 

diff --git a/...va/org/apache/hudi/table/action/commit/FlinkInsertOverwriteTableCommitActionExecutor.java b/...va/org/apache/hudi/table/action/commit/FlinkInsertOverwriteTableCommitActionExecutor.java
@@ -30,6 +30,9 @@
 
 import java.util.List;
 
+/**
+ * Flink INSERT OVERWRITE TABLE commit action executor.
+ */
 public class FlinkInsertOverwriteTableCommitActionExecutor<T extends HoodieRecordPayload<T>>
     extends FlinkInsertOverwriteCommitActionExecutor<T> {
 

diff --git a/...main/java/org/apache/hudi/table/action/commit/FlinkInsertPreppedCommitActionExecutor.java b/...main/java/org/apache/hudi/table/action/commit/FlinkInsertPreppedCommitActionExecutor.java
@@ -30,6 +30,9 @@
 
 import java.util.List;
 
+/**
+ * Flink insert prepped commit action executor.
+ */
 public class FlinkInsertPreppedCommitActionExecutor<T extends HoodieRecordPayload<T>> extends BaseFlinkCommitActionExecutor<T> {
 
   private final List<HoodieRecord<T>> preppedRecords;

diff --git a/...hudi-flink-client/src/main/java/org/apache/hudi/table/action/commit/FlinkMergeHelper.java b/...hudi-flink-client/src/main/java/org/apache/hudi/table/action/commit/FlinkMergeHelper.java
@@ -44,6 +44,9 @@
 import java.util.Iterator;
 import java.util.List;
 
+/**
+ * Flink merge helper.
+ */
 public class FlinkMergeHelper<T extends HoodieRecordPayload> extends BaseMergeHelper<T, List<HoodieRecord<T>>,
     List<HoodieKey>, List<WriteStatus>> {
 

diff --git a/...nt/src/main/java/org/apache/hudi/table/action/commit/FlinkUpsertCommitActionExecutor.java b/...nt/src/main/java/org/apache/hudi/table/action/commit/FlinkUpsertCommitActionExecutor.java
@@ -30,6 +30,9 @@
 
 import java.util.List;
 
+/**
+ * Flink upsert commit action executor.
+ */
 public class FlinkUpsertCommitActionExecutor<T extends HoodieRecordPayload<T>> extends BaseFlinkCommitActionExecutor<T> {
 
   private List<HoodieRecord<T>> inputRecords;

diff --git a/...main/java/org/apache/hudi/table/action/commit/FlinkUpsertPreppedCommitActionExecutor.java b/...main/java/org/apache/hudi/table/action/commit/FlinkUpsertPreppedCommitActionExecutor.java
@@ -30,6 +30,9 @@
 
 import java.util.List;
 
+/**
+ * Flink upsert prepped commit action executor.
+ */
 public class FlinkUpsertPreppedCommitActionExecutor<T extends HoodieRecordPayload<T>> extends BaseFlinkCommitActionExecutor<T> {
 
   private final List<HoodieRecord<T>> preppedRecords;

diff --git a/...in/java/org/apache/hudi/table/action/commit/delta/BaseFlinkDeltaCommitActionExecutor.java b/...in/java/org/apache/hudi/table/action/commit/delta/BaseFlinkDeltaCommitActionExecutor.java
@@ -34,6 +34,9 @@
 import java.util.Iterator;
 import java.util.List;
 
+/**
+ * Base flink delta commit action executor.
+ */
 public abstract class BaseFlinkDeltaCommitActionExecutor<T extends HoodieRecordPayload<T>>
     extends BaseFlinkCommitActionExecutor<T> {
 

diff --git a/.../java/org/apache/hudi/table/action/commit/delta/FlinkUpsertDeltaCommitActionExecutor.java b/.../java/org/apache/hudi/table/action/commit/delta/FlinkUpsertDeltaCommitActionExecutor.java
@@ -30,6 +30,9 @@
 
 import java.util.List;
 
+/**
+ * Flink upsert delta commit action executor.
+ */
 public class FlinkUpsertDeltaCommitActionExecutor<T extends HoodieRecordPayload<T>>
     extends BaseFlinkDeltaCommitActionExecutor<T> {
   private final List<HoodieRecord<T>> inputRecords;

diff --git a/...rg/apache/hudi/table/action/commit/delta/FlinkUpsertPreppedDeltaCommitActionExecutor.java b/...rg/apache/hudi/table/action/commit/delta/FlinkUpsertPreppedDeltaCommitActionExecutor.java
@@ -30,6 +30,9 @@
 
 import java.util.List;
 
+/**
+ * Flink upsert prepped delta commit action executor.
+ */
 public class FlinkUpsertPreppedDeltaCommitActionExecutor<T extends HoodieRecordPayload<T>>
     extends BaseFlinkDeltaCommitActionExecutor<T> {
 

diff --git a/...link-client/src/test/java/org/apache/hudi/client/common/TestHoodieFlinkEngineContext.java b/...link-client/src/test/java/org/apache/hudi/client/common/TestHoodieFlinkEngineContext.java
@@ -19,8 +19,8 @@
 package org.apache.hudi.client.common;
 
 import org.apache.hudi.client.FlinkTaskContextSupplier;
-
 import org.apache.hudi.common.util.collection.ImmutablePair;
+
 import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
-Original file line number
+Diff line change
@@ Expand Up / @@ -61,5 +61,5 @@ public Option<String> getProperty(EngineProperty prop) { @@
         // no operation for now
         return Option.empty();
       }
     }