From 465c665db1dc65e3b02c584cf7f8d06b24909b0c Mon Sep 17 00:00:00 2001 From: Shixiong Zhu Date: Thu, 3 Mar 2016 22:53:07 -0800 Subject: [SPARK-13652][CORE] Copy ByteBuffer in sendRpcSync as it will be recycled ## What changes were proposed in this pull request? `sendRpcSync` should copy the response content because the underlying buffer will be recycled and reused. ## How was this patch tested? Jenkins unit tests. Author: Shixiong Zhu Closes #11499 from zsxwing/SPARK-13652. --- .../java/org/apache/spark/network/client/RpcResponseCallback.java | 7 ++++++- .../main/java/org/apache/spark/network/client/TransportClient.java | 6 +++++- 2 files changed, 11 insertions(+), 2 deletions(-) diff --git a/common/network-common/src/main/java/org/apache/spark/network/client/RpcResponseCallback.java b/common/network-common/src/main/java/org/apache/spark/network/client/RpcResponseCallback.java index 47e93f9846..6afc63f71b 100644 --- a/common/network-common/src/main/java/org/apache/spark/network/client/RpcResponseCallback.java +++ b/common/network-common/src/main/java/org/apache/spark/network/client/RpcResponseCallback.java @@ -24,7 +24,12 @@ import java.nio.ByteBuffer; * failure. */ public interface RpcResponseCallback { - /** Successful serialized result from server. */ + /** + * Successful serialized result from server. + * + * After `onSuccess` returns, `response` will be recycled and its content will become invalid. + * Please copy the content of `response` if you want to use it after `onSuccess` returns. + */ void onSuccess(ByteBuffer response); /** Exception either propagated from server or raised on client side. */ diff --git a/common/network-common/src/main/java/org/apache/spark/network/client/TransportClient.java b/common/network-common/src/main/java/org/apache/spark/network/client/TransportClient.java index e15f096d36..64a83171e9 100644 --- a/common/network-common/src/main/java/org/apache/spark/network/client/TransportClient.java +++ b/common/network-common/src/main/java/org/apache/spark/network/client/TransportClient.java @@ -257,7 +257,11 @@ public class TransportClient implements Closeable { sendRpc(message, new RpcResponseCallback() { @Override public void onSuccess(ByteBuffer response) { - result.set(response); + ByteBuffer copy = ByteBuffer.allocate(response.remaining()); + copy.put(response); + // flip "copy" to make it readable + copy.flip(); + result.set(copy); } @Override -- cgit v1.2.3