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


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

commit 8798fde04e4a28af8bee7f2641b4aa89f2e0995f
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) {