Browse Source

add ZkServer for UT

pull/2/head
dk.technoboy 5 years ago
parent
commit
2f3d270469
  1. 98
      dolphinscheduler-common/src/test/java/org/apache/dolphinscheduler/common/zk/StandaloneZKServerForTest.java
  2. 43
      dolphinscheduler-common/src/test/java/org/apache/dolphinscheduler/common/zk/TestZk.java
  3. 165
      dolphinscheduler-common/src/test/java/org/apache/dolphinscheduler/common/zk/ZKServer.java

98
dolphinscheduler-common/src/test/java/org/apache/dolphinscheduler/common/zk/StandaloneZKServerForTest.java

@ -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);
}
}
}

43
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();
}
}

165
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);
}
}
}
}
Loading…
Cancel
Save