@Test public void testManageAckStreamingOperationNoClustering() throws Exception { when(configurationContext.isClusteringEnabled()).thenReturn(false); final ClusterSyncManager partiallyMockedClusterSyncManager = PowerMockito.spy(new ClusterSyncManager(STREAMING .ZK_CLUSTER_MANAGER_PATH, "id", configurationContext, failOverTask, curatorFramework, zkUtils, clusterBarrierManager)); doNothing().when(zkUtils).createZNodeJsonReply(any(StratioStreamingMessage.class), any(Object.class)); ActionCallbackDto reply = partiallyMockedClusterSyncManager.manageAckStreamingOperation(message, nodeReply); assertEquals(nodeReply, reply); }
public ActionCallbackDto manageAckStreamingOperation(StratioStreamingMessage message, ActionCallbackDto reply) { ActionCallbackDto ackReply = reply; try { if (!clusteringEnabled){ zkUtils.createZNodeJsonReply(message, reply); } else { if (allAckEnabled) { zkUtils.createTempZNodeJsonReply(message, reply, groupId); String path = zkUtils.getTempZNodeJsonReplyPath(message); String barrierPath = path.concat("/").concat(BARRIER_RELATIVE_PATH); if (this.clusterBarrierManager==null){ clusterBarrierManager = new ClusterBarrierManager(this,ackTimeout); } boolean success = clusterBarrierManager.manageAckBarrier(path, clusterNodes.size()); if (isLeader()) { ackReply = manageBarrierResults(message, reply, path, success); } } else { if (isLeader()) { zkUtils.createZNodeJsonReply(message, reply); } } } }catch (Exception e){ logger.error("Exception managing the ack of the group {} for the request {}: {}", groupId, message .getRequest_id(), e .getMessage()); } return ackReply; }
ClusterSyncManager { public ActionCallbackDto manageAckStreamingOperation(StratioStreamingMessage message, ActionCallbackDto reply) { ActionCallbackDto ackReply = reply; try { if (!clusteringEnabled){ zkUtils.createZNodeJsonReply(message, reply); } else { if (allAckEnabled) { zkUtils.createTempZNodeJsonReply(message, reply, groupId); String path = zkUtils.getTempZNodeJsonReplyPath(message); String barrierPath = path.concat("/").concat(BARRIER_RELATIVE_PATH); if (this.clusterBarrierManager==null){ clusterBarrierManager = new ClusterBarrierManager(this,ackTimeout); } boolean success = clusterBarrierManager.manageAckBarrier(path, clusterNodes.size()); if (isLeader()) { ackReply = manageBarrierResults(message, reply, path, success); } } else { if (isLeader()) { zkUtils.createZNodeJsonReply(message, reply); } } } }catch (Exception e){ logger.error("Exception managing the ack of the group {} for the request {}: {}", groupId, message .getRequest_id(), e .getMessage()); } return ackReply; } }
ClusterSyncManager { public ActionCallbackDto manageAckStreamingOperation(StratioStreamingMessage message, ActionCallbackDto reply) { ActionCallbackDto ackReply = reply; try { if (!clusteringEnabled){ zkUtils.createZNodeJsonReply(message, reply); } else { if (allAckEnabled) { zkUtils.createTempZNodeJsonReply(message, reply, groupId); String path = zkUtils.getTempZNodeJsonReplyPath(message); String barrierPath = path.concat("/").concat(BARRIER_RELATIVE_PATH); if (this.clusterBarrierManager==null){ clusterBarrierManager = new ClusterBarrierManager(this,ackTimeout); } boolean success = clusterBarrierManager.manageAckBarrier(path, clusterNodes.size()); if (isLeader()) { ackReply = manageBarrierResults(message, reply, path, success); } } else { if (isLeader()) { zkUtils.createZNodeJsonReply(message, reply); } } } }catch (Exception e){ logger.error("Exception managing the ack of the group {} for the request {}: {}", groupId, message .getRequest_id(), e .getMessage()); } return ackReply; } 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 ActionCallbackDto manageAckStreamingOperation(StratioStreamingMessage message, ActionCallbackDto reply) { ActionCallbackDto ackReply = reply; try { if (!clusteringEnabled){ zkUtils.createZNodeJsonReply(message, reply); } else { if (allAckEnabled) { zkUtils.createTempZNodeJsonReply(message, reply, groupId); String path = zkUtils.getTempZNodeJsonReplyPath(message); String barrierPath = path.concat("/").concat(BARRIER_RELATIVE_PATH); if (this.clusterBarrierManager==null){ clusterBarrierManager = new ClusterBarrierManager(this,ackTimeout); } boolean success = clusterBarrierManager.manageAckBarrier(path, clusterNodes.size()); if (isLeader()) { ackReply = manageBarrierResults(message, reply, path, success); } } else { if (isLeader()) { zkUtils.createZNodeJsonReply(message, reply); } } } }catch (Exception e){ logger.error("Exception managing the ack of the group {} for the request {}: {}", groupId, message .getRequest_id(), e .getMessage()); } return ackReply; } 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 ActionCallbackDto manageAckStreamingOperation(StratioStreamingMessage message, ActionCallbackDto reply) { ActionCallbackDto ackReply = reply; try { if (!clusteringEnabled){ zkUtils.createZNodeJsonReply(message, reply); } else { if (allAckEnabled) { zkUtils.createTempZNodeJsonReply(message, reply, groupId); String path = zkUtils.getTempZNodeJsonReplyPath(message); String barrierPath = path.concat("/").concat(BARRIER_RELATIVE_PATH); if (this.clusterBarrierManager==null){ clusterBarrierManager = new ClusterBarrierManager(this,ackTimeout); } boolean success = clusterBarrierManager.manageAckBarrier(path, clusterNodes.size()); if (isLeader()) { ackReply = manageBarrierResults(message, reply, path, success); } } else { if (isLeader()) { zkUtils.createZNodeJsonReply(message, reply); } } } }catch (Exception e){ logger.error("Exception managing the ack of the group {} for the request {}: {}", groupId, message .getRequest_id(), e .getMessage()); } return ackReply; } 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();  }