Skip to content

Commit

Permalink
KAA-1281: kaa node blocks startup if zookeeper is unavailable
Browse files Browse the repository at this point in the history
  • Loading branch information
@nocs00
nocs00 committed Sep 6, 2016
1 parent f076993 commit da29bc9
Show file tree
Hide file tree
Showing 16 changed files with 411 additions and 352 deletions.
44 changes: 7 additions & 37 deletions server/common/zk/src/main/java/org/kaaproject/kaa/server/common/zk/ControlNodeTracker.java
View file
Expand Up @@ -21,9 +21,7 @@
import java.util.List;
import java.util.concurrent.CopyOnWriteArrayList;

import org.apache.curator.RetryPolicy;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.api.UnhandledErrorListener;
import org.apache.curator.framework.recipes.cache.ChildData;
import org.apache.curator.framework.recipes.cache.NodeCache;
Expand Down Expand Up @@ -56,7 +54,7 @@ public abstract class ControlNodeTracker implements ControlNodeAware, Closeable
.getLogger(ControlNodeTracker.class);

/** The client. */
protected CuratorFramework client;
protected CuratorFramework zkClient;

/** The control cache. */
private NodeCache controlCache;
Expand Down Expand Up @@ -107,34 +105,9 @@ public void unhandledError(String message, Throwable e) {
/**
* Instantiates a new control node tracker.
*
* @param zkHostPortList
* the zk host port list
* @param retryPolicy
* the retry policy
*/
public ControlNodeTracker(String zkHostPortList, RetryPolicy retryPolicy) {
this(zkHostPortList, -1, -1, retryPolicy);
}

/**
* Instantiates a new control node tracker.
*
* @param zkHostPortList
* the zk host port list
* @param sessionTimeoutMs
* session timeout
* @param connectionTimeoutMs
* connection timeout
* @param retryPolicy
* the retry policy
*/
public ControlNodeTracker(String zkHostPortList, int sessionTimeoutMs, int connectionTimeoutMs, RetryPolicy retryPolicy) {
public ControlNodeTracker() {
super();
if (sessionTimeoutMs > -1 && connectionTimeoutMs > -1) {
this.client = CuratorFrameworkFactory.newClient(zkHostPortList, sessionTimeoutMs, connectionTimeoutMs, retryPolicy);
} else {
this.client = CuratorFrameworkFactory.newClient(zkHostPortList, retryPolicy);
}
this.listeners = new CopyOnWriteArrayList<ControlNodeListener>();
}

Expand All @@ -146,10 +119,9 @@ public ControlNodeTracker(String zkHostPortList, int sessionTimeoutMs, int conne
*/
public void start() throws Exception { //NOSONAR
LOG.info("Starting node tracker");
client.start();
client.getUnhandledErrorListenable().addListener(errorsListener);
zkClient.getUnhandledErrorListenable().addListener(errorsListener);
if(createZkNode()){
controlCache = new NodeCache(client, CONTROL_SERVER_NODE_PATH);
controlCache = new NodeCache(zkClient, CONTROL_SERVER_NODE_PATH);
controlCache.getListenable().addListener(new NodeCacheListener() {

@Override
Expand Down Expand Up @@ -204,7 +176,7 @@ protected void onMasterChange(ChildData currentData) {
* @return true, if is connected
*/
public boolean isConnected() {
return client.getZookeeperClient().isConnected();
return zkClient.getZookeeperClient().isConnected();
}

/**
Expand Down Expand Up @@ -250,14 +222,12 @@ public void close() throws IOException {

if(nodePath != null){
try {
client.delete().forPath(nodePath);
zkClient.delete().forPath(nodePath);
LOG.debug("Node with path {} successfully deleted", nodePath);
} catch (Exception e) {
LOG.debug("Failed to delete node", e);
}
}

client.close();
}

/*
Expand Down Expand Up @@ -299,7 +269,7 @@ public boolean doZKClientAction(ZKClientAction action) throws IOException{

public boolean doZKClientAction(ZKClientAction action, boolean throwIOException) throws IOException{
try{
action.doWithZkClient(client);
action.doWithZkClient(zkClient);
return true;
} catch (Exception e) {
LOG.error("Unknown Error", e);
Expand Down
28 changes: 8 additions & 20 deletions server/common/zk/src/main/java/org/kaaproject/kaa/server/common/zk/WorkerNodeTracker.java
View file
Expand Up @@ -20,7 +20,6 @@
import java.util.*;
import java.util.concurrent.CopyOnWriteArrayList;

import org.apache.curator.RetryPolicy;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.recipes.cache.ChildData;
import org.apache.curator.framework.recipes.cache.PathChildrenCache;
Expand Down Expand Up @@ -65,36 +64,25 @@ public abstract class WorkerNodeTracker extends ControlNodeTracker {
/**
* Instantiates a new worker node tracker.
*
* @param zkHostPortList
* the zk host port list
* @param retryPolicy
* the retry policy
*/
public WorkerNodeTracker(String zkHostPortList, RetryPolicy retryPolicy) {
super(zkHostPortList, retryPolicy);
public WorkerNodeTracker() {
super();
init();
}

/**
* Instantiates a new worker node tracker.
*
* @param zkHostPortList
* the zk host port list
* @param sessionTimeoutMs
* the session timeout
* @param connectionTimeoutMs
* the connection timeout
* @param retryPolicy
* the retry policy
* @param zkClient Zookeeper client
*/
public WorkerNodeTracker(String zkHostPortList, int sessionTimeoutMs, int connectionTimeoutMs, RetryPolicy retryPolicy) {
super(zkHostPortList, sessionTimeoutMs, connectionTimeoutMs, retryPolicy);
public WorkerNodeTracker(CuratorFramework zkClient) {
super();
this.zkClient = zkClient;
init();
}

private void init() {
endpointCache = new PathChildrenCache(client, OPERATIONS_SERVER_NODE_PATH, true);
bootstrapCache = new PathChildrenCache(client, BOOTSTRAP_SERVER_NODE_PATH, true);
endpointCache = new PathChildrenCache(zkClient, OPERATIONS_SERVER_NODE_PATH, true);
bootstrapCache = new PathChildrenCache(zkClient, BOOTSTRAP_SERVER_NODE_PATH, true);
endpointListeners = new CopyOnWriteArrayList<OperationsNodeListener>();
bootstrapListeners = new CopyOnWriteArrayList<BootstrapNodeListener>();
operationNodesStartTimes = new HashMap<String, Long>();
Expand Down
18 changes: 13 additions & 5 deletions .../common/zk/src/main/java/org/kaaproject/kaa/server/common/zk/bootstrap/BootstrapNode.java
View file
Expand Up @@ -43,12 +43,20 @@ public class BootstrapNode extends WorkerNodeTracker {
* Instantiates a new bootstrap node.
*
* @param nodeInfo the node info
* @param zkHostPortList the zk host port list
* @param retryPolicy the retry policy
*/
public BootstrapNode(BootstrapNodeInfo nodeInfo, String zkHostPortList,
RetryPolicy retryPolicy) {
super(zkHostPortList, retryPolicy);
public BootstrapNode(BootstrapNodeInfo nodeInfo) {
super();
this.nodeInfo = nodeInfo;
}

/**
* Instantiates a new bootstrap node.
*
* @param nodeInfo the node info
* @param zkClient Zookeeper client
*/
public BootstrapNode(BootstrapNodeInfo nodeInfo, CuratorFramework zkClient) {
super(zkClient);
this.nodeInfo = nodeInfo;
}

Expand Down
23 changes: 6 additions & 17 deletions server/common/zk/src/main/java/org/kaaproject/kaa/server/common/zk/control/ControlNode.java
View file
Expand Up @@ -46,13 +46,9 @@ public class ControlNode extends WorkerNodeTracker {
*
* @param currentNodeInfo
* the current node info
* @param zkHostPortList
* the zk host port list
* @param retryPolicy
* the retry policy
*/
public ControlNode(ControlNodeInfo currentNodeInfo, String zkHostPortList, RetryPolicy retryPolicy) {
super(zkHostPortList, retryPolicy);
public ControlNode(ControlNodeInfo currentNodeInfo) {
super();
this.currentNodeInfo = currentNodeInfo;
}

Expand All @@ -61,17 +57,10 @@ public ControlNode(ControlNodeInfo currentNodeInfo, String zkHostPortList, Retry
*
* @param currentNodeInfo
* the current node info
* @param zkHostPortList
* the zk host port list
* @param sessionTimeoutMs
* session timeout
* @param connectionTimeoutMs
* connection timeout
* @param retryPolicy
* the retry policy
* @param zkClient Zookeeper client
*/
public ControlNode(ControlNodeInfo currentNodeInfo, String zkHostPortList, int sessionTimeoutMs, int connectionTimeoutMs, RetryPolicy retryPolicy) {
super(zkHostPortList, sessionTimeoutMs, connectionTimeoutMs, retryPolicy);
public ControlNode(ControlNodeInfo currentNodeInfo, CuratorFramework zkClient) {
super(zkClient);
this.currentNodeInfo = currentNodeInfo;
}

Expand Down Expand Up @@ -127,7 +116,7 @@ public ControlNodeInfo getCurrentNodeInfo() {
@Override
public boolean createZkNode() throws IOException {
try {
nodePath = client.create().withMode(CreateMode.EPHEMERAL)
nodePath = zkClient.create().withMode(CreateMode.EPHEMERAL)
.forPath(ControlNodeTracker.CONTROL_SERVER_NODE_PATH, controlNodeAvroConverter.get().toByteArray(currentNodeInfo));
LOG.info("Created node with path: " + nodePath);
} catch (NodeExistsException e) {
Expand Down
21 changes: 15 additions & 6 deletions ...ommon/zk/src/main/java/org/kaaproject/kaa/server/common/zk/operations/OperationsNode.java
View file
Expand Up @@ -43,13 +43,22 @@ public class OperationsNode extends WorkerNodeTracker {
*
* @param nodeInfo
* the node info
* @param zkHostPortList
* the zk host port list
* @param retryPolicy
* the retry policy
*/
public OperationsNode(OperationsNodeInfo nodeInfo, String zkHostPortList, RetryPolicy retryPolicy) {
super(zkHostPortList, retryPolicy);
public OperationsNode(OperationsNodeInfo nodeInfo) {
super();
this.nodeInfo = nodeInfo;
this.nodeInfo.setTimeStarted(System.currentTimeMillis());
}

/**
* Instantiates a new endpoint node.
*
* @param nodeInfo
* the node info
* @param zkClient Zookeeper client
*/
public OperationsNode(OperationsNodeInfo nodeInfo, CuratorFramework zkClient) {
super(zkClient);
this.nodeInfo = nodeInfo;
this.nodeInfo.setTimeStarted(System.currentTimeMillis());
}
Expand Down

0 comments on commit da29bc9

Please sign in to comment.