Copilot commented on code in PR #3842:
URL: https://github.com/apache/celeborn/pull/3842#discussion_r3967203622
##########
common/src/main/java/org/apache/celeborn/common/network/client/TransportClient.java:
##########
@@ -350,6 +350,11 @@ public void timeOut() {
this.timedOut = true;
}
+ private void invalidate() {
+ this.timedOut = true;
+ channel.close();
+ }
Review Comment:
This duplicates state-setting behavior already present in `timeOut()` (also
sets `timedOut = true`). To reduce divergence risk, consider reusing the
existing method (e.g., call `timeOut()` from `invalidate()`), keeping
`invalidate()` focused on closing the channel.
##########
common/src/test/java/org/apache/celeborn/common/network/TransportClientFactorySuiteJ.java:
##########
@@ -184,6 +194,82 @@ 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();
Review Comment:
The test relies on a push timeout value of `0` to make the first request
immediately expired. If `failExpiredPushRequest()` uses a strict `now >
dueTime` comparison (common pattern), a `0` ms timeout can be non-deterministic
within the same millisecond tick and make the test flaky. Make the request
deterministically expired (e.g., use a small positive timeout and wait until it
elapses, or otherwise ensure `dueTime` is in the past before calling
`failExpiredPushRequest()`).
--
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]
