Github user vanzin commented on a diff in the pull request:
https://github.com/apache/spark/pull/21346#discussion_r195590730
--- Diff:
common/network-common/src/test/java/org/apache/spark/network/RpcIntegrationSuite.java
---
@@ -130,6 +200,60 @@ public void onFailure(Throwable e) {
return res;
}
+ private RpcResult sendRpcWithStream(String... streams) throws Exception {
+ TransportClient client =
clientFactory.createClient(TestUtils.getLocalHost(), server.getPort());
+ final Semaphore sem = new Semaphore(0);
+ RpcResult res = new RpcResult();
+ res.successMessages = Collections.synchronizedSet(new
HashSet<String>());
+ res.errorMessages = Collections.synchronizedSet(new HashSet<String>());
+
+ for (String stream : streams) {
+ int idx = stream.lastIndexOf('/');
+ ManagedBuffer meta = new
NioManagedBuffer(JavaUtils.stringToBytes(stream));
+ String streamName = (idx == -1) ? stream : stream.substring(idx + 1);
+ ManagedBuffer data = testData.openStream(conf, streamName);
+ client.uploadStream(meta, data, new RpcStreamCallback(stream, res,
sem));
+ }
+ streamCallbacks.values().forEach(streamCallback -> {
--- End diff --
I'm trying to follow the logic here...
- in L215, client sends a stream to the remote
- in L82, remote receives the stream and registers the callback
- here, you wait for the callbacks to finish in the order they were
registered.
Isn't there a race between steps 2 and 3, as in you might miss one or more
callback registration?
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
