From e64fe5505487f3ee591b2b5d044c2c57989f8991 Mon Sep 17 00:00:00 2001
From: Danny Chan <yuzhao.cyz@gmail.com>
Date: Fri, 25 Jun 2021 08:57:56 +0800
Subject: [PATCH] [HUDI-2068] Skip the assign state for SmallFileAssign when
 the state can not assign initially (#3148)

---
 .../java/org/apache/hudi/sink/partitioner/BucketAssigner.java  | 3 +--
 1 file changed, 1 insertion(+), 2 deletions(-)

diff --git a/hudi-flink/src/main/java/org/apache/hudi/sink/partitioner/BucketAssigner.java b/hudi-flink/src/main/java/org/apache/hudi/sink/partitioner/BucketAssigner.java
index e2f32937b..6d805ce8d 100644
--- a/hudi-flink/src/main/java/org/apache/hudi/sink/partitioner/BucketAssigner.java
+++ b/hudi-flink/src/main/java/org/apache/hudi/sink/partitioner/BucketAssigner.java
@@ -228,7 +228,7 @@ public class BucketAssigner implements AutoCloseable {
         return false;
       }
       SmallFileAssignState state = states[assignIdx];
-      if (!state.canAssign()) {
+      while (!state.canAssign()) {
         assignIdx += 1;
         if (assignIdx >= states.length) {
           noSpace = true;
@@ -236,7 +236,6 @@ public class BucketAssigner implements AutoCloseable {
         }
         // move to next slot if possible
         state = states[assignIdx];
-        assert state.canAssign();
       }
       state.assign();
       return true;