[apache#1472][part-6] followup: Fix Netty transport time when sending…

… shuffle data requests (apache#1551) ### What changes were proposed in this pull request? Refresh `timestamp` when sending `SendShuffleDataRequest`. ### Why are the changes needed? A follow-up PR for: apache#1534 ### Does this PR introduce _any_ user-facing change? No. ### How was this patch tested? Existing UTs.
zhengchenyu · Feb 29, 2024 · 9f4ccde · 9f4ccde
1 parent 37ec78b
commit 9f4ccde
Show file tree

Hide file tree

Showing 2 changed files with 6 additions and 2 deletions.
diff --git a/common/src/main/java/org/apache/uniffle/common/netty/protocol/SendShuffleDataRequest.java b/common/src/main/java/org/apache/uniffle/common/netty/protocol/SendShuffleDataRequest.java
@@ -141,4 +141,8 @@ public Map<Integer, List<ShuffleBlockInfo>> getPartitionToBlocks() {
   public long getTimestamp() {
     return timestamp;
   }
+
+  public void setTimestamp(long timestamp) {
+    this.timestamp = timestamp;
+  }
 }
diff --git a/...lient/src/main/java/org/apache/uniffle/client/impl/grpc/ShuffleServerGrpcNettyClient.java b/...lient/src/main/java/org/apache/uniffle/client/impl/grpc/ShuffleServerGrpcNettyClient.java
@@ -116,7 +116,7 @@ public RssSendShuffleDataResponse sendShuffleData(RssSendShuffleDataRequest requ
       try {
         RetryUtils.retryWithCondition(
             () -> {
-              TransportClient transportClient = getTransportClient();
+              final TransportClient transportClient = getTransportClient();
               long requireId =
                   requirePreAllocation(
                       request.getAppId(),
@@ -129,8 +129,8 @@ public RssSendShuffleDataResponse sendShuffleData(RssSendShuffleDataRequest requ
                         "requirePreAllocation failed! size[%s], host[%s], port[%s]",
                         allocateSize, host, port));
               }
-
               sendShuffleDataRequest.setRequireId(requireId);
+              sendShuffleDataRequest.setTimestamp(System.currentTimeMillis());
               long start = System.currentTimeMillis();
               RpcResponse rpcResponse =
                   transportClient.sendRpcSync(sendShuffleDataRequest, rpcTimeout);