chore(config): dynamic change config based on heap size

Author: superhx

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

@@ -154,7 +154,7 @@ public class AutoMQConfig {
     public static final String S3_NETWORK_BASELINE_BANDWIDTH_CONFIG = "s3.network.baseline.bandwidth";
     public static final String S3_NETWORK_BASELINE_BANDWIDTH_DOC = "The total available bandwidth for object storage requests. This is used to prevent stream set object compaction and catch-up read from monopolizing normal read and write traffic. Produce and Consume will also separately consume traffic in and traffic out. " +
         "For example, suppose this value is set to 100MB/s, and the normal read and write traffic is 80MB/s, then the available traffic for stream set object compaction is 20MB/s.";
-    public static final long S3_NETWORK_BASELINE_BANDWIDTH = 100 * 1024 * 1024; // 100MB/s
+    public static final long S3_NETWORK_BASELINE_BANDWIDTH = 1024 * 1024 * 1024; // 1GBps
 
     public static final String S3_NETWORK_REFILL_PERIOD_MS_CONFIG = "s3.network.refill.period.ms";
     public static final String S3_NETWORK_REFILL_PERIOD_MS_DOC = "The network bandwidth token refill period in milliseconds.";

core/src/main/java/kafka/automq/zerozone/SnapshotReadPartitionsManager.java

@@ -282,7 +282,7 @@ public void close() {
             });
         }
 
-        private void run() {
+        void run() {
             eventLoop.execute(this::unsafeRun);
         }
 

core/src/main/scala/kafka/log/streamaspect/ElasticLog.scala

@@ -640,7 +640,10 @@ class ElasticLog(val metaStream: MetaStream,
 }
 
 object ElasticLog extends Logging {
-    private val APPEND_PERMIT = 100 * 1024 * 1024
+    private val APPEND_PERMIT = Systems.getEnvInt("AUTOMQ_APPEND_PERMIT_SIZE",
+        // autoscale the append permit size based on heap size, min 100MiB, max 1GiB, every 6GB heap add 100MiB permit
+        Math.min(1024, 100 * Math.max(1, (Systems.HEAP_MEMORY_SIZE / (1024 * 1024 * 1024) / 6)).asInstanceOf[Int]) * 1024 * 1024
+    )
     private val APPEND_PERMIT_SEMAPHORE = new Semaphore(APPEND_PERMIT)
     S3StreamKafkaMetricsManager.setLogAppendPermitNumSupplier(() => APPEND_PERMIT_SEMAPHORE.availablePermits())
 

core/src/main/scala/kafka/server/streamaspect/ElasticReplicaManager.scala

@@ -4,7 +4,7 @@ import com.automq.stream.api.exceptions.FastReadFailFastException
 import com.automq.stream.s3.metrics.stats.NetworkStats
 import com.automq.stream.s3.metrics.{MetricsLevel, TimerUtil}
 import com.automq.stream.s3.network.{AsyncNetworkBandwidthLimiter, GlobalNetworkBandwidthLimiters, ThrottleStrategy}
-import com.automq.stream.utils.FutureUtil
+import com.automq.stream.utils.{FutureUtil, Systems}
 import com.automq.stream.utils.threads.S3StreamThreadPoolMonitor
 import kafka.automq.interceptor.{ClientIdKey, ClientIdMetadata, TrafficInterceptor}
 import kafka.automq.kafkalinking.KafkaLinkingManager
@@ -124,8 +124,12 @@ class ElasticReplicaManager(
     fetchExecutorQueueSizeGaugeMap
   })
 
-  private val fastFetchLimiter = new FairLimiter(200 * 1024 * 1024, FETCH_LIMITER_FAST_NAME) // 200MiB
-  private val slowFetchLimiter = new FairLimiter(200 * 1024 * 1024, FETCH_LIMITER_SLOW_NAME) // 200MiB
+  private val fetchLimiterSize = Systems.getEnvInt("AUTOMQ_FETCH_LIMITER_SIZE",
+    // autoscale the fetch limiter size based on heap size, min 200MiB, max 1GiB, every 3GB heap add 100MiB limiter
+     Math.min(1024, 100 * Math.max(2, (Systems.HEAP_MEMORY_SIZE / (1024 * 1024 * 1024) / 3)).asInstanceOf[Int]) * 1024 * 1024
+  )
+  private val fastFetchLimiter = new FairLimiter(fetchLimiterSize, FETCH_LIMITER_FAST_NAME)
+  private val slowFetchLimiter = new FairLimiter(fetchLimiterSize, FETCH_LIMITER_SLOW_NAME)
   private val fetchLimiterWaitingTasksGaugeMap = new util.HashMap[String, Integer]()
   S3StreamKafkaMetricsManager.setFetchLimiterWaitingTaskNumSupplier(() => {
     fetchLimiterWaitingTasksGaugeMap.put(FETCH_LIMITER_FAST_NAME, fastFetchLimiter.waitingThreads())

core/src/test/java/kafka/automq/zerozone/SnapshotReadPartitionsManagerTest.java

@@ -62,6 +62,7 @@
 import static org.mockito.ArgumentMatchers.anyLong;
 import static org.mockito.Mockito.doAnswer;
 import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.timeout;
 import static org.mockito.Mockito.times;
 import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.when;
@@ -152,10 +153,8 @@ public void testSubscriber_zerozonev1() throws Exception {
         operationBatch.operations.add(snapshotWithOperation(2, Map.of(4L, 6L), SnapshotOperation.ADD));
         subscriber.onNewOperationBatch(operationBatch);
 
-        subscriber.unsafeRun();
-
-        verify(dataLoader, times(1)).replayWal();
-
+        subscriber.run();
+        verify(dataLoader, timeout(1000L).times(1)).replayWal();
         awaitEventLoopClear();
 
         assertEquals(2, subscriber.partitions.size());

s3stream/src/main/java/com/automq/stream/s3/cache/blockcache/StreamReader.java

@@ -34,6 +34,7 @@
 import com.automq.stream.s3.objects.ObjectManager;
 import com.automq.stream.utils.FutureUtil;
 import com.automq.stream.utils.LogSuppressor;
+import com.automq.stream.utils.Systems;
 import com.automq.stream.utils.Time;
 import com.automq.stream.utils.threads.EventLoop;
 import com.automq.stream.utils.threads.EventLoopSafe;
@@ -64,7 +65,7 @@
     public static final int GET_OBJECT_STEP = 4;
     private static final Logger LOGGER = LoggerFactory.getLogger(StreamReader.class);
     static final int READAHEAD_SIZE_UNIT = 1024 * 1024 / 2;
-    private static final int MAX_READAHEAD_SIZE = 32 * 1024 * 1024;
+    private static final int MAX_READAHEAD_SIZE = Systems.getEnvInt("AUTOMQ_MAX_READAHEAD_SIZE", 32 * 1024 * 1024);
     private static final long READAHEAD_RESET_COLD_DOWN_MILLS = TimeUnit.MINUTES.toMillis(1);
     private static final long READAHEAD_AVAILABLE_BYTES_THRESHOLD = 32L * 1024 * 1024;
     private static final LogSuppressor READAHEAD_RESET_LOG_SUPPRESSOR = new LogSuppressor(LOGGER, 30000);

s3stream/src/main/java/com/automq/stream/utils/Systems.java

@@ -21,8 +21,12 @@
 
 import org.apache.commons.lang3.StringUtils;
 
+import io.netty.util.internal.PlatformDependent;
+
 public class Systems {
     public static final int CPU_CORES = Runtime.getRuntime().availableProcessors();
+    public static final long HEAP_MEMORY_SIZE = Runtime.getRuntime().maxMemory();
+    public static final long DIRECT_MEMORY_SIZE = PlatformDependent.maxDirectMemory();
 
     public static long getEnvLong(String name, long defaultValue) {
         String value = System.getenv(name);