Skip to content

MINOR: fix flaky testRecordThreadIdleRatio #15987

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

Merged
dajac merged 3 commits into from
May 23, 2024
Merged

MINOR: fix flaky testRecordThreadIdleRatio #15987

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
d165e04
fix
jeffkbkim May 17, 2024
5a34639
comment
jeffkbkim May 17, 2024
044319a
address comments
jeffkbkim May 20, 2024
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
3 changes: 0 additions & 3 deletions ...src/main/java/org/apache/kafka/coordinator/group/runtime/MultiThreadedEventProcessor.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -112,9 +112,6 @@ public MultiThreadedEventProcessor(
*/
private class EventProcessorThread extends Thread {
private final Logger log;
private long pollStartMs;
private long timeSinceLastPollMs;
private long lastPollMs;

EventProcessorThread(
String name
Expand Down
9 changes: 5 additions & 4 deletions ...test/java/org/apache/kafka/coordinator/group/runtime/MultiThreadedEventProcessorTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -61,8 +61,9 @@ public DelayEventAccumulator(Time time, long takeDelayMs) {

@Override
public CoordinatorEvent take() {
CoordinatorEvent event = super.take();

@gaurav-narula gaurav-narula May 17, 2024

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

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

IIUC, the fix suggests the flakyness is because super.take() returns null spuriously. I don't quite follow why that would happen though. Javadoc for EventAccumulator::take mentions it returns null only when the accumulator is closed. We also assert the value captured within doAnswer is 100, recordedIdleTimesMs.size() == 8 and mockRuntimeMetrics.recordThreadIdleTime() is invoked 8 times so where is the extra invocation coming from?

@jeffkbkim jeffkbkim May 17, 2024

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

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

That's correct and was my initial confusion as well. The event processor shuts down at the end of the test so we have the processor thread still running when we perform the assertions.

When we call

long diff = time.milliseconds() - startMs;

we may be at the point when we process the 8 events (diff == 800ms) or when the thread does another iteration over handleEvents(), which does another time.sleep() (diff == 900ms). This was causing the flakiness.

@gaurav-narula gaurav-narula May 17, 2024

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

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

we may be at the point when we process the 8 events (diff == 800ms) or when the thread does another iteration over handleEvents(), which does another time.sleep() (diff == 900ms). This was causing the flakiness.

Sorry, I'm still confused. Why would another iteration over handleEvents() invoke the doAnswer? In other words, shouldn't accumulator.take() block since there are no more events? I'd imagine EventAccumulator::take to block on condition.await() after 8 events have been processed.

@gaurav-narula gaurav-narula May 17, 2024 •
edited
Loading

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

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

Nvm, I see your point - accumulator.take() may be invoked - the block on the delegate may happen after the timer has been incremented.

@gaurav-narula gaurav-narula May 17, 2024 •
edited
Loading

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

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

IIUC, it's the ordering that matters so we can avoid the null check and increment time only after super.take() returns.

            CoordinatorEvent event = super.take(); // blocks until there are new events
            time.sleep(takeDelayMs); // only increment timer after we get unblocked
            return event;

@jeffkbkim jeffkbkim May 20, 2024

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

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

@gaurav-narula thanks for the feedback. that makes sense, i have incorporated your suggestion

time.sleep(takeDelayMs);
return super.take();
return event;
}
}

Expand Down Expand Up @@ -475,9 +476,9 @@ public void testRecordThreadIdleRatio() throws Exception {
doAnswer(invocation -> {
long threadIdleTime = idleTimeCaptured.getValue();
assertEquals(100, threadIdleTime);
synchronized (recordedIdleTimesMs) {
recordedIdleTimesMs.add(threadIdleTime);
}

// No synchronization required as the test uses a single event processor thread.
recordedIdleTimesMs.add(threadIdleTime);

@gaurav-narula gaurav-narula May 17, 2024

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

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

Nit: might be useful to add a comment that this is safe because numThreads in line 470 is 1.

return null;
}).when(mockRuntimeMetrics).recordThreadIdleTime(idleTimeCaptured.capture());

Expand Down