[flink] branch master updated: [FLINK-31139][state/changelog] not upload empty state changelog file

[roman]

This is an automated email from the ASF dual-hosted git repository.

roman pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git


The following commit(s) were added to refs/heads/master by this push:
     new fb37e6a87d9 [FLINK-31139][state/changelog] not upload empty state 
changelog file
fb37e6a87d9 is described below

commit fb37e6a87d92912897c6a8c4b182048e13686dee
Author: wangfeifan <zoltar9...@163.com>
AuthorDate: Mon Feb 20 20:26:58 2023 +0800

    [FLINK-31139][state/changelog] not upload empty state changelog file
---
 .../flink/changelog/fs/BatchingStateChangeUploadScheduler.java       | 5 +++++
 1 file changed, 5 insertions(+)

diff --git 
a/flink-dstl/flink-dstl-dfs/src/main/java/org/apache/flink/changelog/fs/BatchingStateChangeUploadScheduler.java
 
b/flink-dstl/flink-dstl-dfs/src/main/java/org/apache/flink/changelog/fs/BatchingStateChangeUploadScheduler.java
index 3495c6bc525..39023bf96d1 100644
--- 
a/flink-dstl/flink-dstl-dfs/src/main/java/org/apache/flink/changelog/fs/BatchingStateChangeUploadScheduler.java
+++ 
b/flink-dstl/flink-dstl-dfs/src/main/java/org/apache/flink/changelog/fs/BatchingStateChangeUploadScheduler.java
@@ -223,6 +223,11 @@ class BatchingStateChangeUploadScheduler implements 
StateChangeUploadScheduler {
             scheduledBytesCounter = 0;
             scheduledFuture = null;
         }
+
+        if (tasks.size() == 0) {
+            return;
+        }
+
         try {
             Throwable error = getErrorSafe();
             if (error != null) {