@Test public void testInitializedNodeStatusShouldBeGroupInitialize() throws Exception { when(configurationContext.isClusteringEnabled()).thenReturn(false); clusterSyncManager = new ClusterSyncManager(STREAMING.ZK_CLUSTER_MANAGER_PATH, "id", configurationContext, failOverTask,curatorFramework, zkUtils, clusterBarrierManager); String nodeStatusPath = clusterSyncManager.initializedNodeStatus(); assertNotNull(nodeStatusPath); assertEquals(STREAMING.ZK_EPHEMERAL_NODE_STATUS_PATH,nodeStatusPath); }
public String initializedNodeStatus() throws Exception { String nodeStatusPath; if (!clusteringEnabled){ nodeStatusPath = STREAMING.ZK_EPHEMERAL_NODE_STATUS_PATH; } else { nodeStatusPath = STREAMING.ZK_EPHEMERAL_GROUPS_STATUS_BASE_PATH.concat("/").concat(STREAMING.GROUPS_STATUS_BASE_PREFIX) .concat(groupId); } zkUtils.createEphemeralZNode(nodeStatusPath, STREAMING.ZK_EPHEMERAL_NODE_STATUS_INITIALIZED.getBytes()); return nodeStatusPath; }
ClusterSyncManager { public String initializedNodeStatus() throws Exception { String nodeStatusPath; if (!clusteringEnabled){ nodeStatusPath = STREAMING.ZK_EPHEMERAL_NODE_STATUS_PATH; } else { nodeStatusPath = STREAMING.ZK_EPHEMERAL_GROUPS_STATUS_BASE_PATH.concat("/").concat(STREAMING.GROUPS_STATUS_BASE_PREFIX) .concat(groupId); } zkUtils.createEphemeralZNode(nodeStatusPath, STREAMING.ZK_EPHEMERAL_NODE_STATUS_INITIALIZED.getBytes()); return nodeStatusPath; } }
ClusterSyncManager { public String initializedNodeStatus() throws Exception { String nodeStatusPath; if (!clusteringEnabled){ nodeStatusPath = STREAMING.ZK_EPHEMERAL_NODE_STATUS_PATH; } else { nodeStatusPath = STREAMING.ZK_EPHEMERAL_GROUPS_STATUS_BASE_PATH.concat("/").concat(STREAMING.GROUPS_STATUS_BASE_PREFIX) .concat(groupId); } zkUtils.createEphemeralZNode(nodeStatusPath, STREAMING.ZK_EPHEMERAL_NODE_STATUS_INITIALIZED.getBytes()); return nodeStatusPath; } ClusterSyncManager(String latchpath, String id, ConfigurationContext
            configurationContext, FailOverTask failOverTask, CuratorFramework curatorClient, ZKUtils zkUtils,
            ClusterBarrierManager clusterBarrierManager);  ClusterSyncManager(String latchpath, String id, ConfigurationContext
            configurationContext, FailOverTask failOverTask); }
ClusterSyncManager { public String initializedNodeStatus() throws Exception { String nodeStatusPath; if (!clusteringEnabled){ nodeStatusPath = STREAMING.ZK_EPHEMERAL_NODE_STATUS_PATH; } else { nodeStatusPath = STREAMING.ZK_EPHEMERAL_GROUPS_STATUS_BASE_PATH.concat("/").concat(STREAMING.GROUPS_STATUS_BASE_PREFIX) .concat(groupId); } zkUtils.createEphemeralZNode(nodeStatusPath, STREAMING.ZK_EPHEMERAL_NODE_STATUS_INITIALIZED.getBytes()); return nodeStatusPath; } ClusterSyncManager(String latchpath, String id, ConfigurationContext
            configurationContext, FailOverTask failOverTask, CuratorFramework curatorClient, ZKUtils zkUtils,
            ClusterBarrierManager clusterBarrierManager);  ClusterSyncManager(String latchpath, String id, ConfigurationContext
            configurationContext, FailOverTask failOverTask); static ClusterSyncManager getClusterSyncManager(ConfigurationContext configurationContext,
            FailOverTask failOverTask); static ClusterSyncManager getNode(); void start(); boolean isStarted(); ActionCallbackDto manageAckStreamingOperation(StratioStreamingMessage message, ActionCallbackDto reply); String initializedNodeStatus(); String initializedGroupStatus(); String updateNodeStatus(String nodeId, PathChildrenCacheEvent.Type eventType); boolean isLeader(); CuratorFramework getClient(); String getGroupId(); }
ClusterSyncManager { public String initializedNodeStatus() throws Exception { String nodeStatusPath; if (!clusteringEnabled){ nodeStatusPath = STREAMING.ZK_EPHEMERAL_NODE_STATUS_PATH; } else { nodeStatusPath = STREAMING.ZK_EPHEMERAL_GROUPS_STATUS_BASE_PATH.concat("/").concat(STREAMING.GROUPS_STATUS_BASE_PREFIX) .concat(groupId); } zkUtils.createEphemeralZNode(nodeStatusPath, STREAMING.ZK_EPHEMERAL_NODE_STATUS_INITIALIZED.getBytes()); return nodeStatusPath; } ClusterSyncManager(String latchpath, String id, ConfigurationContext
            configurationContext, FailOverTask failOverTask, CuratorFramework curatorClient, ZKUtils zkUtils,
            ClusterBarrierManager clusterBarrierManager);  ClusterSyncManager(String latchpath, String id, ConfigurationContext
            configurationContext, FailOverTask failOverTask); static ClusterSyncManager getClusterSyncManager(ConfigurationContext configurationContext,
            FailOverTask failOverTask); static ClusterSyncManager getNode(); void start(); boolean isStarted(); ActionCallbackDto manageAckStreamingOperation(StratioStreamingMessage message, ActionCallbackDto reply); String initializedNodeStatus(); String initializedGroupStatus(); String updateNodeStatus(String nodeId, PathChildrenCacheEvent.Type eventType); boolean isLeader(); CuratorFramework getClient(); String getGroupId();  }