Skip to content

fix: stabilize flaky ParamFlowDefaultCheckerTest by using mocked time #3626

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
EvanYao826 wants to merge 1 commit into
base: 1.8
Choose a base branch
from
Open

fix: stabilize flaky ParamFlowDefaultCheckerTest by using mocked time #3626

Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
108 changes: 51 additions & 57 deletions ...est/java/com/alibaba/csp/sentinel/slots/block/flow/param/ParamFlowDefaultCheckerTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -248,71 +248,65 @@ public void testParamFlowDefaultCheckQpsInDifferentDuration() throws Interrupted

@Test
public void testParamFlowDefaultCheckSingleValueCheckQpsMultipleThreads() throws Exception {
final String resourceName = "testParamFlowDefaultCheckSingleValueCheckQpsMultipleThreads";
final ResourceWrapper resourceWrapper = new StringResourceWrapper(resourceName, EntryType.IN);
int paramIdx = 0;

long threshold = 5L;

final ParamFlowRule rule = new ParamFlowRule();
rule.setResource(resourceName);
rule.setCount(threshold);
rule.setParamIdx(paramIdx);
rule.setDurationInSec(3);

final String valueA = "valueA";
ParameterMetric metric = new ParameterMetric();
ParameterMetricStorage.getMetricsMap().put(resourceWrapper.getName(), metric);
metric.getRuleTimeCounterMap().put(rule, new ConcurrentLinkedHashMapWrapper<Object, AtomicLong>(4000));
metric.getRuleTokenCounterMap().put(rule, new ConcurrentLinkedHashMapWrapper<>(4000));
int threadCount = 40;

final CountDownLatch waitLatch = new CountDownLatch(threadCount);
final AtomicInteger successCount = new AtomicInteger();
for (int i = 0; i < threadCount; i++) {
Thread t = new Thread(() -> {
if (ParamFlowChecker.passSingleValueCheck(resourceWrapper, rule, 1, valueA)) {
successCount.incrementAndGet();
}
waitLatch.countDown();
});
t.setName("sentinel-simulate-traffic-task-" + i);
t.start();
}
waitLatch.await();
try (MockedStatic<TimeUtil> mocked = super.mockTimeUtil()) {
final String resourceName = "testParamFlowDefaultCheckSingleValueCheckQpsMultipleThreads";
final ResourceWrapper resourceWrapper = new StringResourceWrapper(resourceName, EntryType.IN);
int paramIdx = 0;

long threshold = 5L;

final ParamFlowRule rule = new ParamFlowRule();
rule.setResource(resourceName);
rule.setCount(threshold);
rule.setParamIdx(paramIdx);
rule.setDurationInSec(3);

assertEquals(threshold, successCount.get());
successCount.set(0);
final String valueA = "valueA";
ParameterMetric metric = new ParameterMetric();
ParameterMetricStorage.getMetricsMap().put(resourceWrapper.getName(), metric);
metric.getRuleTimeCounterMap().put(rule, new ConcurrentLinkedHashMapWrapper<Object, AtomicLong>(4000));
metric.getRuleTokenCounterMap().put(rule, new ConcurrentLinkedHashMapWrapper<>(4000));
int threadCount = 40;

System.out.println("testParamFlowDefaultCheckSingleValueCheckQpsMultipleThreads: sleep for 3 seconds");
TimeUnit.SECONDS.sleep(3);
setCurrentMillis(mocked, System.currentTimeMillis());

successCount.set(0);
final CountDownLatch waitLatch1 = new CountDownLatch(threadCount);
final long currentTime = TimeUtil.currentTimeMillis();
final long endTime = currentTime + rule.getDurationInSec() * 1000 - 500;
for (int i = 0; i < threadCount; i++) {
Thread t = new Thread(() -> {
while (TimeUtil.currentTimeMillis() <= endTime) {
final CountDownLatch waitLatch = new CountDownLatch(threadCount);
final AtomicInteger successCount = new AtomicInteger();
for (int i = 0; i < threadCount; i++) {
Thread t = new Thread(() -> {
if (ParamFlowChecker.passSingleValueCheck(resourceWrapper, rule, 1, valueA)) {
successCount.incrementAndGet();
}

try {
TimeUnit.MILLISECONDS.sleep(ThreadLocalRandom.current().nextInt(20));
} catch (InterruptedException e) {
e.printStackTrace();
waitLatch.countDown();
});
t.setName("sentinel-simulate-traffic-task-" + i);
t.start();
}
waitLatch.await();

assertEquals(threshold, successCount.get());
successCount.set(0);

// Advance time past the duration window to reset counters
sleep(mocked, rule.getDurationInSec() * 1000);

@oss-sentinel-ai oss-sentinel-ai Sep 3, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

MockedStatic scopes stubbing to the thread that created it, so the 40 worker threads still call the real TimeUtil.currentTimeMillis() inside ParamFlowChecker. Advancing this mock therefore does not advance their token-counter timestamps; the second round runs less than three real seconds after the first and is rejected rather than allowing threshold requests. Use a time source that is shared by the worker threads (or install the static mock within each worker) before removing the real wait.


// Second round: 40 threads each call once, should again allow exactly threshold
successCount.set(0);
final CountDownLatch waitLatch1 = new CountDownLatch(threadCount);
for (int i = 0; i < threadCount; i++) {
Thread t = new Thread(() -> {
if (ParamFlowChecker.passSingleValueCheck(resourceWrapper, rule, 1, valueA)) {
successCount.incrementAndGet();
}
}

waitLatch1.countDown();
});
t.setName("sentinel-simulate-traffic-task-" + i);
t.start();
waitLatch1.countDown();
});
t.setName("sentinel-simulate-traffic-task-" + i);
t.start();
}
waitLatch1.await();

assertEquals(threshold, successCount.get());
}
waitLatch1.await();

assertEquals(threshold, successCount.get());
}

@Before
Expand Down