[ISSUE #220] Add unit test(controller module)

pom.xml

@@ -50,7 +50,7 @@
         <slf4j.version>1.7.7</slf4j.version>
         <logback.version>1.2.9</logback.version>
         <commons.cli.version>1.2</commons.cli.version>
-        <reflections.version>0.9.12</reflections.version>
+        <reflections.version>0.10.2</reflections.version>
         <guava.version>22.0</guava.version>
 
         <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>

rocketmq-connect-runtime/src/main/java/org/apache/rocketmq/connect/runtime/controller/isolation/DelegatingClassLoader.java

@@ -298,7 +298,7 @@ private PluginScanResult doLoad(
         builder.setClassLoaders(new ClassLoader[]{loader});
         builder.addUrls(urls);
         builder.setScanners(new SubTypesScanner());
-        builder.useParallelExecutor();
+        builder.setParallel(true);
         Reflections reflections = new InternalReflections(builder);
 
         return new PluginScanResult(
@@ -413,20 +413,6 @@ private static class InternalReflections extends Reflections {
         public InternalReflections(Configuration configuration) {
             super(configuration);
         }
-
-        // When Reflections is used for parallel scans, it has a bug where it propagates ReflectionsException
-        // as RuntimeException.  Override the scan behavior to emulate the singled-threaded logic.
-        @Override
-        protected void scan(URL url) {
-            try {
-                super.scan(url);
-            } catch (ReflectionsException e) {
-                Logger log = Reflections.log;
-                if (log != null && log.isWarnEnabled()) {
-                    log.warn("could not create Vfs.Dir from url. ignoring the exception and continuing", e);
-                }
-            }
-        }
     }
 
 }

rocketmq-connect-runtime/src/test/java/org/apache/rocketmq/connect/runtime/connectorwrapper/NameServerMocker.java

@@ -0,0 +1,66 @@
+/*
+ * 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.rocketmq.connect.runtime.connectorwrapper;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import org.apache.rocketmq.common.MixAll;
+import org.apache.rocketmq.common.protocol.route.BrokerData;
+import org.apache.rocketmq.common.protocol.route.TopicRouteData;
+
+/**
+ * tools class
+ */
+public class NameServerMocker {
+
+    /**
+     * use the specified port to start the nameserver
+     *
+     * @param nameServerPort    nameServer port
+     * @param brokerPort    broker port
+     * @return  ServerResponseMocker
+     */
+    public static ServerResponseMocker startByDefaultConf(int nameServerPort, int brokerPort) {
+        return startByDefaultConf(nameServerPort, brokerPort, null);
+    }
+
+    /**
+     * use the specified port to start the nameserver
+     *
+     * @param nameServerPort    nameServer port
+     * @param brokerPort    broker port
+     * @param extMap    extend config
+     * @return  ServerResponseMocker
+     */
+    public static ServerResponseMocker startByDefaultConf(int nameServerPort, int brokerPort,
+                                                          HashMap<String, String> extMap) {
+
+        System.setProperty(MixAll.NAMESRV_ADDR_PROPERTY, "127.0.0.1:" + nameServerPort);
+        TopicRouteData topicRouteData = new TopicRouteData();
+        List<BrokerData> dataList = new ArrayList<>();
+        HashMap<Long, String> brokerAddress = new HashMap<>();
+        brokerAddress.put(1L, "127.0.0.1:" + brokerPort);
+        BrokerData brokerData = new BrokerData("mockCluster", "mockBrokerName", brokerAddress);
+        brokerData.setBrokerName("mockBrokerName");
+        dataList.add(brokerData);
+        topicRouteData.setBrokerDatas(dataList);
+        // start name server
+        return ServerResponseMocker.startServer(nameServerPort, topicRouteData.encode(), extMap);
+    }
+
+}

rocketmq-connect-runtime/src/test/java/org/apache/rocketmq/connect/runtime/connectorwrapper/ServerResponseMocker.java

@@ -0,0 +1,218 @@
+/*
+ * 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.rocketmq.connect.runtime.connectorwrapper;
+
+import com.alibaba.fastjson.JSON;
+import io.netty.bootstrap.ServerBootstrap;
+import io.netty.channel.ChannelDuplexHandler;
+import io.netty.channel.ChannelFuture;
+import io.netty.channel.ChannelHandler;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.ChannelInitializer;
+import io.netty.channel.ChannelOption;
+import io.netty.channel.SimpleChannelInboundHandler;
+import io.netty.channel.nio.NioEventLoopGroup;
+import io.netty.channel.socket.SocketChannel;
+import io.netty.channel.socket.nio.NioServerSocketChannel;
+import io.netty.handler.timeout.IdleStateHandler;
+import io.netty.util.concurrent.Future;
+import java.net.InetSocketAddress;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ExecutionException;
+import org.apache.rocketmq.common.DataVersion;
+import org.apache.rocketmq.common.MixAll;
+import org.apache.rocketmq.common.protocol.RequestCode;
+import org.apache.rocketmq.common.protocol.body.ClusterInfo;
+import org.apache.rocketmq.common.protocol.body.SubscriptionGroupWrapper;
+import org.apache.rocketmq.common.protocol.route.BrokerData;
+import org.apache.rocketmq.common.subscription.SubscriptionGroupConfig;
+import org.apache.rocketmq.remoting.netty.NettyDecoder;
+import org.apache.rocketmq.remoting.netty.NettyEncoder;
+import org.apache.rocketmq.remoting.protocol.RemotingCommand;
+import org.apache.rocketmq.remoting.protocol.RemotingSysResponseCode;
+import org.junit.After;
+import org.junit.Before;
+
+/**
+ * mock server response for command
+ */
+public abstract class ServerResponseMocker {
+
+    private final NioEventLoopGroup eventLoopGroup = new NioEventLoopGroup();
+
+    @Before
+    public void before() {
+        start();
+    }
+
+    @After
+    public void shutdown() {
+        if (eventLoopGroup.isShutdown()) {
+            return;
+        }
+        Future<?> future = eventLoopGroup.shutdownGracefully();
+        try {
+            future.get();
+        } catch (InterruptedException | ExecutionException e) {
+            e.printStackTrace();
+        }
+    }
+
+    protected abstract int getPort();
+
+    protected abstract byte[] getBody();
+
+    public void start() {
+        start(null);
+    }
+
+    public void start(HashMap<String, String> extMap) {
+        ServerBootstrap serverBootstrap = new ServerBootstrap();
+        serverBootstrap.group(eventLoopGroup)
+                .channel(NioServerSocketChannel.class)
+                .option(ChannelOption.SO_BACKLOG, 1024)
+                .option(ChannelOption.SO_REUSEADDR, true)
+                .option(ChannelOption.SO_KEEPALIVE, false)
+                .childOption(ChannelOption.TCP_NODELAY, true)
+                .childOption(ChannelOption.SO_SNDBUF, 65535)
+                .childOption(ChannelOption.SO_RCVBUF, 65535)
+                .localAddress(new InetSocketAddress(getPort()))
+                .childHandler(new ChannelInitializer<SocketChannel>() {
+                    @Override
+                    public void initChannel(SocketChannel ch) throws Exception {
+                        ch.pipeline()
+                                .addLast(eventLoopGroup,
+                                        new NettyEncoder(),
+                                        new NettyDecoder(),
+                                        new IdleStateHandler(0, 0, 120),
+                                        new ChannelDuplexHandler(),
+                                        new NettyServerHandler(extMap)
+                                );
+                    }
+                });
+        try {
+            ChannelFuture sync = serverBootstrap.bind().sync();
+            InetSocketAddress addr = (InetSocketAddress) sync.channel().localAddress();
+        } catch (InterruptedException e1) {
+            throw new RuntimeException("this.serverBootstrap.bind().sync() InterruptedException", e1);
+        }
+    }
+
+    @ChannelHandler.Sharable
+    private class NettyServerHandler extends SimpleChannelInboundHandler<RemotingCommand> {
+        private HashMap<String, String> extMap;
+
+        public NettyServerHandler(HashMap<String, String> extMap) {
+            this.extMap = extMap;
+        }
+
+        @Override
+        protected void channelRead0(ChannelHandlerContext ctx, RemotingCommand msg) throws Exception {
+            String remark = "mock data";
+            final RemotingCommand response =
+                    RemotingCommand.createResponseCommand(RemotingSysResponseCode.SUCCESS, remark);
+            response.setOpaque(msg.getOpaque());
+            response.setBody(getBody());
+
+            switch (msg.getCode()) {
+                case RequestCode.GET_BROKER_CLUSTER_INFO: {
+                    final ClusterInfo clusterInfo = buildClusterInfo();
+                    response.setBody(JSON.toJSONBytes(clusterInfo));
+                    break;
+                }
+
+                case RequestCode.GET_ALL_SUBSCRIPTIONGROUP_CONFIG: {
+                    final SubscriptionGroupWrapper wrapper = buildSubscriptionGroupWrapper();
+                    response.setBody(JSON.toJSONBytes(wrapper));
+                    break;
+                }
+                default:
+                    break;
+            }
+
+            if (extMap != null && extMap.size() > 0) {
+                response.setExtFields(extMap);
+            }
+            ctx.writeAndFlush(response);
+        }
+    }
+
+    public static ServerResponseMocker startServer(int port, byte[] body) {
+        return startServer(port, body, null);
+    }
+
+
+    public static ServerResponseMocker startServer(int port, byte[] body, HashMap<String, String> extMap) {
+        ServerResponseMocker mocker = new ServerResponseMocker() {
+            @Override
+            protected int getPort() {
+                return port;
+            }
+
+            @Override
+            protected byte[] getBody() {
+                return body;
+            }
+        };
+        mocker.start(extMap);
+        // add jvm hook, close connection when jvm down
+        Runtime.getRuntime().addShutdownHook(new Thread(mocker::shutdown));
+        return mocker;
+    }
+
+    private ClusterInfo buildClusterInfo() {
+        ClusterInfo clusterInfo = new ClusterInfo();
+        HashMap<String, Set<String>> clusterAddrTable = new HashMap<>();
+        Set<String> brokerNames = new HashSet<>();
+        brokerNames.add("mockBrokerName");
+        clusterAddrTable.put("mockCluster", brokerNames);
+        clusterInfo.setClusterAddrTable(clusterAddrTable);
+        HashMap<String, BrokerData> brokerAddrTable = new HashMap<String, BrokerData>();
+
+        //build brokerData
+        BrokerData brokerData = new BrokerData();
+        brokerData.setBrokerName("mockBrokerName");
+        brokerData.setCluster("mockCluster");
+
+        //build brokerAddrs
+        HashMap<Long, String> brokerAddrs = new HashMap<Long, String>();
+        brokerAddrs.put(MixAll.MASTER_ID, "127.0.0.1:10911");
+
+        brokerData.setBrokerAddrs(brokerAddrs);
+        brokerAddrTable.put("master", brokerData);
+        clusterInfo.setBrokerAddrTable(brokerAddrTable);
+        return clusterInfo;
+    }
+
+    private SubscriptionGroupWrapper buildSubscriptionGroupWrapper() {
+        SubscriptionGroupWrapper subscriptionGroupWrapper = new SubscriptionGroupWrapper();
+        ConcurrentHashMap<String, SubscriptionGroupConfig> subscriptions = new ConcurrentHashMap<>();
+        SubscriptionGroupConfig subscriptionGroupConfig = new SubscriptionGroupConfig();
+        subscriptionGroupConfig.setConsumeBroadcastEnable(true);
+        subscriptionGroupConfig.setBrokerId(0);
+        subscriptionGroupConfig.setGroupName("Consumer-group-one");
+        subscriptions.put("Consumer-group-one", subscriptionGroupConfig);
+        subscriptionGroupWrapper.setSubscriptionGroupTable(subscriptions);
+        DataVersion dataVersion = new DataVersion();
+        dataVersion.nextVersion();
+        subscriptionGroupWrapper.setDataVersion(dataVersion);
+        return subscriptionGroupWrapper;
+    }
+}