fix(s3stream): remove unnecessary config check (#2084)

Signed-off-by: Shichao Nie <[email protected]>

Author: SCNieh

s3stream/src/main/java/com/automq/stream/s3/operator/AbstractObjectStorage.java

@@ -122,7 +122,6 @@ protected AbstractObjectStorage(
         if (!manualMergeRead) {
             scheduler.scheduleWithFixedDelay(this::tryMergeRead, 1, 1, TimeUnit.MILLISECONDS);
         }
-        checkConfig();
         S3StreamMetricsManager.registerInflightS3ReadQuotaSupplier(inflightReadLimiter::availablePermits, currentIndex);
         S3StreamMetricsManager.registerInflightS3WriteQuotaSupplier(inflightWriteLimiter::availablePermits, currentIndex);
 
@@ -663,15 +662,6 @@ static int getMaxObjectStorageConcurrency() {
         return Math.max(MIN_CONCURRENCY, Math.min(cpuCores * DEFAULT_CONCURRENCY_PER_CORE, MAX_CONCURRENCY));
     }
 
-    private void checkConfig() {
-        if (this.networkInboundBandwidthLimiter != null) {
-            if (this.networkInboundBandwidthLimiter.getMaxTokens() < Writer.MIN_PART_SIZE) {
-                throw new IllegalArgumentException(String.format("Network inbound burst bandwidth limit %d must be no less than min part size %d",
-                    this.networkInboundBandwidthLimiter.getMaxTokens(), Writer.MIN_PART_SIZE));
-            }
-        }
-    }
-
     /**
      * Acquire read permit, permit will auto release when cf complete.
      *

s3stream/src/test/java/com/automq/stream/s3/AsyncNetworkBandwidthLimiterTest.java

@@ -152,4 +152,18 @@ public void testThrottleConsumeWithPriority1() {
         result.join();
         cf2.join();
     }
+
+    @Test
+    @Timeout(10)
+    public void testThrottleConsumeWithLargeChunk() {
+        AsyncNetworkBandwidthLimiter bucket = new AsyncNetworkBandwidthLimiter(AsyncNetworkBandwidthLimiter.Type.INBOUND, 1024 * 1024, 100);
+        CompletableFuture<Void> bypassCf = bucket.consume(ThrottleStrategy.BYPASS, 1024 * 1024);
+        CompletableFuture<Void> cf = bucket.consume(ThrottleStrategy.CATCH_UP, 10 * 1024 * 1024);
+        CompletableFuture<Void> result = cf.whenComplete((v, e) -> {
+            Assertions.assertNull(e);
+            Assertions.assertEquals(0, bucket.getAvailableTokens());
+            Assertions.assertTrue(bypassCf.isDone());
+        });
+        result.join();
+    }
 }