You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iotdb.apache.org by ta...@apache.org on 2022/06/06 03:09:48 UTC
[iotdb] branch jira3395 created (now a58dffabce)
This is an automated email from the ASF dual-hosted git repository.
tanxinyu pushed a change to branch jira3395
in repository https://gitbox.apache.org/repos/asf/iotdb.git
at a58dffabce Use thrift server to fix clientManagerTest bind address already used issue
This branch includes the following new commits:
new a58dffabce Use thrift server to fix clientManagerTest bind address already used issue
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
[iotdb] 01/01: Use thrift server to fix clientManagerTest bind address already used issue
Posted by ta...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
tanxinyu pushed a commit to branch jira3395
in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit a58dffabce74cec845229aa8175d7cd3b31e3ee7
Author: LebronAl <TX...@gmail.com>
AuthorDate: Mon Jun 6 11:09:27 2022 +0800
Use thrift server to fix clientManagerTest bind address already used issue
---
.../commons/{ => client}/ClientManagerTest.java | 56 ++-------
.../client/mock/MockInternalRPCService.java | 88 +++++++++++++
.../client/mock/MockInternalRPCServiceMBean.java | 22 ++++
.../mock/MockInternalRPCServiceProcessor.java | 136 +++++++++++++++++++++
.../mock/MockInternalServiceThriftHandler.java | 44 +++++++
5 files changed, 300 insertions(+), 46 deletions(-)
diff --git a/node-commons/src/test/java/org/apache/iotdb/commons/ClientManagerTest.java b/node-commons/src/test/java/org/apache/iotdb/commons/client/ClientManagerTest.java
similarity index 93%
rename from node-commons/src/test/java/org/apache/iotdb/commons/ClientManagerTest.java
rename to node-commons/src/test/java/org/apache/iotdb/commons/client/ClientManagerTest.java
index f760082c26..81e2432994 100644
--- a/node-commons/src/test/java/org/apache/iotdb/commons/ClientManagerTest.java
+++ b/node-commons/src/test/java/org/apache/iotdb/commons/client/ClientManagerTest.java
@@ -17,17 +17,15 @@
* under the License.
*/
-package org.apache.iotdb.commons;
+package org.apache.iotdb.commons.client;
import org.apache.iotdb.common.rpc.thrift.TEndPoint;
-import org.apache.iotdb.commons.client.ClientFactoryProperty;
-import org.apache.iotdb.commons.client.ClientManager;
-import org.apache.iotdb.commons.client.ClientPoolProperty;
import org.apache.iotdb.commons.client.ClientPoolProperty.DefaultProperty;
-import org.apache.iotdb.commons.client.IClientManager;
-import org.apache.iotdb.commons.client.IClientPoolFactory;
import org.apache.iotdb.commons.client.async.AsyncDataNodeInternalServiceClient;
+import org.apache.iotdb.commons.client.mock.MockInternalRPCService;
+import org.apache.iotdb.commons.client.mock.MockInternalRPCServiceProcessor;
import org.apache.iotdb.commons.client.sync.SyncDataNodeInternalServiceClient;
+import org.apache.iotdb.commons.exception.StartupException;
import org.apache.commons.pool2.KeyedObjectPool;
import org.apache.commons.pool2.impl.GenericKeyedObjectPool;
@@ -37,24 +35,23 @@ import org.junit.Before;
import org.junit.Test;
import java.io.IOException;
-import java.net.InetSocketAddress;
-import java.net.ServerSocket;
public class ClientManagerTest {
private final TEndPoint endPoint = new TEndPoint("localhost", 9003);
- private ServerSocket metaServer;
- private Thread metaServerListeningThread;
+ private MockInternalRPCService service;
@Before
- public void setUp() throws IOException {
- startServer();
+ public void setUp() throws StartupException {
+ service = new MockInternalRPCService(endPoint);
+ service.initSyncedServiceImpl(new MockInternalRPCServiceProcessor());
+ service.start();
}
@After
public void tearDown() throws IOException, InterruptedException {
- stopServer();
+ service.waitAndStop(10_000L);
}
@Test
@@ -418,39 +415,6 @@ public class ClientManagerTest {
Assert.assertEquals(0, asyncClusterManager.getPool().getNumIdle(endPoint));
}
- public void startServer() throws IOException {
- metaServer = new ServerSocket();
- // reuse the port to avoid `Bind Address already in use` which is caused by TIME_WAIT state
- // because port won't be usable immediately after we close it.
- metaServer.setReuseAddress(true);
- metaServer.bind(new InetSocketAddress(endPoint.getIp(), endPoint.getPort()), 0);
-
- metaServerListeningThread =
- new Thread(
- () -> {
- while (!Thread.interrupted()) {
- try {
- metaServer.accept();
- } catch (IOException e) {
- return;
- }
- }
- });
- metaServerListeningThread.start();
- }
-
- public void stopServer() throws InterruptedException, IOException {
- if (metaServer != null) {
- metaServer.close();
- metaServer = null;
- }
- if (metaServerListeningThread != null) {
- metaServerListeningThread.interrupt();
- metaServerListeningThread.join();
- metaServerListeningThread = null;
- }
- }
-
public static class TestSyncDataNodeInternalServiceClientPoolFactory
implements IClientPoolFactory<TEndPoint, SyncDataNodeInternalServiceClient> {
@Override
diff --git a/node-commons/src/test/java/org/apache/iotdb/commons/client/mock/MockInternalRPCService.java b/node-commons/src/test/java/org/apache/iotdb/commons/client/mock/MockInternalRPCService.java
new file mode 100644
index 0000000000..e61d2388a0
--- /dev/null
+++ b/node-commons/src/test/java/org/apache/iotdb/commons/client/mock/MockInternalRPCService.java
@@ -0,0 +1,88 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iotdb.commons.client.mock;
+
+import org.apache.iotdb.common.rpc.thrift.TEndPoint;
+import org.apache.iotdb.commons.concurrent.ThreadName;
+import org.apache.iotdb.commons.conf.IoTDBConstant;
+import org.apache.iotdb.commons.exception.runtime.RPCServiceException;
+import org.apache.iotdb.commons.service.ServiceType;
+import org.apache.iotdb.commons.service.ThriftService;
+import org.apache.iotdb.commons.service.ThriftServiceThread;
+import org.apache.iotdb.mpp.rpc.thrift.InternalService;
+
+public class MockInternalRPCService extends ThriftService implements MockInternalRPCServiceMBean {
+
+ private final TEndPoint thisNode;
+ private MockInternalRPCServiceProcessor internalServiceImpl;
+
+ public MockInternalRPCService(TEndPoint thisNode) {
+ this.thisNode = thisNode;
+ }
+
+ @Override
+ public ServiceType getID() {
+ return ServiceType.INTERNAL_SERVICE;
+ }
+
+ @Override
+ public void initSyncedServiceImpl(Object internalServiceImpl) {
+ this.internalServiceImpl = (MockInternalRPCServiceProcessor) internalServiceImpl;
+ super.mbeanName =
+ String.format(
+ "%s:%s=%s", this.getClass().getPackage(), IoTDBConstant.JMX_TYPE, getID().getJmxName());
+ super.initSyncedServiceImpl(this.internalServiceImpl);
+ }
+
+ @Override
+ public void initTProcessor() {
+ processor = new InternalService.Processor<>(internalServiceImpl);
+ }
+
+ @Override
+ public void initThriftServiceThread() throws IllegalAccessException {
+ try {
+ thriftServiceThread =
+ new ThriftServiceThread(
+ processor,
+ getID().getName(),
+ ThreadName.INTERNAL_SERVICE_RPC_SERVER.getName(),
+ getBindIP(),
+ getBindPort(),
+ 65535,
+ 60,
+ new MockInternalServiceThriftHandler(),
+ false);
+ } catch (RPCServiceException e) {
+ throw new IllegalAccessException(e.getMessage());
+ }
+ thriftServiceThread.setName(ThreadName.INTERNAL_SERVICE_RPC_SERVER.getName());
+ }
+
+ @Override
+ public String getBindIP() {
+ return thisNode.getIp();
+ }
+
+ @Override
+ public int getBindPort() {
+ return thisNode.getPort();
+ }
+}
diff --git a/node-commons/src/test/java/org/apache/iotdb/commons/client/mock/MockInternalRPCServiceMBean.java b/node-commons/src/test/java/org/apache/iotdb/commons/client/mock/MockInternalRPCServiceMBean.java
new file mode 100644
index 0000000000..35c48eb381
--- /dev/null
+++ b/node-commons/src/test/java/org/apache/iotdb/commons/client/mock/MockInternalRPCServiceMBean.java
@@ -0,0 +1,22 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iotdb.commons.client.mock;
+
+public interface MockInternalRPCServiceMBean {}
diff --git a/node-commons/src/test/java/org/apache/iotdb/commons/client/mock/MockInternalRPCServiceProcessor.java b/node-commons/src/test/java/org/apache/iotdb/commons/client/mock/MockInternalRPCServiceProcessor.java
new file mode 100644
index 0000000000..41d0823d16
--- /dev/null
+++ b/node-commons/src/test/java/org/apache/iotdb/commons/client/mock/MockInternalRPCServiceProcessor.java
@@ -0,0 +1,136 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iotdb.commons.client.mock;
+
+import org.apache.iotdb.common.rpc.thrift.TConsensusGroupId;
+import org.apache.iotdb.common.rpc.thrift.THeartbeatReq;
+import org.apache.iotdb.common.rpc.thrift.THeartbeatResp;
+import org.apache.iotdb.common.rpc.thrift.TSStatus;
+import org.apache.iotdb.mpp.rpc.thrift.InternalService;
+import org.apache.iotdb.mpp.rpc.thrift.TCancelFragmentInstanceReq;
+import org.apache.iotdb.mpp.rpc.thrift.TCancelPlanFragmentReq;
+import org.apache.iotdb.mpp.rpc.thrift.TCancelQueryReq;
+import org.apache.iotdb.mpp.rpc.thrift.TCancelResp;
+import org.apache.iotdb.mpp.rpc.thrift.TCreateDataRegionReq;
+import org.apache.iotdb.mpp.rpc.thrift.TCreateFunctionRequest;
+import org.apache.iotdb.mpp.rpc.thrift.TCreateSchemaRegionReq;
+import org.apache.iotdb.mpp.rpc.thrift.TDropFunctionRequest;
+import org.apache.iotdb.mpp.rpc.thrift.TFetchFragmentInstanceStateReq;
+import org.apache.iotdb.mpp.rpc.thrift.TFragmentInstanceStateResp;
+import org.apache.iotdb.mpp.rpc.thrift.TInvalidateCacheReq;
+import org.apache.iotdb.mpp.rpc.thrift.TInvalidatePermissionCacheReq;
+import org.apache.iotdb.mpp.rpc.thrift.TMigrateDataRegionReq;
+import org.apache.iotdb.mpp.rpc.thrift.TMigrateSchemaRegionReq;
+import org.apache.iotdb.mpp.rpc.thrift.TSchemaFetchRequest;
+import org.apache.iotdb.mpp.rpc.thrift.TSchemaFetchResponse;
+import org.apache.iotdb.mpp.rpc.thrift.TSendFragmentInstanceReq;
+import org.apache.iotdb.mpp.rpc.thrift.TSendFragmentInstanceResp;
+
+import org.apache.thrift.TException;
+
+public class MockInternalRPCServiceProcessor implements InternalService.Iface {
+
+ @Override
+ public TSendFragmentInstanceResp sendFragmentInstance(TSendFragmentInstanceReq req)
+ throws TException {
+ return null;
+ }
+
+ @Override
+ public TFragmentInstanceStateResp fetchFragmentInstanceState(TFetchFragmentInstanceStateReq req)
+ throws TException {
+ return null;
+ }
+
+ @Override
+ public TCancelResp cancelQuery(TCancelQueryReq req) throws TException {
+ return null;
+ }
+
+ @Override
+ public TCancelResp cancelPlanFragment(TCancelPlanFragmentReq req) throws TException {
+ return null;
+ }
+
+ @Override
+ public TCancelResp cancelFragmentInstance(TCancelFragmentInstanceReq req) throws TException {
+ return null;
+ }
+
+ @Override
+ public TSchemaFetchResponse fetchSchema(TSchemaFetchRequest req) throws TException {
+ return null;
+ }
+
+ @Override
+ public TSStatus createSchemaRegion(TCreateSchemaRegionReq req) throws TException {
+ return null;
+ }
+
+ @Override
+ public TSStatus createDataRegion(TCreateDataRegionReq req) throws TException {
+ return null;
+ }
+
+ @Override
+ public TSStatus invalidatePartitionCache(TInvalidateCacheReq req) throws TException {
+ return null;
+ }
+
+ @Override
+ public TSStatus invalidateSchemaCache(TInvalidateCacheReq req) throws TException {
+ return null;
+ }
+
+ @Override
+ public TSStatus deleteRegion(TConsensusGroupId consensusGroupId) throws TException {
+ return null;
+ }
+
+ @Override
+ public TSStatus migrateSchemaRegion(TMigrateSchemaRegionReq req) throws TException {
+ return null;
+ }
+
+ @Override
+ public TSStatus migrateDataRegion(TMigrateDataRegionReq req) throws TException {
+ return null;
+ }
+
+ @Override
+ public THeartbeatResp getHeartBeat(THeartbeatReq req) throws TException {
+ return null;
+ }
+
+ @Override
+ public TSStatus createFunction(TCreateFunctionRequest req) throws TException {
+ return null;
+ }
+
+ @Override
+ public TSStatus dropFunction(TDropFunctionRequest req) throws TException {
+ return null;
+ }
+
+ @Override
+ public TSStatus invalidatePermissionCache(TInvalidatePermissionCacheReq req) throws TException {
+ return null;
+ }
+}
diff --git a/node-commons/src/test/java/org/apache/iotdb/commons/client/mock/MockInternalServiceThriftHandler.java b/node-commons/src/test/java/org/apache/iotdb/commons/client/mock/MockInternalServiceThriftHandler.java
new file mode 100644
index 0000000000..7c14991d93
--- /dev/null
+++ b/node-commons/src/test/java/org/apache/iotdb/commons/client/mock/MockInternalServiceThriftHandler.java
@@ -0,0 +1,44 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iotdb.commons.client.mock;
+
+import org.apache.thrift.protocol.TProtocol;
+import org.apache.thrift.server.ServerContext;
+import org.apache.thrift.server.TServerEventHandler;
+import org.apache.thrift.transport.TTransport;
+
+public class MockInternalServiceThriftHandler implements TServerEventHandler {
+
+ @Override
+ public void preServe() {}
+
+ @Override
+ public ServerContext createContext(TProtocol tProtocol, TProtocol tProtocol1) {
+ return null;
+ }
+
+ @Override
+ public void deleteContext(
+ ServerContext serverContext, TProtocol tProtocol, TProtocol tProtocol1) {}
+
+ @Override
+ public void processContext(
+ ServerContext serverContext, TTransport tTransport, TTransport tTransport1) {}
+}