Copilot commented on code in PR #3842:
URL: https://github.com/apache/celeborn/pull/3842#discussion_r3954216858
##########
common/src/test/java/org/apache/celeborn/common/network/TransportClientFactorySuiteJ.java:
##########
@@ -184,6 +194,83 @@ public void neverReturnInactiveClients() throws
IOException, InterruptedExceptio
factory.close();
}
+ @Test
+ public void replaceClientAfterPushTimeout() throws Exception {
+ AtomicBoolean respond = new AtomicBoolean(false);
+ BaseMessageHandler handler =
+ new BaseMessageHandler() {
+ @Override
+ public void receive(TransportClient client, RequestMessage message) {
+ PushData pushData = (PushData) message;
+ if (respond.get()) {
+ client
+ .getChannel()
+ .writeAndFlush(
+ new RpcResponse(
+ pushData.requestId, new
NioManagedBuffer(ByteBuffer.allocate(0))));
+ }
+ }
+
+ @Override
+ public boolean checkRegistered() {
+ return true;
+ }
+ };
+ TransportConf conf =
+ new TransportConf(TransportModuleConstants.REPLICATE_MODULE, new
CelebornConf());
+
+ try (TransportContext timeoutContext = new TransportContext(conf, handler);
+ TransportServer timeoutServer = timeoutContext.createServer();
+ TransportClientFactory factory = timeoutContext.createClientFactory())
{
+ TransportClient timedOutClient =
+ factory.createClient(getLocalHost(), timeoutServer.getPort());
+ RpcResponseCallback timedOutCallback = mock(RpcResponseCallback.class);
+ RpcResponseCallback outstandingCallback =
mock(RpcResponseCallback.class);
+ timedOutClient.pushData(
+ new PushData(
+ (byte) 0,
+ "shuffleKey",
+ "partitionId-1",
+ new NioManagedBuffer(ByteBuffer.allocate(1))),
+ 0,
+ timedOutCallback);
+ timedOutClient.pushData(
+ new PushData(
+ (byte) 0,
+ "shuffleKey",
+ "partitionId-2",
+ new NioManagedBuffer(ByteBuffer.allocate(1))),
+ TimeUnit.MINUTES.toMillis(1),
+ outstandingCallback);
+
+ timedOutClient.getHandler().failExpiredPushRequest();
+
+ Mockito.verify(timedOutCallback, Mockito.after(200).times(1))
+ .onFailure(
+ Mockito.argThat(
+ error ->
+
error.getMessage().startsWith(StatusCode.PUSH_DATA_TIMEOUT_REPLICA.name())));
+ Mockito.verify(outstandingCallback,
Mockito.after(200).times(1)).onFailure(Mockito.any());
+ assertFalse(timedOutClient.isActive());
Review Comment:
这里对异步回调的校验使用 Mockito.after(200)，200ms 在
CI/负载较高环境下可能偏短，容易导致用例偶发超时失败（flaky）。建议改用
Mockito.timeout(5000)（或至少显著增大等待时间）来保证稳定性。
--
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]
