Github user squito commented on a diff in the pull request:
https://github.com/apache/spark/pull/21346#discussion_r195795803
--- 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 --
oh great point, thanks for catching that. I will move this after the
semaphore check, that will ensure that everything has been added to
`streamCallbacks`
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
