This is an automated email from the ASF dual-hosted git repository.
wangchao316 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/master by this push:
new 3b59ee22a0 [IOTDB-3395] Use thrift server to fix clientManagerTest
"bind address already used" issue (#6163)
3b59ee22a0 is described below
commit 3b59ee22a0504ec091b58864feb6600fe7ebca80
Author: Potato <[email protected]>
AuthorDate: Mon Jun 6 14:47:13 2022 +0800
[IOTDB-3395] Use thrift server to fix clientManagerTest "bind address
already used" issue (#6163)
[IOTDB-3395] Use thrift server to fix clientManagerTest "bind address
already used" issue
---
node-commons/pom.xml | 5 ++
.../commons/{ => client}/ClientManagerTest.java | 58 +++-----------
.../client/mock/MockInternalRPCService.java | 92 ++++++++++++++++++++++
.../client/mock/MockInternalRPCServiceMBean.java | 22 ++++++
4 files changed, 131 insertions(+), 46 deletions(-)
diff --git a/node-commons/pom.xml b/node-commons/pom.xml
index ef0ca10146..aee851e866 100644
--- a/node-commons/pom.xml
+++ b/node-commons/pom.xml
@@ -113,6 +113,11 @@
<artifactId>oauth2-oidc-sdk</artifactId>
<version>8.3</version>
</dependency>
+ <dependency>
+ <groupId>org.mockito</groupId>
+ <artifactId>mockito-core</artifactId>
+ <scope>test</scope>
+ </dependency>
</dependencies>
<build>
<plugins>
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..2f8f4cca56 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.sync.SyncDataNodeInternalServiceClient;
+import org.apache.iotdb.commons.exception.StartupException;
+import org.apache.iotdb.mpp.rpc.thrift.InternalService;
import org.apache.commons.pool2.KeyedObjectPool;
import org.apache.commons.pool2.impl.GenericKeyedObjectPool;
@@ -37,24 +35,25 @@ import org.junit.Before;
import org.junit.Test;
import java.io.IOException;
-import java.net.InetSocketAddress;
-import java.net.ServerSocket;
+
+import static org.mockito.Mockito.mock;
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(mock(InternalService.Iface.class));
+ service.start();
}
@After
public void tearDown() throws IOException, InterruptedException {
- stopServer();
+ service.waitAndStop(10_000L);
}
@Test
@@ -418,39 +417,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..9cf8ae042a
--- /dev/null
+++
b/node-commons/src/test/java/org/apache/iotdb/commons/client/mock/MockInternalRPCService.java
@@ -0,0 +1,92 @@
+/*
+ * 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;
+
+import org.apache.thrift.server.TServerEventHandler;
+
+import static org.mockito.Mockito.mock;
+
+public class MockInternalRPCService extends ThriftService implements
MockInternalRPCServiceMBean {
+
+ private final TEndPoint thisNode;
+ private InternalService.Iface mockedProcessor;
+
+ public MockInternalRPCService(TEndPoint thisNode) {
+ this.thisNode = thisNode;
+ }
+
+ @Override
+ public ServiceType getID() {
+ return ServiceType.INTERNAL_SERVICE;
+ }
+
+ @Override
+ public void initSyncedServiceImpl(Object mockedProcessor) {
+ this.mockedProcessor = (InternalService.Iface) mockedProcessor;
+ super.mbeanName =
+ String.format(
+ "%s:%s=%s", this.getClass().getPackage(), IoTDBConstant.JMX_TYPE,
getID().getJmxName());
+ super.initSyncedServiceImpl(this.mockedProcessor);
+ }
+
+ @Override
+ public void initTProcessor() {
+ processor = new InternalService.Processor<>(mockedProcessor);
+ }
+
+ @Override
+ public void initThriftServiceThread() throws IllegalAccessException {
+ try {
+ thriftServiceThread =
+ new ThriftServiceThread(
+ processor,
+ getID().getName(),
+ ThreadName.INTERNAL_SERVICE_RPC_SERVER.getName(),
+ getBindIP(),
+ getBindPort(),
+ 65535,
+ 60,
+ mock(TServerEventHandler.class),
+ 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 {}
