You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@asterixdb.apache.org by AsterixDB Code Review <do...@asterix-gerrit.ics.uci.edu> on 2024/02/19 03:32:33 UTC
Change in asterixdb[trinity]: [NO ISSUE][HYR][NET] Use executor service for async handshakes
From Michael Blow <mb...@apache.org>:
Michael Blow has uploaded this change for review. ( https://asterix-gerrit.ics.uci.edu/c/asterixdb/+/18168 )
Change subject: [NO ISSUE][HYR][NET] Use executor service for async handshakes
......................................................................
[NO ISSUE][HYR][NET] Use executor service for async handshakes
Change-Id: Ibe2fdce979a49d962517494dc77a5616a19f6abf
---
M hyracks-fullstack/hyracks/hyracks-ipc/src/main/java/org/apache/hyracks/ipc/impl/IPCSystem.java
M asterixdb/asterix-app/src/main/java/org/apache/asterix/app/nc/NCAppRuntimeContext.java
M hyracks-fullstack/hyracks/hyracks-client/src/main/java/org/apache/hyracks/client/result/ResultDirectory.java
M hyracks-fullstack/hyracks/hyracks-control/hyracks-control-nc/src/main/java/org/apache/hyracks/control/nc/net/ResultNetworkManager.java
M hyracks-fullstack/hyracks/hyracks-hdfs/src/main/java/org/apache/hyracks/hdfs2/scheduler/Scheduler.java
M asterixdb/asterix-app/src/main/java/org/apache/asterix/app/result/ResultReader.java
M asterixdb/asterix-app/src/main/java/org/apache/asterix/hyracks/bootstrap/CCApplication.java
M asterixdb/asterix-app/src/test/java/org/apache/asterix/api/common/AsterixHyracksIntegrationUtil.java
M hyracks-fullstack/hyracks/hyracks-ipc/src/main/java/org/apache/hyracks/ipc/impl/HyracksConnection.java
M asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/util/HDFSUtils.java
M hyracks-fullstack/hyracks/hyracks-control/hyracks-control-nc/src/main/java/org/apache/hyracks/control/nc/NodeControllerService.java
M hyracks-fullstack/hyracks/hyracks-examples/hyracks-integration-tests/src/test/java/org/apache/hyracks/tests/integration/AbstractMultiNCIntegrationTest.java
M hyracks-fullstack/hyracks/hyracks-ipc/src/test/java/org/apache/hyracks/ipc/tests/IPCTest.java
M hyracks-fullstack/hyracks/hyracks-control/hyracks-control-nc/src/main/java/org/apache/hyracks/control/nc/net/MessagingNetworkManager.java
M hyracks-fullstack/hyracks/hyracks-control/hyracks-control-nc/src/main/java/org/apache/hyracks/control/nc/net/NetworkManager.java
M hyracks-fullstack/hyracks/hyracks-net/src/test/java/org/apache/hyracks/net/tests/NetTest.java
M hyracks-fullstack/hyracks/hyracks-control/hyracks-control-cc/src/main/java/org/apache/hyracks/control/cc/ClusterControllerService.java
M hyracks-fullstack/hyracks/hyracks-hdfs/src/main/java/org/apache/hyracks/hdfs/scheduler/Scheduler.java
M hyracks-fullstack/hyracks/hyracks-net/src/main/java/org/apache/hyracks/net/protocols/tcp/TCPEndpoint.java
M asterixdb/asterix-app/src/main/java/org/apache/asterix/app/cc/CcApplicationContext.java
M asterixdb/asterix-app/src/test/java/org/apache/asterix/test/common/TestExecutor.java
M hyracks-fullstack/hyracks/hyracks-client/src/main/java/org/apache/hyracks/client/net/ClientNetworkManager.java
M hyracks-fullstack/hyracks/hyracks-client/src/main/java/org/apache/hyracks/client/result/ResultSet.java
M hyracks-fullstack/hyracks/hyracks-net/src/main/java/org/apache/hyracks/net/protocols/muxdemux/MuxDemux.java
M asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/library/ExternalLibraryManager.java
M hyracks-fullstack/hyracks/hyracks-ipc/src/main/java/org/apache/hyracks/ipc/impl/IPCConnectionManager.java
M hyracks-fullstack/hyracks/hyracks-examples/hyracks-integration-tests/src/test/java/org/apache/hyracks/tests/integration/AbstractIntegrationTest.java
27 files changed, 127 insertions(+), 83 deletions(-)
git pull ssh://asterix-gerrit.ics.uci.edu:29418/asterixdb refs/changes/68/18168/1
diff --git a/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/cc/CcApplicationContext.java b/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/cc/CcApplicationContext.java
index 880880e..ade440b 100644
--- a/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/cc/CcApplicationContext.java
+++ b/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/cc/CcApplicationContext.java
@@ -209,7 +209,8 @@
NetworkUtil.closeQuietly(rs);
NetworkUtil.closeQuietly(hc);
- hcc = hc = new HyracksConnection(hcc.getHost(), hcc.getPort());
+ hcc = hc = new HyracksConnection(hcc.getHost(), hcc.getPort(),
+ ccServiceCtx.getControllerService().getExecutor());
} catch (Exception e) {
throw HyracksDataException.create(e);
}
diff --git a/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/nc/NCAppRuntimeContext.java b/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/nc/NCAppRuntimeContext.java
index 4d32967..be4242f 100644
--- a/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/nc/NCAppRuntimeContext.java
+++ b/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/nc/NCAppRuntimeContext.java
@@ -532,7 +532,7 @@
ClusterControllerInfo ccInfo = ncSrv.getNodeParameters(primaryCcId).getClusterControllerInfo();
NetworkUtil.closeQuietly(hc);
hcc = hc = new HyracksConnection(ccInfo.getClientNetAddress(), ccInfo.getClientNetPort(),
- ncSrv.getNetworkSecurityManager().getSocketChannelFactory());
+ ncSrv.getNetworkSecurityManager().getSocketChannelFactory(), getThreadExecutor());
} catch (Exception e) {
throw HyracksDataException.create(e);
}
diff --git a/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/result/ResultReader.java b/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/result/ResultReader.java
index 56dfd5e..c63c9de 100644
--- a/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/result/ResultReader.java
+++ b/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/result/ResultReader.java
@@ -64,7 +64,7 @@
public static ResultSet createResultSet(IHyracksClientConnection hcc, IControllerService srv,
CompilerProperties compilerProperties) throws Exception {
- return new ResultSet(hcc, srv.getNetworkSecurityManager().getSocketChannelFactory(),
+ return new ResultSet(srv.getExecutor(), hcc, srv.getNetworkSecurityManager().getSocketChannelFactory(),
compilerProperties.getFrameSize(), ResultReader.NUM_READERS);
}
}
diff --git a/asterixdb/asterix-app/src/main/java/org/apache/asterix/hyracks/bootstrap/CCApplication.java b/asterixdb/asterix-app/src/main/java/org/apache/asterix/hyracks/bootstrap/CCApplication.java
index bc67823..98fa125 100644
--- a/asterixdb/asterix-app/src/main/java/org/apache/asterix/hyracks/bootstrap/CCApplication.java
+++ b/asterixdb/asterix-app/src/main/java/org/apache/asterix/hyracks/bootstrap/CCApplication.java
@@ -151,7 +151,8 @@
String strIP = ccServiceCtx.getCCContext().getClusterControllerInfo().getClientNetAddress();
int port = ccServiceCtx.getCCContext().getClusterControllerInfo().getClientNetPort();
hcc = new HyracksConnection(strIP, port,
- ccServiceCtx.getControllerService().getNetworkSecurityManager().getSocketChannelFactory());
+ ccServiceCtx.getControllerService().getNetworkSecurityManager().getSocketChannelFactory(),
+ ccServiceCtx.getControllerService().getExecutor());
MetadataBuiltinFunctions.init();
ReplicationProperties repProp =
new ReplicationProperties(PropertiesAccessor.getInstance(ccServiceCtx.getAppConfig()));
diff --git a/asterixdb/asterix-app/src/test/java/org/apache/asterix/api/common/AsterixHyracksIntegrationUtil.java b/asterixdb/asterix-app/src/test/java/org/apache/asterix/api/common/AsterixHyracksIntegrationUtil.java
index 1e096ac..f99d663 100644
--- a/asterixdb/asterix-app/src/test/java/org/apache/asterix/api/common/AsterixHyracksIntegrationUtil.java
+++ b/asterixdb/asterix-app/src/test/java/org/apache/asterix/api/common/AsterixHyracksIntegrationUtil.java
@@ -182,7 +182,7 @@
// Wait until cluster becomes active
((ICcApplicationContext) cc.getApplicationContext()).getClusterStateManager().waitForState(ClusterState.ACTIVE);
hcc = new HyracksConnection(cc.getConfig().getClientListenAddress(), cc.getConfig().getClientListenPort(),
- cc.getNetworkSecurityManager().getSocketChannelFactory());
+ cc.getNetworkSecurityManager().getSocketChannelFactory(), cc.getExecutor());
this.ncs = nodeControllers.toArray(new NodeControllerService[nodeControllers.size()]);
}
diff --git a/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/common/TestExecutor.java b/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/common/TestExecutor.java
index 04da930..ce66d17 100644
--- a/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/common/TestExecutor.java
+++ b/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/common/TestExecutor.java
@@ -2157,7 +2157,7 @@
Assert.fail("No test files found for test: " + testCaseCtx.getTestCase().getFilePath() + "/"
+ cUnit.getName());
}
- List<TestFileContext> expectedResultFileCtxs = testCaseCtx.getExpectedResultFiles(cUnit);
+ List<TestFileContext> expectedResultFileCtxs = getExpectedResultFiles(testCaseCtx, cUnit);
int[] savedQueryCounts = new int[numOfFiles + testFileCtxs.size()];
loopIteration = 0;
for (ListIterator<TestFileContext> iter = testFileCtxs.listIterator(); iter.hasNext();) {
@@ -2218,6 +2218,11 @@
}
}
+ protected List<TestFileContext> getExpectedResultFiles(TestCaseContext testCaseCtx, CompilationUnit cUnit)
+ throws Exception {
+ return testCaseCtx.getExpectedResultFiles(cUnit);
+ }
+
private String applySubstitution(String statement, List<Parameter> parameters) throws Exception {
// Ensure all macro parameters are available
Parameter startParameter = parameters.stream()
diff --git a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/library/ExternalLibraryManager.java b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/library/ExternalLibraryManager.java
index 050d9f0..29aeda7 100755
--- a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/library/ExternalLibraryManager.java
+++ b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/library/ExternalLibraryManager.java
@@ -162,7 +162,7 @@
public void initialize(boolean resetStorageData) throws HyracksDataException {
try {
- pythonIPC = new IPCSystem(new InetSocketAddress(InetAddress.getLoopbackAddress(), 0),
+ pythonIPC = new IPCSystem(ncs.getExecutor(), new InetSocketAddress(InetAddress.getLoopbackAddress(), 0),
PlainSocketChannelFactory.INSTANCE, router, new ExternalFunctionResultRouter.NoOpNoSerJustDe());
pythonIPC.start();
Path baseDirPath = baseDir.getFile().toPath();
diff --git a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/util/HDFSUtils.java b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/util/HDFSUtils.java
index f4edafc..25926c9 100644
--- a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/util/HDFSUtils.java
+++ b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/util/HDFSUtils.java
@@ -78,7 +78,7 @@
try {
scheduler = new Scheduler(ccContext.getClusterControllerInfo().getClientNetAddress(),
ccContext.getClusterControllerInfo().getClientNetPort(),
- networkSecurityManager.getSocketChannelFactory());
+ networkSecurityManager.getSocketChannelFactory(), serviceCtx.getControllerService().getExecutor());
} catch (HyracksException e) {
throw new RuntimeDataException(ErrorCode.UTIL_HDFS_UTILS_CANNOT_OBTAIN_HDFS_SCHEDULER);
}
diff --git a/hyracks-fullstack/hyracks/hyracks-client/src/main/java/org/apache/hyracks/client/net/ClientNetworkManager.java b/hyracks-fullstack/hyracks/hyracks-client/src/main/java/org/apache/hyracks/client/net/ClientNetworkManager.java
index 5fe6ecb..94e83c5 100644
--- a/hyracks-fullstack/hyracks/hyracks-client/src/main/java/org/apache/hyracks/client/net/ClientNetworkManager.java
+++ b/hyracks-fullstack/hyracks/hyracks-client/src/main/java/org/apache/hyracks/client/net/ClientNetworkManager.java
@@ -21,6 +21,7 @@
import java.io.IOException;
import java.net.InetSocketAddress;
import java.net.SocketAddress;
+import java.util.concurrent.ExecutorService;
import org.apache.hyracks.api.exceptions.NetException;
import org.apache.hyracks.api.network.ISocketChannelFactory;
@@ -36,12 +37,12 @@
private final MuxDemux md;
- public ClientNetworkManager(int nThreads, ISocketChannelFactory socketChannelFactory) {
+ public ClientNetworkManager(int nThreads, ISocketChannelFactory socketChannelFactory, ExecutorService executor) {
/* This is a connect only socket and does not listen to any incoming connections, so pass null to
* localAddress and listener.
*/
md = new MuxDemux(null, null, nThreads, MAX_CONNECTION_ATTEMPTS, FullFrameChannelInterfaceFactory.INSTANCE,
- socketChannelFactory);
+ socketChannelFactory, executor);
}
public void start() throws IOException {
diff --git a/hyracks-fullstack/hyracks/hyracks-client/src/main/java/org/apache/hyracks/client/result/ResultDirectory.java b/hyracks-fullstack/hyracks/hyracks-client/src/main/java/org/apache/hyracks/client/result/ResultDirectory.java
index 5dd5fff..bfb1da0 100644
--- a/hyracks-fullstack/hyracks/hyracks-client/src/main/java/org/apache/hyracks/client/result/ResultDirectory.java
+++ b/hyracks-fullstack/hyracks/hyracks-client/src/main/java/org/apache/hyracks/client/result/ResultDirectory.java
@@ -21,6 +21,7 @@
import java.io.Closeable;
import java.io.IOException;
import java.net.InetSocketAddress;
+import java.util.concurrent.ExecutorService;
import org.apache.hyracks.api.job.JobId;
import org.apache.hyracks.api.network.ISocketChannelFactory;
@@ -40,10 +41,10 @@
private final IPCSystem ipc;
private final IResultDirectory remoteResultDirectory;
- public ResultDirectory(String resultHost, int resultPort, ISocketChannelFactory socketChannelFactory)
- throws IOException, IPCException {
+ public ResultDirectory(String resultHost, int resultPort, ISocketChannelFactory socketChannelFactory,
+ ExecutorService executor) throws IOException, IPCException {
RPCInterface rpci = new RPCInterface();
- ipc = new IPCSystem(new InetSocketAddress(0), socketChannelFactory, rpci,
+ ipc = new IPCSystem(executor, new InetSocketAddress(0), socketChannelFactory, rpci,
new JavaSerializationBasedPayloadSerializerDeserializer());
ipc.start();
IIPCHandle ddsIpchandle = ipc.getReconnectingHandle(new InetSocketAddress(resultHost, resultPort));
diff --git a/hyracks-fullstack/hyracks/hyracks-client/src/main/java/org/apache/hyracks/client/result/ResultSet.java b/hyracks-fullstack/hyracks/hyracks-client/src/main/java/org/apache/hyracks/client/result/ResultSet.java
index 8a88045..d52b8c3 100644
--- a/hyracks-fullstack/hyracks/hyracks-client/src/main/java/org/apache/hyracks/client/result/ResultSet.java
+++ b/hyracks-fullstack/hyracks/hyracks-client/src/main/java/org/apache/hyracks/client/result/ResultSet.java
@@ -20,6 +20,7 @@
import java.io.Closeable;
import java.io.IOException;
+import java.util.concurrent.ExecutorService;
import org.apache.hyracks.api.client.IHyracksClientConnection;
import org.apache.hyracks.api.comm.NetworkAddress;
@@ -42,12 +43,13 @@
private final IHyracksCommonContext resultClientCtx;
- public ResultSet(IHyracksClientConnection hcc, ISocketChannelFactory socketChannelFactory, int frameSize,
- int nReaders) throws Exception {
+ public ResultSet(ExecutorService executor, IHyracksClientConnection hcc, ISocketChannelFactory socketChannelFactory,
+ int frameSize, int nReaders) throws Exception {
NetworkAddress ddsAddress = hcc.getResultDirectoryAddress();
- resultDirectory = new ResultDirectory(ddsAddress.getAddress(), ddsAddress.getPort(), socketChannelFactory);
+ resultDirectory =
+ new ResultDirectory(ddsAddress.getAddress(), ddsAddress.getPort(), socketChannelFactory, executor);
- netManager = new ClientNetworkManager(nReaders, socketChannelFactory);
+ netManager = new ClientNetworkManager(nReaders, socketChannelFactory, executor);
netManager.start();
resultClientCtx = new ResultClientContext(frameSize);
diff --git a/hyracks-fullstack/hyracks/hyracks-control/hyracks-control-cc/src/main/java/org/apache/hyracks/control/cc/ClusterControllerService.java b/hyracks-fullstack/hyracks/hyracks-control/hyracks-control-cc/src/main/java/org/apache/hyracks/control/cc/ClusterControllerService.java
index d6698fe..6e76dca 100644
--- a/hyracks-fullstack/hyracks/hyracks-control/hyracks-control-cc/src/main/java/org/apache/hyracks/control/cc/ClusterControllerService.java
+++ b/hyracks-fullstack/hyracks/hyracks-control/hyracks-control-cc/src/main/java/org/apache/hyracks/control/cc/ClusterControllerService.java
@@ -217,14 +217,14 @@
LOGGER.log(Level.INFO, "Starting ClusterControllerService: " + this);
serverCtx = new ServerContext(ServerContext.ServerType.CLUSTER_CONTROLLER, new File(ccConfig.getRootDir()));
IIPCI ccIPCI = new ClusterControllerIPCI(this);
- clusterIPC = new IPCSystem(
+ clusterIPC = new IPCSystem(executor,
new InetSocketAddress(ccConfig.getClusterListenAddress(), ccConfig.getClusterListenPort()),
networkSecurityManager.getSocketChannelFactory(), ccIPCI, new CCNCFunctions.SerializerDeserializer());
IIPCI ciIPCI = new ClientInterfaceIPCI(this, jobIdFactory);
- clientIPC =
- new IPCSystem(new InetSocketAddress(ccConfig.getClientListenAddress(), ccConfig.getClientListenPort()),
- networkSecurityManager.getSocketChannelFactory(), ciIPCI,
- new JavaSerializationBasedPayloadSerializerDeserializer());
+ clientIPC = new IPCSystem(executor,
+ new InetSocketAddress(ccConfig.getClientListenAddress(), ccConfig.getClientListenPort()),
+ networkSecurityManager.getSocketChannelFactory(), ciIPCI,
+ new JavaSerializationBasedPayloadSerializerDeserializer());
clusterIPC.start();
clientIPC.start();
startWebServers();
diff --git a/hyracks-fullstack/hyracks/hyracks-control/hyracks-control-nc/src/main/java/org/apache/hyracks/control/nc/NodeControllerService.java b/hyracks-fullstack/hyracks/hyracks-control/hyracks-control-nc/src/main/java/org/apache/hyracks/control/nc/NodeControllerService.java
index e173dcb..186f288 100644
--- a/hyracks-fullstack/hyracks/hyracks-control/hyracks-control-nc/src/main/java/org/apache/hyracks/control/nc/NodeControllerService.java
+++ b/hyracks-fullstack/hyracks/hyracks-control/hyracks-control-nc/src/main/java/org/apache/hyracks/control/nc/NodeControllerService.java
@@ -267,7 +267,7 @@
resultNetworkManager = new ResultNetworkManager(ncConfig.getResultListenAddress(),
ncConfig.getResultListenPort(), resultPartitionManager, ncConfig.getNetThreadCount(),
ncConfig.getNetBufferCount(), ncConfig.getResultPublicAddress(), ncConfig.getResultPublicPort(),
- FullFrameChannelInterfaceFactory.INSTANCE, networkSecurityManager.getSocketChannelFactory());
+ FullFrameChannelInterfaceFactory.INSTANCE, networkSecurityManager.getSocketChannelFactory(), executor);
if (ncConfig.getMessagingListenAddress() != null && serviceCtx.getMessagingChannelInterfaceFactory() != null) {
messagingNetManager = new MessagingNetworkManager(this, ncConfig.getMessagingListenAddress(),
ncConfig.getMessagingListenPort(), ncConfig.getNetThreadCount(),
@@ -279,7 +279,8 @@
@Override
public void start() throws Exception {
LOGGER.log(Level.INFO, "Starting NodeControllerService");
- ipc = new IPCSystem(new InetSocketAddress(ncConfig.getClusterListenAddress(), ncConfig.getClusterListenPort()),
+ ipc = new IPCSystem(executor,
+ new InetSocketAddress(ncConfig.getClusterListenAddress(), ncConfig.getClusterListenPort()),
networkSecurityManager.getSocketChannelFactory(), new NodeControllerIPCI(this),
new CCNCFunctions.SerializerDeserializer());
ipc.start();
@@ -287,7 +288,7 @@
netManager = new NetworkManager(ncConfig.getDataListenAddress(), ncConfig.getDataListenPort(), partitionManager,
ncConfig.getNetThreadCount(), ncConfig.getNetBufferCount(), ncConfig.getDataPublicAddress(),
ncConfig.getDataPublicPort(), FullFrameChannelInterfaceFactory.INSTANCE,
- networkSecurityManager.getSocketChannelFactory());
+ networkSecurityManager.getSocketChannelFactory(), executor);
netManager.start();
startApplication();
init();
diff --git a/hyracks-fullstack/hyracks/hyracks-control/hyracks-control-nc/src/main/java/org/apache/hyracks/control/nc/net/MessagingNetworkManager.java b/hyracks-fullstack/hyracks/hyracks-control/hyracks-control-nc/src/main/java/org/apache/hyracks/control/nc/net/MessagingNetworkManager.java
index 44f9f02..48bd3cb 100644
--- a/hyracks-fullstack/hyracks/hyracks-control/hyracks-control-nc/src/main/java/org/apache/hyracks/control/nc/net/MessagingNetworkManager.java
+++ b/hyracks-fullstack/hyracks/hyracks-control/hyracks-control-nc/src/main/java/org/apache/hyracks/control/nc/net/MessagingNetworkManager.java
@@ -58,7 +58,7 @@
ISocketChannelFactory socketChannelFactory) {
this.ncs = ncs;
md = new MuxDemux(new InetSocketAddress(inetAddress, inetPort), new ChannelOpenListener(), nThreads,
- MAX_CONNECTION_ATTEMPTS, channelInterfaceFactory, socketChannelFactory);
+ MAX_CONNECTION_ATTEMPTS, channelInterfaceFactory, socketChannelFactory, ncs.getExecutor());
publicNetworkAddress = new NetworkAddress(publicInetAddress, publicInetPort);
}
diff --git a/hyracks-fullstack/hyracks/hyracks-control/hyracks-control-nc/src/main/java/org/apache/hyracks/control/nc/net/NetworkManager.java b/hyracks-fullstack/hyracks/hyracks-control/hyracks-control-nc/src/main/java/org/apache/hyracks/control/nc/net/NetworkManager.java
index 6876618..3e8e702 100644
--- a/hyracks-fullstack/hyracks/hyracks-control/hyracks-control-nc/src/main/java/org/apache/hyracks/control/nc/net/NetworkManager.java
+++ b/hyracks-fullstack/hyracks/hyracks-control/hyracks-control-nc/src/main/java/org/apache/hyracks/control/nc/net/NetworkManager.java
@@ -22,6 +22,7 @@
import java.net.InetSocketAddress;
import java.net.SocketAddress;
import java.nio.ByteBuffer;
+import java.util.concurrent.ExecutorService;
import org.apache.hyracks.api.comm.IChannelInterfaceFactory;
import org.apache.hyracks.api.comm.ICloseableBufferAcceptor;
@@ -61,11 +62,12 @@
public NetworkManager(String inetAddress, int inetPort, PartitionManager partitionManager, int nThreads,
int nBuffers, String publicInetAddress, int publicInetPort,
- IChannelInterfaceFactory channelInterfaceFactory, ISocketChannelFactory socketChannelFactory) {
+ IChannelInterfaceFactory channelInterfaceFactory, ISocketChannelFactory socketChannelFactory,
+ ExecutorService executor) {
this.partitionManager = partitionManager;
this.nBuffers = nBuffers;
md = new MuxDemux(new InetSocketAddress(inetAddress, inetPort), new ChannelOpenListener(), nThreads,
- MAX_CONNECTION_ATTEMPTS, channelInterfaceFactory, socketChannelFactory);
+ MAX_CONNECTION_ATTEMPTS, channelInterfaceFactory, socketChannelFactory, executor);
// Just save these values for the moment; may be reset in start()
publicNetworkAddress = new NetworkAddress(publicInetAddress, publicInetPort);
}
diff --git a/hyracks-fullstack/hyracks/hyracks-control/hyracks-control-nc/src/main/java/org/apache/hyracks/control/nc/net/ResultNetworkManager.java b/hyracks-fullstack/hyracks/hyracks-control/hyracks-control-nc/src/main/java/org/apache/hyracks/control/nc/net/ResultNetworkManager.java
index fe8f2af..7fd46a5 100644
--- a/hyracks-fullstack/hyracks/hyracks-control/hyracks-control-nc/src/main/java/org/apache/hyracks/control/nc/net/ResultNetworkManager.java
+++ b/hyracks-fullstack/hyracks/hyracks-control/hyracks-control-nc/src/main/java/org/apache/hyracks/control/nc/net/ResultNetworkManager.java
@@ -22,6 +22,7 @@
import java.net.InetSocketAddress;
import java.net.SocketAddress;
import java.nio.ByteBuffer;
+import java.util.concurrent.ExecutorService;
import org.apache.hyracks.api.comm.IChannelInterfaceFactory;
import org.apache.hyracks.api.comm.ICloseableBufferAcceptor;
@@ -61,24 +62,22 @@
private NetworkAddress publicNetworkAddress;
/**
- * @param inetAddress
- * - Internet address to bind the listen port to
- * @param inetPort
- * - Port to bind on inetAddress
- * @param publicInetAddress
- * - Internet address to report to consumers;
- * useful when behind NAT. null = same as inetAddress
- * @param publicInetPort
- * - Port to report to consumers; useful when
- * behind NAT. Ignored if publicInetAddress is null. 0 = same as inetPort
+ * @param inetAddress - Internet address to bind the listen port to
+ * @param inetPort - Port to bind on inetAddress
+ * @param publicInetAddress - Internet address to report to consumers;
+ * useful when behind NAT. null = same as inetAddress
+ * @param publicInetPort - Port to report to consumers; useful when
+ * behind NAT. Ignored if publicInetAddress is null. 0 = same as inetPort
+ * @param executor
*/
public ResultNetworkManager(String inetAddress, int inetPort, IResultPartitionManager partitionManager,
int nThreads, int nBuffers, String publicInetAddress, int publicInetPort,
- IChannelInterfaceFactory channelInterfaceFactory, ISocketChannelFactory socketChannelFactory) {
+ IChannelInterfaceFactory channelInterfaceFactory, ISocketChannelFactory socketChannelFactory,
+ ExecutorService executor) {
this.partitionManager = partitionManager;
this.nBuffers = nBuffers;
md = new MuxDemux(new InetSocketAddress(inetAddress, inetPort), new ChannelOpenListener(), nThreads,
- MAX_CONNECTION_ATTEMPTS, channelInterfaceFactory, socketChannelFactory);
+ MAX_CONNECTION_ATTEMPTS, channelInterfaceFactory, socketChannelFactory, executor);
// Just save these values for the moment; may be reset in start()
publicNetworkAddress = new NetworkAddress(publicInetAddress, publicInetPort);
}
diff --git a/hyracks-fullstack/hyracks/hyracks-examples/hyracks-integration-tests/src/test/java/org/apache/hyracks/tests/integration/AbstractIntegrationTest.java b/hyracks-fullstack/hyracks/hyracks-examples/hyracks-integration-tests/src/test/java/org/apache/hyracks/tests/integration/AbstractIntegrationTest.java
index 6f59ed8..f5ddf2e 100644
--- a/hyracks-fullstack/hyracks/hyracks-examples/hyracks-integration-tests/src/test/java/org/apache/hyracks/tests/integration/AbstractIntegrationTest.java
+++ b/hyracks-fullstack/hyracks/hyracks-examples/hyracks-integration-tests/src/test/java/org/apache/hyracks/tests/integration/AbstractIntegrationTest.java
@@ -28,6 +28,7 @@
import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
+import java.util.concurrent.Executors;
import java.util.concurrent.atomic.AtomicInteger;
import org.apache.commons.io.FileUtils;
@@ -153,7 +154,7 @@
IFrameTupleAccessor frameTupleAccessor = new ResultFrameTupleAccessor();
- IResultSet resultSet = new ResultSet(hcc, PlainSocketChannelFactory.INSTANCE, spec.getFrameSize(), nReaders);
+ IResultSet resultSet = new ResultSet(Executors.newCachedThreadPool(), hcc, PlainSocketChannelFactory.INSTANCE, spec.getFrameSize(), nReaders);
IResultSetReader reader = resultSet.createReader(jobId, resultSetId);
List<String> resultRecords = new ArrayList<>();
diff --git a/hyracks-fullstack/hyracks/hyracks-examples/hyracks-integration-tests/src/test/java/org/apache/hyracks/tests/integration/AbstractMultiNCIntegrationTest.java b/hyracks-fullstack/hyracks/hyracks-examples/hyracks-integration-tests/src/test/java/org/apache/hyracks/tests/integration/AbstractMultiNCIntegrationTest.java
index be22b9c..98871c6 100644
--- a/hyracks-fullstack/hyracks/hyracks-examples/hyracks-integration-tests/src/test/java/org/apache/hyracks/tests/integration/AbstractMultiNCIntegrationTest.java
+++ b/hyracks-fullstack/hyracks/hyracks-examples/hyracks-integration-tests/src/test/java/org/apache/hyracks/tests/integration/AbstractMultiNCIntegrationTest.java
@@ -23,6 +23,7 @@
import java.util.ArrayList;
import java.util.EnumSet;
import java.util.List;
+import java.util.concurrent.Executors;
import org.apache.commons.io.FileUtils;
import org.apache.hyracks.api.client.IHyracksClientConnection;
@@ -162,7 +163,7 @@
if (!spec.getResultSetIds().isEmpty()) {
IResultSet resultSet =
- new ResultSet(hcc, PlainSocketChannelFactory.INSTANCE, spec.getFrameSize(), nReaders);
+ new ResultSet(Executors.newCachedThreadPool(), hcc, PlainSocketChannelFactory.INSTANCE, spec.getFrameSize(), nReaders);
IResultSetReader reader = resultSet.createReader(jobId, spec.getResultSetIds().get(0));
ObjectMapper om = new ObjectMapper();
diff --git a/hyracks-fullstack/hyracks/hyracks-hdfs/src/main/java/org/apache/hyracks/hdfs/scheduler/Scheduler.java b/hyracks-fullstack/hyracks/hyracks-hdfs/src/main/java/org/apache/hyracks/hdfs/scheduler/Scheduler.java
index f5bf07b..46aa91d 100644
--- a/hyracks-fullstack/hyracks/hyracks-hdfs/src/main/java/org/apache/hyracks/hdfs/scheduler/Scheduler.java
+++ b/hyracks-fullstack/hyracks/hyracks-hdfs/src/main/java/org/apache/hyracks/hdfs/scheduler/Scheduler.java
@@ -30,6 +30,7 @@
import java.util.Map;
import java.util.PriorityQueue;
import java.util.Random;
+import java.util.concurrent.ExecutorService;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.mapred.InputSplit;
@@ -88,12 +89,14 @@
* @param ipAddress IP address
* @param port Port
* @param channelFactory Channel Factory
+ * @param executor
* @throws HyracksException
*/
- public Scheduler(String ipAddress, int port, ISocketChannelFactory channelFactory) throws HyracksException {
+ public Scheduler(String ipAddress, int port, ISocketChannelFactory channelFactory, ExecutorService executor)
+ throws HyracksException {
try {
- IHyracksClientConnection hcc = new HyracksConnection(ipAddress, port, channelFactory);
+ IHyracksClientConnection hcc = new HyracksConnection(ipAddress, port, channelFactory, executor);
this.ncNameToNcInfos = hcc.getNodeControllerInfos();
ClusterTopology topology = hcc.getClusterTopology();
this.ncCollectionBuilder = topology == null ? new IPProximityNcCollectionBuilder()
diff --git a/hyracks-fullstack/hyracks/hyracks-hdfs/src/main/java/org/apache/hyracks/hdfs2/scheduler/Scheduler.java b/hyracks-fullstack/hyracks/hyracks-hdfs/src/main/java/org/apache/hyracks/hdfs2/scheduler/Scheduler.java
index a26a5f7..27c775d 100644
--- a/hyracks-fullstack/hyracks/hyracks-hdfs/src/main/java/org/apache/hyracks/hdfs2/scheduler/Scheduler.java
+++ b/hyracks-fullstack/hyracks/hyracks-hdfs/src/main/java/org/apache/hyracks/hdfs2/scheduler/Scheduler.java
@@ -21,6 +21,7 @@
import java.util.List;
import java.util.Map;
+import java.util.concurrent.ExecutorService;
import org.apache.hadoop.mapreduce.InputSplit;
import org.apache.hyracks.api.client.NodeControllerInfo;
@@ -43,10 +44,12 @@
* The constructor of the scheduler
*
* @param ncNameToNcInfos
+ * @param executor
* @throws HyracksException
*/
- public Scheduler(String ipAddress, int port, ISocketChannelFactory channelFactory) throws HyracksException {
- scheduler = new org.apache.hyracks.hdfs.scheduler.Scheduler(ipAddress, port, channelFactory);
+ public Scheduler(String ipAddress, int port, ISocketChannelFactory channelFactory, ExecutorService executor)
+ throws HyracksException {
+ scheduler = new org.apache.hyracks.hdfs.scheduler.Scheduler(ipAddress, port, channelFactory, executor);
}
/**
diff --git a/hyracks-fullstack/hyracks/hyracks-ipc/src/main/java/org/apache/hyracks/ipc/impl/HyracksConnection.java b/hyracks-fullstack/hyracks/hyracks-ipc/src/main/java/org/apache/hyracks/ipc/impl/HyracksConnection.java
index 1b6d8af..057bce4 100644
--- a/hyracks-fullstack/hyracks/hyracks-ipc/src/main/java/org/apache/hyracks/ipc/impl/HyracksConnection.java
+++ b/hyracks-fullstack/hyracks/hyracks-ipc/src/main/java/org/apache/hyracks/ipc/impl/HyracksConnection.java
@@ -98,19 +98,19 @@
/**
* Constructor to create a connection to the Hyracks Cluster Controller.
*
- * @param ccHost
- * Host name (or IP Address) where the Cluster Controller can be
- * reached.
- * @param ccPort
- * Port to reach the Hyracks Cluster Controller at the specified
- * host name.
+ * @param ccHost Host name (or IP Address) where the Cluster Controller can be
+ * reached.
+ * @param ccPort Port to reach the Hyracks Cluster Controller at the specified
+ * host name.
+ * @param executor
* @throws Exception
*/
- public HyracksConnection(String ccHost, int ccPort, ISocketChannelFactory socketChannelFactory) throws Exception {
+ public HyracksConnection(String ccHost, int ccPort, ISocketChannelFactory socketChannelFactory,
+ ExecutorService executor) throws Exception {
this.ccHost = ccHost;
this.ccPort = ccPort;
RPCInterface rpci = new RPCInterface();
- ipc = new IPCSystem(new InetSocketAddress(0), socketChannelFactory, rpci,
+ ipc = new IPCSystem(executor, new InetSocketAddress(0), socketChannelFactory, rpci,
new JavaSerializationBasedPayloadSerializerDeserializer());
ipc.start();
hci = new HyracksClientInterfaceRemoteProxy(ipc.getReconnectingHandle(new InetSocketAddress(ccHost, ccPort)),
@@ -121,7 +121,11 @@
}
public HyracksConnection(String ccHost, int ccPort) throws Exception {
- this(ccHost, ccPort, PlainSocketChannelFactory.INSTANCE);
+ this(ccHost, ccPort, Executors.newCachedThreadPool());
+ }
+
+ public HyracksConnection(String ccHost, int ccPort, ExecutorService executor) throws Exception {
+ this(ccHost, ccPort, PlainSocketChannelFactory.INSTANCE, executor);
}
@Override
diff --git a/hyracks-fullstack/hyracks/hyracks-ipc/src/main/java/org/apache/hyracks/ipc/impl/IPCConnectionManager.java b/hyracks-fullstack/hyracks/hyracks-ipc/src/main/java/org/apache/hyracks/ipc/impl/IPCConnectionManager.java
index 5834f26..5105071 100644
--- a/hyracks-fullstack/hyracks/hyracks-ipc/src/main/java/org/apache/hyracks/ipc/impl/IPCConnectionManager.java
+++ b/hyracks-fullstack/hyracks/hyracks-ipc/src/main/java/org/apache/hyracks/ipc/impl/IPCConnectionManager.java
@@ -37,6 +37,7 @@
import java.util.Map;
import java.util.Set;
import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutorService;
import java.util.concurrent.TimeUnit;
import org.apache.hyracks.api.network.ISocketChannel;
@@ -77,11 +78,13 @@
private volatile boolean stopped;
private final ISocketChannelFactory socketChannelFactory;
+ private final ExecutorService executor;
- IPCConnectionManager(IPCSystem system, InetSocketAddress socketAddress, ISocketChannelFactory socketChannelFactory)
- throws IOException {
+ IPCConnectionManager(IPCSystem system, InetSocketAddress socketAddress, ISocketChannelFactory socketChannelFactory,
+ ExecutorService executor) throws IOException {
this.system = system;
this.socketChannelFactory = socketChannelFactory;
+ this.executor = executor;
this.serverSocketChannel = ServerSocketChannel.open();
serverSocketChannel.socket().setReuseAddress(true);
serverSocketChannel.configureBlocking(false);
@@ -513,8 +516,10 @@
}
private void asyncHandshake(ISocketChannel socketChannel, IPCHandle handle, SelectionKey channelKey) {
- CompletableFuture.supplyAsync(socketChannel::handshake).exceptionally(ex -> false).thenAccept(
- handshakeSuccess -> handleHandshakeCompletion(handshakeSuccess, socketChannel, handle, channelKey));
+ CompletableFuture.supplyAsync(socketChannel::handshake, IPCConnectionManager.this.executor)
+ .exceptionally(ex -> false)
+ .thenAccept(handshakeSuccess -> handleHandshakeCompletion(handshakeSuccess, socketChannel, handle,
+ channelKey));
}
private void handleHandshakeCompletion(Boolean handshakeSuccess, ISocketChannel socketChannel, IPCHandle handle,
diff --git a/hyracks-fullstack/hyracks/hyracks-ipc/src/main/java/org/apache/hyracks/ipc/impl/IPCSystem.java b/hyracks-fullstack/hyracks/hyracks-ipc/src/main/java/org/apache/hyracks/ipc/impl/IPCSystem.java
index 4a19a33..5d4fc68 100644
--- a/hyracks-fullstack/hyracks/hyracks-ipc/src/main/java/org/apache/hyracks/ipc/impl/IPCSystem.java
+++ b/hyracks-fullstack/hyracks/hyracks-ipc/src/main/java/org/apache/hyracks/ipc/impl/IPCSystem.java
@@ -20,6 +20,7 @@
import java.io.IOException;
import java.net.InetSocketAddress;
+import java.util.concurrent.ExecutorService;
import java.util.concurrent.atomic.AtomicLong;
import org.apache.hyracks.api.network.ISocketChannelFactory;
@@ -41,9 +42,10 @@
private final IPCPerformanceCounters perfCounters;
- public IPCSystem(InetSocketAddress socketAddress, ISocketChannelFactory socketChannelFactory, IIPCI ipci,
- IPayloadSerializerDeserializer serde) throws IOException {
- cMgr = new IPCConnectionManager(this, socketAddress, socketChannelFactory);
+ public IPCSystem(ExecutorService executor, InetSocketAddress socketAddress,
+ ISocketChannelFactory socketChannelFactory, IIPCI ipci, IPayloadSerializerDeserializer serde)
+ throws IOException {
+ cMgr = new IPCConnectionManager(this, socketAddress, socketChannelFactory, executor);
this.ipci = ipci;
this.serde = serde;
midFactory = new AtomicLong();
diff --git a/hyracks-fullstack/hyracks/hyracks-ipc/src/test/java/org/apache/hyracks/ipc/tests/IPCTest.java b/hyracks-fullstack/hyracks/hyracks-ipc/src/test/java/org/apache/hyracks/ipc/tests/IPCTest.java
index 0d8d69b..75d039e 100644
--- a/hyracks-fullstack/hyracks/hyracks-ipc/src/test/java/org/apache/hyracks/ipc/tests/IPCTest.java
+++ b/hyracks-fullstack/hyracks/hyracks-ipc/src/test/java/org/apache/hyracks/ipc/tests/IPCTest.java
@@ -20,7 +20,7 @@
import java.io.IOException;
import java.net.InetSocketAddress;
-import java.util.concurrent.Executor;
+import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import org.apache.hyracks.ipc.api.IIPCHandle;
@@ -59,7 +59,7 @@
}
private IPCSystem createServerIPCSystem() throws IOException {
- final Executor executor = Executors.newCachedThreadPool();
+ final ExecutorService executor = Executors.newCachedThreadPool();
IIPCI ipci = new IIPCI() {
@Override
public void deliverIncomingMessage(IIPCHandle handle, long mid, long rmid, Object payload) {
@@ -81,12 +81,13 @@
exception.printStackTrace();
}
};
- return new IPCSystem(new InetSocketAddress("127.0.0.1", 0), PlainSocketChannelFactory.INSTANCE, ipci,
+ return new IPCSystem(executor, new InetSocketAddress("127.0.0.1", 0), PlainSocketChannelFactory.INSTANCE, ipci,
new JavaSerializationBasedPayloadSerializerDeserializer());
}
private IPCSystem createClientIPCSystem(RPCInterface rpci) throws IOException {
- return new IPCSystem(new InetSocketAddress("127.0.0.1", 0), PlainSocketChannelFactory.INSTANCE, rpci,
+ final ExecutorService executor = Executors.newCachedThreadPool();
+ return new IPCSystem(executor, new InetSocketAddress("127.0.0.1", 0), PlainSocketChannelFactory.INSTANCE, rpci,
new JavaSerializationBasedPayloadSerializerDeserializer());
}
}
diff --git a/hyracks-fullstack/hyracks/hyracks-net/src/main/java/org/apache/hyracks/net/protocols/muxdemux/MuxDemux.java b/hyracks-fullstack/hyracks/hyracks-net/src/main/java/org/apache/hyracks/net/protocols/muxdemux/MuxDemux.java
index 3189ea1..a63c91a 100644
--- a/hyracks-fullstack/hyracks/hyracks-net/src/main/java/org/apache/hyracks/net/protocols/muxdemux/MuxDemux.java
+++ b/hyracks-fullstack/hyracks/hyracks-net/src/main/java/org/apache/hyracks/net/protocols/muxdemux/MuxDemux.java
@@ -22,6 +22,7 @@
import java.net.InetSocketAddress;
import java.util.HashMap;
import java.util.Map;
+import java.util.concurrent.ExecutorService;
import org.apache.hyracks.api.comm.IChannelInterfaceFactory;
import org.apache.hyracks.api.exceptions.NetException;
@@ -61,22 +62,17 @@
/**
* Constructor.
*
- * @param localAddress
- * - TCP/IP socket address to listen on. Null for non-listening unidirectional sockets
- * @param listener
- * - Callback interface to report channel events. Null for non-listening unidirectional sockets
- * @param nThreads
- * - Number of threads to use for data transfer
- * @param maxConnectionAttempts
- * - Maximum number of connection attempts
- * @param channelInterfaceFactory
- * - The channel interface factory
- * @param socketChannelFactory
- * - The socket channel factory
+ * @param localAddress - TCP/IP socket address to listen on. Null for non-listening unidirectional sockets
+ * @param listener - Callback interface to report channel events. Null for non-listening unidirectional sockets
+ * @param nThreads - Number of threads to use for data transfer
+ * @param maxConnectionAttempts - Maximum number of connection attempts
+ * @param channelInterfaceFactory - The channel interface factory
+ * @param socketChannelFactory - The socket channel factory
+ * @param executor
*/
public MuxDemux(InetSocketAddress localAddress, IChannelOpenListener listener, int nThreads,
int maxConnectionAttempts, IChannelInterfaceFactory channelInterfaceFactory,
- ISocketChannelFactory socketChannelFactory) {
+ ISocketChannelFactory socketChannelFactory, ExecutorService executor) {
this.localAddress = localAddress;
this.channelOpenListener = listener;
this.maxConnectionAttempts = maxConnectionAttempts;
@@ -132,7 +128,7 @@
}
}
}
- }, nThreads, socketChannelFactory);
+ }, nThreads, socketChannelFactory, executor);
perfCounters = new MuxDemuxPerformanceCounters();
}
diff --git a/hyracks-fullstack/hyracks/hyracks-net/src/main/java/org/apache/hyracks/net/protocols/tcp/TCPEndpoint.java b/hyracks-fullstack/hyracks/hyracks-net/src/main/java/org/apache/hyracks/net/protocols/tcp/TCPEndpoint.java
index b75e3c6..07db828 100644
--- a/hyracks-fullstack/hyracks/hyracks-net/src/main/java/org/apache/hyracks/net/protocols/tcp/TCPEndpoint.java
+++ b/hyracks-fullstack/hyracks/hyracks-net/src/main/java/org/apache/hyracks/net/protocols/tcp/TCPEndpoint.java
@@ -34,6 +34,7 @@
import java.util.Iterator;
import java.util.List;
import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutorService;
import org.apache.hyracks.api.network.ISocketChannel;
import org.apache.hyracks.api.network.ISocketChannelFactory;
@@ -58,12 +59,14 @@
private int nextThread;
private final ISocketChannelFactory socketChannelFactory;
+ private final ExecutorService executor;
public TCPEndpoint(ITCPConnectionListener connectionListener, int nThreads,
- ISocketChannelFactory socketChannelFactory) {
+ ISocketChannelFactory socketChannelFactory, ExecutorService executor) {
this.connectionListener = connectionListener;
this.nThreads = nThreads;
this.socketChannelFactory = socketChannelFactory;
+ this.executor = executor;
}
public void start(InetSocketAddress localAddress) throws IOException {
@@ -314,7 +317,8 @@
}
private void asyncHandshake(PendingHandshakeConnection connection) {
- CompletableFuture.supplyAsync(connection.socketChannel::handshake).exceptionally(ex -> false)
+ CompletableFuture.supplyAsync(connection.socketChannel::handshake, TCPEndpoint.this.executor)
+ .exceptionally(ex -> false)
.thenAccept(handshakeSuccess -> handleHandshakeCompletion(handshakeSuccess, connection));
}
diff --git a/hyracks-fullstack/hyracks/hyracks-net/src/test/java/org/apache/hyracks/net/tests/NetTest.java b/hyracks-fullstack/hyracks/hyracks-net/src/test/java/org/apache/hyracks/net/tests/NetTest.java
index ec1d21c..823d638 100644
--- a/hyracks-fullstack/hyracks/hyracks-net/src/test/java/org/apache/hyracks/net/tests/NetTest.java
+++ b/hyracks-fullstack/hyracks/hyracks-net/src/test/java/org/apache/hyracks/net/tests/NetTest.java
@@ -22,6 +22,7 @@
import java.nio.ByteBuffer;
import java.util.LinkedList;
import java.util.Queue;
+import java.util.concurrent.Executors;
import java.util.concurrent.Semaphore;
import java.util.concurrent.atomic.AtomicBoolean;
@@ -163,7 +164,8 @@
}
};
return new MuxDemux(new InetSocketAddress("127.0.0.1", 0), md1OpenListener, 1, 5,
- FullFrameChannelInterfaceFactory.INSTANCE, PlainSocketChannelFactory.INSTANCE);
+ FullFrameChannelInterfaceFactory.INSTANCE, PlainSocketChannelFactory.INSTANCE,
+ Executors.newCachedThreadPool());
}
private class ChannelIO {
--
To view, visit https://asterix-gerrit.ics.uci.edu/c/asterixdb/+/18168
To unsubscribe, or for help writing mail filters, visit https://asterix-gerrit.ics.uci.edu/settings
Gerrit-Project: asterixdb
Gerrit-Branch: trinity
Gerrit-Change-Id: Ibe2fdce979a49d962517494dc77a5616a19f6abf
Gerrit-Change-Number: 18168
Gerrit-PatchSet: 1
Gerrit-Owner: Michael Blow <mb...@apache.org>
Gerrit-MessageType: newchange