feat(s3stream): add trigger wal upload interval (#2347)

Signed-off-by: Robin Han <[email protected]>

Author: superhx

core/src/main/java/kafka/automq/AutoMQConfig.java

@@ -71,6 +71,10 @@ public class AutoMQConfig {
     public static final String S3_WAL_UPLOAD_THRESHOLD_CONFIG = "s3.wal.upload.threshold";
     public static final String S3_WAL_UPLOAD_THRESHOLD_DOC = "The threshold at which WAL triggers upload to object storage. The configuration value needs to be less than s3.wal.cache.size. The larger the configuration value, the higher the data aggregation and the lower the cost of metadata storage.";
 
+    public static final String S3_WAL_UPLOAD_INTERVAL_MS_CONFIG = "s3.wal.upload.interval.ms";
+    public static final String S3_WAL_UPLOAD_INTERVAL_MS_DOC = "The interval at which WAL triggers upload to object storage. -1 means only upload by size trigger";
+    public static final long S3_WAL_UPLOAD_INTERVAL_MS_DEFAULT = -1L;
+
     public static final String S3_STREAM_SPLIT_SIZE_CONFIG = "s3.stream.object.split.size";
     public static final String S3_STREAM_SPLIT_SIZE_DOC = "The S3 stream object split size threshold when upload delta WAL or compact stream set object.";
 
@@ -246,6 +250,7 @@ public static void define(ConfigDef configDef) {
             .define(AutoMQConfig.S3_WAL_PATH_CONFIG, STRING, null, HIGH, AutoMQConfig.S3_WAL_PATH_DOC)
             .define(AutoMQConfig.S3_WAL_CACHE_SIZE_CONFIG, LONG, -1L, MEDIUM, AutoMQConfig.S3_WAL_CACHE_SIZE_DOC)
             .define(AutoMQConfig.S3_WAL_UPLOAD_THRESHOLD_CONFIG, LONG, -1L, MEDIUM, AutoMQConfig.S3_WAL_UPLOAD_THRESHOLD_DOC)
+            .define(AutoMQConfig.S3_WAL_UPLOAD_INTERVAL_MS_CONFIG, LONG, S3_WAL_UPLOAD_INTERVAL_MS_DEFAULT, LOW, AutoMQConfig.S3_WAL_UPLOAD_INTERVAL_MS_DOC)
             .define(AutoMQConfig.S3_STREAM_SPLIT_SIZE_CONFIG, INT, 8388608, MEDIUM, AutoMQConfig.S3_STREAM_SPLIT_SIZE_DOC)
             .define(AutoMQConfig.S3_OBJECT_BLOCK_SIZE_CONFIG, INT, 524288, MEDIUM, AutoMQConfig.S3_OBJECT_BLOCK_SIZE_DOC)
             .define(AutoMQConfig.S3_OBJECT_PART_SIZE_CONFIG, INT, 16777216, MEDIUM, AutoMQConfig.S3_OBJECT_PART_SIZE_DOC)

core/src/main/scala/kafka/log/stream/s3/ConfigUtils.java

@@ -26,6 +26,7 @@ public static Config to(KafkaConfig s) {
             .walConfig(config.walConfig())
             .walCacheSize(s.s3WALCacheSize())
             .walUploadThreshold(s.s3WALUploadThreshold())
+            .walUploadIntervalMs(s.s3WALUploadIntervalMs())
             .streamSplitSize(s.s3StreamSplitSize())
             .objectBlockSize(s.s3ObjectBlockSize())
             .objectPartSize(s.s3ObjectPartSize())

core/src/main/scala/kafka/server/KafkaConfig.scala

@@ -775,6 +775,7 @@ class KafkaConfig private(doLog: Boolean, val props: util.Map[_, _])
   val s3ObjectPartSize = getInt(AutoMQConfig.S3_OBJECT_PART_SIZE_CONFIG)
   val s3StreamAllocatorPolicy = Enum.valueOf(classOf[ByteBufAllocPolicy], getString(AutoMQConfig.S3_STREAM_ALLOCATOR_POLICY_CONFIG))
   val (s3WALCacheSize, s3BlockCacheSize, s3WALUploadThreshold) = adjustS3Configs(s3StreamAllocatorPolicy)
+  val s3WALUploadIntervalMs = getLong(AutoMQConfig.S3_WAL_UPLOAD_INTERVAL_MS_CONFIG)
   val s3StreamObjectCompactionTaskIntervalMinutes = getInt(AutoMQConfig.S3_STREAM_OBJECT_COMPACTION_INTERVAL_MINUTES_CONFIG)
   val s3StreamObjectCompactionMaxSizeBytes = getLong(AutoMQConfig.S3_STREAM_OBJECT_COMPACTION_MAX_SIZE_BYTES_CONFIG)
   val s3ControllerRequestRetryMaxCount = getInt(AutoMQConfig.S3_CONTROLLER_REQUEST_RETRY_MAX_COUNT_CONFIG)

s3stream/src/main/java/com/automq/stream/s3/Config.java

@@ -25,6 +25,8 @@ public class Config {
     private String walConfig = "0@file:///tmp/s3stream_wal";
     private long walCacheSize = 200 * 1024 * 1024;
     private long walUploadThreshold = 100 * 1024 * 1024;
+    // -1L means don't upload by time
+    private long walUploadIntervalMs = -1L;
     private int streamSplitSize = 16777216;
     private int objectBlockSize = 1048576;
     private int objectPartSize = 16777216;
@@ -73,6 +75,10 @@ public long walUploadThreshold() {
         return walUploadThreshold;
     }
 
+    public long walUploadIntervalMs() {
+        return walUploadIntervalMs;
+    }
+
     public int streamSplitSize() {
         return streamSplitSize;
     }
@@ -182,6 +188,11 @@ public Config walUploadThreshold(long s3WALObjectSize) {
         return this;
     }
 
+    public Config walUploadIntervalMs(long s3WALUploadIntervalMs) {
+        this.walUploadIntervalMs = s3WALUploadIntervalMs;
+        return this;
+    }
+
     public Config streamSplitSize(int s3StreamSplitSize) {
         this.streamSplitSize = s3StreamSplitSize;
         return this;

s3stream/src/main/java/com/automq/stream/s3/S3Storage.java

@@ -150,6 +150,9 @@ public S3Storage(Config config, WriteAheadLog deltaWAL, StreamManager streamMana
         this.drainBackoffTask = this.backgroundExecutor.scheduleWithFixedDelay(this::tryDrainBackoffRecords, 100, 100, TimeUnit.MILLISECONDS);
         S3StreamMetricsManager.registerInflightWALUploadTasksCountSupplier(this.inflightWALUploadTasks::size);
         S3StreamMetricsManager.registerDeltaWalPendingUploadBytesSupplier(this.pendingUploadBytes::get);
+        if (config.walUploadIntervalMs() > 0) {
+            this.backgroundExecutor.scheduleWithFixedDelay(this::maybeForceUpload, config.walUploadIntervalMs(), config.walUploadIntervalMs(), TimeUnit.MILLISECONDS);
+        }
     }
 
     /**
@@ -635,7 +638,6 @@ private boolean hasInflightForceUploadTask() {
     }
 
     private CompletableFuture<Void> forceUpload() {
-        LOGGER.info("force upload all streams");
         CompletableFuture<Void> cf = forceUpload(LogCache.MATCH_ALL_STREAMS);
         cf.whenComplete((nil, ignored) -> forceUploadCallback());
         return cf;