[flink] branch release-1.16 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 release-1.16
in repository https://gitbox.apache.org/repos/asf/flink.git


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

commit 5157ac5921d406f577c83a4fe57b373d8ae0bf79
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) {