From 42ec7e30d74104a15ff76cb004fe7a0c2649abe6 Mon Sep 17 00:00:00 2001
From: Danny Chan <yuzhao.cyz@gmail.com>
Date: Tue, 11 May 2021 09:14:51 +0800
Subject: [PATCH] [HUDI-1890] FlinkCreateHandle and FlinkAppendHandle canWrite
 should always return true (#2933)

The method #canWrite should always return true because they can already
write based on file size, e.g. the BucketAssigner.
---
 .../src/main/java/org/apache/hudi/io/FlinkAppendHandle.java | 5 +++++
 .../src/main/java/org/apache/hudi/io/FlinkCreateHandle.java | 6 ++++++
 2 files changed, 11 insertions(+)

diff --git a/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/io/FlinkAppendHandle.java b/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/io/FlinkAppendHandle.java
index b827bf232..bc6eba8eb 100644
--- a/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/io/FlinkAppendHandle.java
+++ b/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/io/FlinkAppendHandle.java
@@ -75,6 +75,11 @@ public class FlinkAppendHandle<T extends HoodieRecordPayload, I, K, O> extends H
         && hoodieRecord.getCurrentLocation().getInstantTime().equals("U");
   }
 
+  @Override
+  public boolean canWrite(HoodieRecord record) {
+    return true;
+  }
+
   /**
    * Returns whether there is need to bootstrap this file handle.
    * E.G. the first time that the handle is created.
diff --git a/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/io/FlinkCreateHandle.java b/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/io/FlinkCreateHandle.java
index 7f85deadc..fc0f8efe0 100644
--- a/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/io/FlinkCreateHandle.java
+++ b/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/io/FlinkCreateHandle.java
@@ -21,6 +21,7 @@ package org.apache.hudi.io;
 import org.apache.hudi.client.WriteStatus;
 import org.apache.hudi.common.engine.TaskContextSupplier;
 import org.apache.hudi.common.fs.FSUtils;
+import org.apache.hudi.common.model.HoodieRecord;
 import org.apache.hudi.common.model.HoodieRecordPayload;
 import org.apache.hudi.common.util.HoodieTimer;
 import org.apache.hudi.common.util.collection.Pair;
@@ -123,6 +124,11 @@ public class FlinkCreateHandle<T extends HoodieRecordPayload, I, K, O>
     }
   }
 
+  @Override
+  public boolean canWrite(HoodieRecord record) {
+    return true;
+  }
+
   /**
    * Use the writeToken + "-" + rollNumber as the new writeToken of a mini-batch write.
    */