[BUG] WAL upload method gets negative stream size which results in guava RateLimiter throwing exception and kafka can't startup

[keashem]

What went wrong?

when broker recovers from crash, it loads WAL's data; but in some case, the datas in wal not upload to s3 storage is too large, which results in WAL upload method getting negative stream size , and guava RateLimiter throws exception and kafka can't startup ,the log info as follow:

[2025-03-30 10:43:34,149] INFO try recover from crash, recover records bytes size 4287034156 (com.automq.stream.s3.S3Storage)
[2025-03-30 10:43:34,194] ERROR upload run with unexpected exception (com.automq.stream.s3.DeltaWALUploadTask)
java.lang.IllegalArgumentException: Requested permits (-326332735) must be positive
at com.google.common.base.Preconditions.checkArgument(Preconditions.java:191)
at com.google.common.util.concurrent.RateLimiter.checkPermits(RateLimiter.java:496)
at com.google.common.util.concurrent.RateLimiter.tryAcquire(RateLimiter.java:413)
at com.google.common.util.concurrent.RateLimiter.tryAcquire(RateLimiter.java:368)
at com.automq.stream.utils.AsyncRateLimiter.acquire(AsyncRateLimiter.java:43)
at com.automq.stream.s3.DeltaWALUploadTask.acquireLimiter(DeltaWALUploadTask.java:112)
at com.automq.stream.s3.DeltaWALUploadTask.upload0(DeltaWALUploadTask.java:147)
at com.automq.stream.s3.DeltaWALUploadTask.lambda$upload$1(DeltaWALUploadTask.java:116)
at com.automq.stream.utils.FutureUtil.exec(FutureUtil.java:89)
at com.automq.stream.s3.DeltaWALUploadTask.lambda$upload$2(DeltaWALUploadTask.java:116)
at java.base/java.util.concurrent.CompletableFuture$UniAccept.tryFire(CompletableFuture.java:718)
at java.base/java.util.concurrent.CompletableFuture$Completion.run(CompletableFuture.java:482)
at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1136)
at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:635)
at java.base/java.lang.Thread.run(Thread.java:833)
[2025-03-30 10:43:34,198] ERROR S3Storage start fail (com.automq.stream.s3.S3Storage)
java.util.concurrent.ExecutionException: java.lang.IllegalArgumentException: Requested permits (-326332735) must be positive
at java.base/java.util.concurrent.CompletableFuture.reportGet(CompletableFuture.java:396)
at java.base/java.util.concurrent.CompletableFuture.get(CompletableFuture.java:2073)
at com.automq.stream.s3.S3Storage.recover0(S3Storage.java:367)
at com.automq.stream.s3.S3Storage.recover(S3Storage.java:333)
at com.automq.stream.s3.S3Storage.startup(S3Storage.java:320)
at kafka.log.stream.s3.DefaultS3Client.start(DefaultS3Client.java:139)
at kafka.log.streamaspect.AlwaysSuccessClient.start(AlwaysSuccessClient.java:101)
at kafka.log.streamaspect.ElasticLogManager.startup(ElasticLogManager.scala:107)
at kafka.log.streamaspect.ElasticLogManager$.$anonfun$init$1(ElasticLogManager.scala:140)
at kafka.log.streamaspect.ElasticLogManager$.$anonfun$init$1$adapted(ElasticLogManager.scala:140)
at scala.Option.foreach(Option.scala:437)
at kafka.log.streamaspect.ElasticLogManager$.init(ElasticLogManager.scala:140)
at kafka.server.BrokerServer.startup(BrokerServer.scala:555)
at kafka.server.KafkaRaftServer.$anonfun$startup$2(KafkaRaftServer.scala:86)
at kafka.server.KafkaRaftServer.$anonfun$startup$2$adapted(KafkaRaftServer.scala:86)
at scala.Option.foreach(Option.scala:437)
at kafka.server.KafkaRaftServer.startup(KafkaRaftServer.scala:86)
at kafka.Kafka$.main(Kafka.scala:153)
at kafka.Kafka.main(Kafka.scala)
Caused by: java.lang.IllegalArgumentException: Requested permits (-326332735) must be positive
at com.google.common.base.Preconditions.checkArgument(Preconditions.java:191)
at com.google.common.util.concurrent.RateLimiter.checkPermits(RateLimiter.java:496)
at com.google.common.util.concurrent.RateLimiter.tryAcquire(RateLimiter.java:413)
at com.google.common.util.concurrent.RateLimiter.tryAcquire(RateLimiter.java:368)
at com.automq.stream.utils.AsyncRateLimiter.acquire(AsyncRateLimiter.java:43)
at com.automq.stream.s3.DeltaWALUploadTask.acquireLimiter(DeltaWALUploadTask.java:112)
at com.automq.stream.s3.DeltaWALUploadTask.upload0(DeltaWALUploadTask.java:147)
at com.automq.stream.s3.DeltaWALUploadTask.lambda$upload$1(DeltaWALUploadTask.java:116)
at com.automq.stream.utils.FutureUtil.exec(FutureUtil.java:89)
at com.automq.stream.s3.DeltaWALUploadTask.lambda$upload$2(DeltaWALUploadTask.java:116)
at java.base/java.util.concurrent.CompletableFuture$UniAccept.tryFire(CompletableFuture.java:718)
at java.base/java.util.concurrent.CompletableFuture$Completion.run(CompletableFuture.java:482)
at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1136)
at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:635)
at java.base/java.lang.Thread.run(Thread.java:833)
[2025-03-30 10:43:34,198] INFO [BrokerServer id=20406062] Transition from STARTING to STARTED (kafka.server.BrokerServer)
[2025-03-30 10:43:34,200] ERROR [BrokerServer id=20406062] Fatal error during broker startup. Prepare to shutdown (kafka.server.BrokerServer)