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