class ClusterSyncManager {

    public ActionCallbackDto manageAckStreamingOperation(StratioStreamingMessage message, ActionCallbackDto reply) {


        ActionCallbackDto ackReply = reply;

        try {

            // Single instance mode
            if (!clusteringEnabled){
                zkUtils.createZNodeJsonReply(message, reply);
            } else {
                // Sharding mode
                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;

    }

    public  ClusterSyncManager(String latchpath, String id, ConfigurationContext
            configurationContext, FailOverTask failOverTask, CuratorFramework curatorClient, ZKUtils zkUtils,
            ClusterBarrierManager clusterBarrierManager);
    public  ClusterSyncManager(String latchpath, String id, ConfigurationContext
            configurationContext, FailOverTask failOverTask);

}