wuchong commented on code in PR #2926:
URL: https://github.com/apache/fluss/pull/2926#discussion_r2985622357
##########
fluss-server/src/test/java/org/apache/fluss/server/replica/fetcher/ReplicaFetcherThreadTest.java:
##########
@@ -374,6 +376,83 @@ tb, genMemoryLogRecordsWithWriterId(DATA1, writerId, 2,
0)),
() ->
assertThat(followerReplica.getLocalLogEndOffset()).isEqualTo(30L));
}
+ @Test
+ void testFetchTimeoutReleasesPooledByteBuf() throws Exception {
Review Comment:
I tried to reproduce the problem. But this test successfully passed, when I
revert the changes of `ReplicaFetcherThread`.
##########
fluss-server/src/test/java/org/apache/fluss/server/replica/fetcher/TestingLeaderEndpoint.java:
##########
@@ -51,6 +55,21 @@ public class TestingLeaderEndpoint implements LeaderEndpoint
{
/** The max fetch size for a bucket in bytes. */
private final int maxFetchSizeForBucket;
+ /**
+ * If set, fetchLog will return a future that completes after the
specified delay, simulating a
+ * slow leader. The response will carry a pooled ByteBuf to track buffer
release.
+ */
+ private volatile ScheduledExecutorService delayExecutor;
+
+ private volatile long delayMs;
+
+ /** The last ByteBuf allocated for a delayed fetch response, for leak
detection in tests. */
+ private volatile ByteBuf lastAllocatedByteBuf;
Review Comment:
This is not used, can be removed.
--
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]
