cshuo commented on code in PR #20189:
URL: https://github.com/apache/hudi/pull/20189#discussion_r4217157774
##########
hudi-common/src/test/java/org/apache/hudi/common/util/TestRatelimiter.java:
##########
@@ -33,25 +41,75 @@ public class TestRatelimiter {
@Test
public void testRateLimiterWithNoThrottling() throws InterruptedException {
RateLimiter limiter = RateLimiter.create(1000, TimeUnit.SECONDS);
- long start = System.currentTimeMillis();
- assertEquals(true, limiter.tryAcquire(1000));
- // Sleep to represent some operation
- Thread.sleep(500);
- long end = System.currentTimeMillis();
- // With a large permit limit, there shouldn't be any throttling of
operations
- assertTrue((end - start) < TimeUnit.SECONDS.toMillis(2));
+ try {
+ long start = System.currentTimeMillis();
+ assertEquals(true, limiter.tryAcquire(1000));
+ // Sleep to represent some operation
+ Thread.sleep(500);
+ long end = System.currentTimeMillis();
+ // With a large permit limit, there shouldn't be any throttling of
operations
+ assertTrue((end - start) < TimeUnit.SECONDS.toMillis(2));
+ } finally {
+ limiter.stop();
+ }
}
@Test
public void testRateLimiterWithThrottling() throws InterruptedException {
RateLimiter limiter = RateLimiter.create(100, TimeUnit.SECONDS);
- long start = System.currentTimeMillis();
- assertEquals(true, limiter.tryAcquire(400));
- // Sleep to represent some operation
- Thread.sleep(500);
- long end = System.currentTimeMillis();
- // As size of operations is more than the maximum permits per second,
- // whole execution should be greater than 1 second
- assertTrue((end - start) >= TimeUnit.SECONDS.toMillis(2));
+ try {
+ long start = System.currentTimeMillis();
+ assertEquals(true, limiter.tryAcquire(400));
+ // Sleep to represent some operation
+ Thread.sleep(500);
+ long end = System.currentTimeMillis();
+ // As size of operations is more than the maximum permits per second,
+ // whole execution should be greater than 1 second
+ assertTrue((end - start) >= TimeUnit.SECONDS.toMillis(2));
+ } finally {
+ limiter.stop();
+ }
+ }
+
+ @Test
+ public void testCustomReleasePeriod() {
+ RateLimiter limiter = RateLimiter.create(1, 100, TimeUnit.MILLISECONDS);
+ try {
+ assertTrue(limiter.acquire(1));
+ long start = System.nanoTime();
+ assertTimeoutPreemptively(Duration.ofSeconds(1), () ->
assertTrue(limiter.acquire(1)));
+ assertTrue(System.nanoTime() - start >=
TimeUnit.MILLISECONDS.toNanos(50));
Review Comment:
Refills are timed from `create()`, not `acquire()`. If the first acquisition
occurs 70 ms after creation, the next refill takes only ~30 ms, so this
assertion fails despite correct behavior. Could we use a controllable scheduler
to avoid this timing assumption?
##########
hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/utils/TestFlinkRateLimiterFactory.java:
##########
@@ -0,0 +1,58 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hudi.sink.utils;
+
+import org.apache.hudi.common.util.RateLimiter;
+
+import org.junit.jupiter.api.Test;
+
+import java.time.Duration;
+
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTimeoutPreemptively;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+class TestFlinkRateLimiterFactory {
+
+ @Test
+ void testRateLimitLowerThanParallelismDoesNotCreateZeroPermitLimiter() {
+ RateLimiter limiter = FlinkRateLimiterFactory.create(1, 4);
+ try {
+ assertTimeoutPreemptively(Duration.ofSeconds(1), () ->
assertTrue(limiter.acquire(1)));
Review Comment:
This consumes only the initial permit, so an incorrect refill period of one
second instead of four would still pass. Could we consume another permit and
check the expected refill interval, ideally with a controllable scheduler?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
