improve jitter config documentation

Author: HuangZhenQiu

docs/layouts/shortcodes/generated/expert_rocksdb_section.html

@@ -18,7 +18,7 @@
             <td><h5>state.backend.rocksdb.checkpoint.upload-jitter</h5></td>
             <td style="word-wrap: break-word;">0 ms</td>
             <td>Duration</td>
-            <td>The time interval used to create jitter for each checkpoint file upload.</td>
+            <td>The jitter time interval determines the amount of random delay introduced before uploading each checkpoint file. This jitter is applied to help stagger the upload times of state files, thereby introducing a small, randomized delay for each upload operation. The purpose of this mechanism is to avoid simultaneous upload bursts, which could otherwise overwhelm the remote storage system. By distributing the upload requests more evenly over time, jitter helps maintain system stability and prevents potential performance degradation during checkpoint completion. It is therefore recommended to configure the jitter interval at the seconds level, ensuring a balanced trade-off between randomness and upload efficiency.</td>
         </tr>
         <tr>
             <td><h5>state.backend.rocksdb.localdir</h5></td>

docs/layouts/shortcodes/generated/rocksdb_configuration.html

@@ -18,7 +18,7 @@
             <td><h5>state.backend.rocksdb.checkpoint.upload-jitter</h5></td>
             <td style="word-wrap: break-word;">0 ms</td>
             <td>Duration</td>
-            <td>The time interval used to create jitter for each checkpoint file upload.</td>
+            <td>The jitter time interval determines the amount of random delay introduced before uploading each checkpoint file. This jitter is applied to help stagger the upload times of state files, thereby introducing a small, randomized delay for each upload operation. The purpose of this mechanism is to avoid simultaneous upload bursts, which could otherwise overwhelm the remote storage system. By distributing the upload requests more evenly over time, jitter helps maintain system stability and prevents potential performance degradation during checkpoint completion. It is therefore recommended to configure the jitter interval at the seconds level, ensuring a balanced trade-off between randomness and upload efficiency.</td>
         </tr>
         <tr>
             <td><h5>state.backend.rocksdb.localdir</h5></td>

flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/state/rocksdb/RocksDBOptions.java

@@ -100,7 +100,13 @@ public class RocksDBOptions {
                     .durationType()
                     .defaultValue(Duration.ZERO)
                     .withDescription(
-                            "The time interval used to create jitter for each checkpoint file upload.");
+                            "The jitter time interval determines the amount of random delay introduced before uploading each checkpoint file. "
+                                    + "This jitter is applied to help stagger the upload times of state files, thereby introducing a small, "
+                                    + "randomized delay for each upload operation. The purpose of this mechanism is to avoid simultaneous upload bursts, "
+                                    + "which could otherwise overwhelm the remote storage system. By distributing the upload requests more evenly over time, "
+                                    + "jitter helps maintain system stability and prevents potential performance degradation during checkpoint completion. "
+                                    + "It is therefore recommended to configure the jitter interval at the seconds level, ensuring a balanced trade-off "
+                                    + "between randomness and upload efficiency.");
 
     /** The predefined settings for RocksDB DBOptions and ColumnFamilyOptions by Flink community. */
     @Documentation.Section(Documentation.Sections.EXPERT_ROCKSDB)

flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/state/rocksdb/RocksDBStateUploader.java

@@ -58,6 +58,7 @@ public class RocksDBStateUploader implements Closeable {
     private final Duration uploadJitter;
     private final Random random;
     private final RocksDBStateDataTransferHelper transfer;
+    private final JitterConsumer jitterConsumer;
 
     @VisibleForTesting
     public RocksDBStateUploader(int numberOfSnapshottingThreads, Duration uploadJitter) {
@@ -69,6 +70,7 @@ public RocksDBStateUploader(int numberOfSnapshottingThreads, Duration uploadJitt
     public RocksDBStateUploader(RocksDBStateDataTransferHelper transfer, Duration uploadJitter) {
         this.transfer = transfer;
         this.uploadJitter = uploadJitter;
+        this.jitterConsumer = new JitterConsumer();
         this.random = new Random();
     }
 
@@ -152,7 +154,7 @@ private HandleAndLocalPath uploadLocalFileToCheckpointFs(
 
         try {
             // add a random jitter
-            applyJitter(new JitterConsumer());
+            applyJitter(jitterConsumer);
             final byte[] buffer = new byte[READ_BUFFER_SIZE];
 
             inputStream = Files.newInputStream(filePath);
@@ -211,7 +213,10 @@ public void accept(Long milliseconds) {
             try {
                 Thread.sleep(milliseconds);
             } catch (InterruptedException e) {
-                LOG.error("Fail to apply jitter in RocksDBStateUploader.", e);
+                LOG.error(
+                        "Fail to apply {} milliseconds jitter in RocksDBStateUploader.",
+                        milliseconds,
+                        e);
             }
         }
     }