lanyuanxiaoyao/hudi
1
0
Fork 0
Code Activity

[HUDI-2049] StreamWriteFunction should wait for the next inflight instant time before flushing (#3123)

Browse Source
This commit is contained in:
Danny Chan
2021-06-21 20:15:27 +08:00
committed by GitHub
parent 429e9fb5fe
commit adf167991a
1 changed files with 1 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
hudi-flink/src/main/java/org/apache/hudi/sink/StreamWriteFunction.java
View File

@@ -54,7 +54,6 @@ import java.util.HashMap;
import java.util.LinkedHashMap; import java.util.LinkedHashMap;
import java.util.List; import java.util.List;
import java.util.Map; import java.util.Map;
import java.util.Objects;
import java.util.Random; import java.util.Random;
import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeUnit;
import java.util.function.BiFunction; import java.util.function.BiFunction;
@@ -494,7 +493,7 @@ public class StreamWriteFunction<K, I, O>
long waitingTime = 0L; long waitingTime = 0L;
long ckpTimeout = config.getLong(FlinkOptions.WRITE_COMMIT_ACK_TIMEOUT); long ckpTimeout = config.getLong(FlinkOptions.WRITE_COMMIT_ACK_TIMEOUT);
long interval = 500L; long interval = 500L;
while (Objects.equals(instant, this.currentInstant)) { while (instant == null || instant.equals(this.currentInstant)) {
// sleep for a while // sleep for a while
try { try {
if (waitingTime > ckpTimeout) { if (waitingTime > ckpTimeout) {