diff --git a/dolphinscheduler-common/src/test/java/org/apache/dolphinscheduler/common/queue/TaskQueueImplTest.java b/dolphinscheduler-common/src/test/java/org/apache/dolphinscheduler/common/queue/TaskQueueImplTest.java index efee627676..14e90ebcdc 100644 --- a/dolphinscheduler-common/src/test/java/org/apache/dolphinscheduler/common/queue/TaskQueueImplTest.java +++ b/dolphinscheduler-common/src/test/java/org/apache/dolphinscheduler/common/queue/TaskQueueImplTest.java @@ -19,7 +19,7 @@ package org.apache.dolphinscheduler.common.queue; import org.apache.dolphinscheduler.common.Constants; import org.apache.dolphinscheduler.common.utils.IpUtils; import org.apache.dolphinscheduler.common.utils.OSUtils; -import org.apache.dolphinscheduler.common.zk.StandaloneZKServerForTest; +import org.apache.dolphinscheduler.common.zk.ZKServer; import org.junit.After; import org.junit.Before; import org.junit.Ignore; @@ -35,7 +35,7 @@ import static org.junit.Assert.assertEquals; /** * task queue test */ -public class TaskQueueImplTest extends StandaloneZKServerForTest { +public class TaskQueueImplTest { private static final Logger logger = LoggerFactory.getLogger(TaskQueueImplTest.class); @@ -43,7 +43,7 @@ public class TaskQueueImplTest extends StandaloneZKServerForTest { @Before public void before(){ - super.before(); + ZKServer.start(); tasksQueue = TaskQueueFactory.getTaskQueueInstance(); @@ -57,6 +57,7 @@ public class TaskQueueImplTest extends StandaloneZKServerForTest { public void after(){ //clear all data tasksQueue.delete(); + ZKServer.stop(); } diff --git a/dolphinscheduler-common/src/test/java/org/apache/dolphinscheduler/common/zk/StandaloneZKServerForTest.java b/dolphinscheduler-common/src/test/java/org/apache/dolphinscheduler/common/zk/StandaloneZKServerForTest.java deleted file mode 100644 index fed9ebb18a..0000000000 --- a/dolphinscheduler-common/src/test/java/org/apache/dolphinscheduler/common/zk/StandaloneZKServerForTest.java +++ /dev/null @@ -1,98 +0,0 @@ -/* - * 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.dolphinscheduler.common.zk; - -import java.io.File; -import java.util.Properties; - -import org.apache.dolphinscheduler.common.thread.ThreadPoolExecutors; -import org.apache.zookeeper.server.ServerConfig; -import org.apache.zookeeper.server.ZooKeeperServerMain; -import org.apache.zookeeper.server.quorum.QuorumPeerConfig; -import org.junit.Before; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; - - -/** - * just for test - */ -public class StandaloneZKServerForTest { - - private static final Logger logger = LoggerFactory.getLogger(StandaloneZKServerForTest.class); - - private static volatile ZooKeeperServerMain zkServer = null; - - - @Before - public void before() { - logger.info("standalone zookeeper server for test service start "); - - ThreadPoolExecutors.getInstance().execute(new Runnable() { - @Override - public void run() { - - //delete zk data dir ? - File zkFile = new File(System.getProperty("java.io.tmpdir"), "zookeeper"); - - startStandaloneServer("2000", zkFile.getAbsolutePath(), "2181", "10", "5"); - } - }); - - } - - - /** - * start zk server - * @param tickTime zookeeper ticktime - * @param dataDir zookeeper data dir - * @param clientPort zookeeper client port - * @param initLimit zookeeper init limit - * @param syncLimit zookeeper sync limit - */ - private void startStandaloneServer(String tickTime, String dataDir, String clientPort, String initLimit, String syncLimit) { - Properties props = new Properties(); - props.setProperty("tickTime", tickTime); - props.setProperty("dataDir", dataDir); - props.setProperty("clientPort", clientPort); - props.setProperty("initLimit", initLimit); - props.setProperty("syncLimit", syncLimit); - - QuorumPeerConfig quorumConfig = new QuorumPeerConfig(); - try { - quorumConfig.parseProperties(props); - - if(zkServer == null ){ - - synchronized (StandaloneZKServerForTest.class){ - if(zkServer == null ){ - zkServer = new ZooKeeperServerMain(); - final ServerConfig config = new ServerConfig(); - config.readFrom(quorumConfig); - zkServer.runFromConfig(config); - } - } - - } - - } catch (Exception e) { - logger.error("start standalone server fail！", e); - } - } - - -} \ No newline at end of file diff --git a/dolphinscheduler-common/src/test/java/org/apache/dolphinscheduler/common/zk/TestZk.java b/dolphinscheduler-common/src/test/java/org/apache/dolphinscheduler/common/zk/TestZk.java new file mode 100644 index 0000000000..5c3db2d5d1 --- /dev/null +++ b/dolphinscheduler-common/src/test/java/org/apache/dolphinscheduler/common/zk/TestZk.java @@ -0,0 +1,43 @@ +/* + * 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.dolphinscheduler.common.zk; + +import org.junit.After; +import org.junit.Assert; +import org.junit.Before; +import org.junit.Test; + +/** + * demo for using zkServer + */ +public class TestZk { + + @Before + public void before(){ + ZKServer.start(); + } + + @Test + public void test(){ + Assert.assertTrue(ZKServer.isStarted()); + } + + @After + public void after(){ + ZKServer.stop(); + } +} diff --git a/dolphinscheduler-common/src/test/java/org/apache/dolphinscheduler/common/zk/ZKServer.java b/dolphinscheduler-common/src/test/java/org/apache/dolphinscheduler/common/zk/ZKServer.java new file mode 100644 index 0000000000..5aba9fd8a1 --- /dev/null +++ b/dolphinscheduler-common/src/test/java/org/apache/dolphinscheduler/common/zk/ZKServer.java @@ -0,0 +1,165 @@ +/* + * 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.dolphinscheduler.common.zk; + +import org.apache.curator.framework.CuratorFramework; +import org.apache.curator.framework.CuratorFrameworkFactory; +import org.apache.curator.retry.ExponentialBackoffRetry; +import org.apache.zookeeper.server.ZooKeeperServerMain; +import org.apache.zookeeper.server.quorum.QuorumPeerConfig; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.io.File; +import java.io.IOException; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; + + +/** + * just for test + */ +public class ZKServer { + + private static final Logger logger = LoggerFactory.getLogger(ZKServer.class); + + private static volatile PublicZooKeeperServerMain zkServer = null; + + public static final int DEFAULT_ZK_TEST_PORT = 22181; + + public static final String DEFAULT_ZK_STR = "localhost:" + DEFAULT_ZK_TEST_PORT; + + private static String dataDir = null; + + private static final AtomicBoolean isStarted = new AtomicBoolean(false); + + public static void start() { + try { + startLocalZkServer(DEFAULT_ZK_TEST_PORT); + } catch (Exception e) { + logger.error("Failed to start ZK: " + e); + } + } + + public static boolean isStarted(){ + return isStarted.get(); + } + + static class PublicZooKeeperServerMain extends ZooKeeperServerMain { + + @Override + public void initializeAndRun(String[] args) + throws QuorumPeerConfig.ConfigException, IOException { + super.initializeAndRun(args); + } + + @Override + public void shutdown() { + super.shutdown(); + } + } + + /** + * Starts a local Zk instance with a generated empty data directory + * + * @param port The port to listen on + */ + public static void startLocalZkServer(final int port) { + startLocalZkServer(port, org.apache.commons.io.FileUtils.getTempDirectoryPath() + File.separator + "test-" + System.currentTimeMillis()); + } + + /** + * Starts a local Zk instance + * + * @param port The port to listen on + * @param dataDirPath The path for the Zk data directory + */ + private static synchronized void startLocalZkServer(final int port, final String dataDirPath) { + if (zkServer != null) { + throw new RuntimeException("Zookeeper server is already started!"); + } + try { + zkServer = new PublicZooKeeperServerMain(); + logger.info("Zookeeper data path : {} ", dataDirPath); + dataDir = dataDirPath; + final String[] args = new String[]{Integer.toString(port), dataDirPath}; + Thread init = new Thread(new Runnable() { + @Override + public void run() { + try { + zkServer.initializeAndRun(args); + } catch (QuorumPeerConfig.ConfigException e) { + logger.warn("Caught exception while starting ZK", e); + } catch (IOException e) { + logger.warn("Caught exception while starting ZK", e); + } + } + }, "init-zk-thread"); + init.start(); + } catch (Exception e) { + logger.warn("Caught exception while starting ZK", e); + throw new RuntimeException(e); + } + + CuratorFramework zkClient = CuratorFrameworkFactory.builder() + .connectString(DEFAULT_ZK_STR) + .retryPolicy(new ExponentialBackoffRetry(10,100)) + .sessionTimeoutMs(1000 * 30) + .connectionTimeoutMs(1000 * 30) + .build(); + + try { + zkClient.blockUntilConnected(10, TimeUnit.SECONDS); + zkClient.close(); + } catch (InterruptedException ignore) { + } + isStarted.compareAndSet(false, true); + logger.info("zk server started"); + } + + /** + * Stops a local Zk instance, deleting its data directory + */ + public static void stop() { + try { + stopLocalZkServer(true); + } catch (Exception e) { + logger.error("Failed to stop ZK ",e); + } + } + + /** + * Stops a local Zk instance. + * + * @param deleteDataDir Whether or not to delete the data directory + */ + private static synchronized void stopLocalZkServer(final boolean deleteDataDir) { + if (zkServer != null) { + try { + zkServer.shutdown(); + zkServer = null; + if (deleteDataDir) { + org.apache.commons.io.FileUtils.deleteDirectory(new File(dataDir)); + } + isStarted.compareAndSet(true, false); + } catch (Exception e) { + logger.warn("Caught exception while stopping ZK server", e); + throw new RuntimeException(e); + } + } + } +} \ No newline at end of file