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 <[email protected]> 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) {} +}
