<commit_before>/*
 * Copyright (c) 2017, WSO2 Inc. (http://www.wso2.org) All Rights Reserved.
 *
 * WSO2 Inc. 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.wso2.carbon.stream.processor.core.ha;

import com.google.gson.Gson;
import org.apache.log4j.Logger;
import org.wso2.carbon.cluster.coordinator.service.ClusterCoordinator;
import org.wso2.carbon.stream.processor.core.internal.beans.DeploymentConfig;
import org.wso2.carbon.stream.processor.core.ha.util.CompressionUtil;
import org.wso2.carbon.stream.processor.core.ha.util.RequestUtil;
import org.wso2.carbon.stream.processor.core.internal.StreamProcessorDataHolder;
import org.wso2.carbon.stream.processor.core.model.HAStateSyncObject;
import org.wso2.siddhi.core.SiddhiAppRuntime;
import org.wso2.siddhi.core.SiddhiManager;
import org.wso2.siddhi.core.stream.input.source.SourceHandler;

import java.io.IOException;
import java.net.URI;
import java.util.HashMap;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Timer;
import java.util.TimerTask;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;

/**
 * Class that manages Active and Passive nodes in a 2 node minimum HA configuration
 */
public class HAManager {

    private ClusterCoordinator clusterCoordinator;
    private ScheduledExecutorService passiveNodeOutputSchedulerService;
    private ScheduledFuture passiveNodeOutputScheduledFuture;
    private boolean liveSyncEnabled;
    private int outputSyncInterval;
    private String localHost;
    private String localPort;
    private Timer syncAfterGracePeriodTimer;
    private int stateSyncGracePeriod;
    private boolean isActiveNode;
    private String nodeId;
    private String clusterId;
    private int sinkQueueCapacity;
    private int sourceQueueCapacity;
    private String activeNodeHost;
    private String activeNodePort;
    private HACoordinationSourceHandlerManager sourceHandlerManager;
    private List<Timer> retrySiddhiAppSyncTimerList;

    public static Map<String, Object> activeNodePropertiesMap = new HashMap<>();
    private static final Logger log = Logger.getLogger(HAManager.class);

    public HAManager(ClusterCoordinator clusterCoordinator, String nodeId, String clusterId,
                     DeploymentConfig deploymentConfig) {
        this.clusterCoordinator = clusterCoordinator;
        this.nodeId = nodeId;
        this.clusterId = clusterId;
        this.localHost = deploymentConfig.getLiveSync().getAdvertisedHost();
        this.localPort = String.valueOf(deploymentConfig.getLiveSync().getAdvertisedPort());
        this.liveSyncEnabled = deploymentConfig.getLiveSync().isEnabled();
        this.outputSyncInterval = deploymentConfig.getOutputSyncInterval();
        this.stateSyncGracePeriod = deploymentConfig.getStateSyncGracePeriod();
        this.sinkQueueCapacity = deploymentConfig.getSinkQueueCapacity();
        this.sourceQueueCapacity = deploymentConfig.getSourceQueueCapacity();
        this.retrySiddhiAppSyncTimerList = new LinkedList<>();
    }

    public void start() {
        sourceHandlerManager = new HACoordinationSourceHandlerManager(sourceQueueCapacity);
        HACoordinationSinkHandlerManager sinkHandlerManager = new HACoordinationSinkHandlerManager(sinkQueueCapacity);
        HACoordinationRecordTableHandlerManager recordTableHandlerManager =
                new HACoordinationRecordTableHandlerManager(sinkQueueCapacity);

        StreamProcessorDataHolder.setSinkHandlerManager(sinkHandlerManager);
        StreamProcessorDataHolder.setSourceHandlerManager(sourceHandlerManager);
        StreamProcessorDataHolder.setRecordTableHandlerManager(recordTableHandlerManager);
        SiddhiManager siddhiManager = StreamProcessorDataHolder.getSiddhiManager();

        siddhiManager.setSourceHandlerManager(StreamProcessorDataHolder.getSourceHandlerManager());
        siddhiManager.setSinkHandlerManager(StreamProcessorDataHolder.getSinkHandlerManager());
        siddhiManager.setRecordTableHandlerManager(StreamProcessorDataHolder.getRecordTableHandlerManager());

        clusterCoordinator.registerEventListener(new HAEventListener());

        //Give time for the cluster to normalize
        while (clusterCoordinator.getLeaderNode() == null) {
            try {
                Thread.sleep(100);
            } catch (InterruptedException e) {
                log.warn("Error in waiting for leader node");
            }
        }

        isActiveNode = clusterCoordinator.isLeaderNode();

        if (isActiveNode) {
            log.info("HA Deployment: Starting up as Active Node");
            clusterCoordinator.setPropertiesMap(activeNodePropertiesMap);
            isActiveNode = true;
            if (!liveSyncEnabled) {
                ScheduledExecutorService scheduledExecutorService = Executors.newSingleThreadScheduledExecutor();
                scheduledExecutorService.scheduleAtFixedRate(new ActiveNodeOutputSyncManager(
                                sinkHandlerManager, recordTableHandlerManager, clusterCoordinator), 0, outputSyncInterval,
                        TimeUnit.MILLISECONDS);
            }
        } else {
            log.info("HA Deployment: Starting up as Passive Node");
            Map<String, Object> activeNodeHostAndPortMap = clusterCoordinator.getLeaderNode().getPropertiesMap();

            //Not checking for null of Map since LeaderNode check is done. Leader node will have properties
            activeNodeHost = (String) activeNodeHostAndPortMap.get("host");
            activeNodePort = (String) activeNodeHostAndPortMap.get("port");

            if (liveSyncEnabled) {
                log.info("Passive Node: Live Sync enabled. State sync from Active node scheduled after "
                        + stateSyncGracePeriod / 1000 + " seconds");
                syncAfterGracePeriodTimer = liveSyncAfterGracePeriod(stateSyncGracePeriod);
            } else {
                log.info("Passive Node: Live Sync disabled. State sync from Active node scheduled after "
                        + stateSyncGracePeriod / 1000 + " seconds");
                syncAfterGracePeriodTimer = persistenceStoreSyncAfterGracePeriod(stateSyncGracePeriod);
            }

            passiveNodeOutputSchedulerService = Executors.newSingleThreadScheduledExecutor();
            passiveNodeOutputScheduledFuture = passiveNodeOutputSchedulerService.scheduleAtFixedRate(
                    new PassiveNodeOutputSyncManager(clusterCoordinator, sinkHandlerManager, recordTableHandlerManager,
                            activeNodeHost, activeNodePort, liveSyncEnabled), outputSyncInterval, outputSyncInterval,
                    TimeUnit.MILLISECONDS);
        }
        HAInfo haInfo = new HAInfo(nodeId, clusterId, isActiveNode);
        StreamProcessorDataHolder.setHaInfo(haInfo);
        StreamProcessorDataHolder.getInstance().getBundleContext().registerService(HAInfo.class.getName(), haInfo,
                null);
    }

    /**
     * Stops Publisher Syncing of Passive Node
     * Cancels scheduled state sync of Passive Node from Active Node
     * Updates Coordination Properties with Advertised Host and Port
     */
    void changeToActive() {
        activeNodePropertiesMap.put("host", localHost);
        activeNodePropertiesMap.put("port", localPort);
        clusterCoordinator.setPropertiesMap(activeNodePropertiesMap);

        if (passiveNodeOutputScheduledFuture != null) {
            passiveNodeOutputScheduledFuture.cancel(false);
        }
        if (passiveNodeOutputSchedulerService != null) {
            passiveNodeOutputSchedulerService.shutdown();
        }
        if (syncAfterGracePeriodTimer != null) {
            syncAfterGracePeriodTimer.cancel();
            syncAfterGracePeriodTimer.purge();
        }
        if (retrySiddhiAppSyncTimerList.size() > 0) {
            for (Timer timer : retrySiddhiAppSyncTimerList) {<commit_msg>Bug fix - Set active node flag true<commit_after>/*
 * Copyright (c) 2017, WSO2 Inc. (http://www.wso2.org) All Rights Reserved.
 *
 * WSO2 Inc. 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.wso2.carbon.stream.processor.core.ha;

import com.google.gson.Gson;
import org.apache.log4j.Logger;
import org.wso2.carbon.cluster.coordinator.service.ClusterCoordinator;
import org.wso2.carbon.stream.processor.core.internal.beans.DeploymentConfig;
import org.wso2.carbon.stream.processor.core.ha.util.CompressionUtil;
import org.wso2.carbon.stream.processor.core.ha.util.RequestUtil;
import org.wso2.carbon.stream.processor.core.internal.StreamProcessorDataHolder;
import org.wso2.carbon.stream.processor.core.model.HAStateSyncObject;
import org.wso2.siddhi.core.SiddhiAppRuntime;
import org.wso2.siddhi.core.SiddhiManager;
import org.wso2.siddhi.core.stream.input.source.SourceHandler;

import java.io.IOException;
import java.net.URI;
import java.util.HashMap;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Timer;
import java.util.TimerTask;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;

/**
 * Class that manages Active and Passive nodes in a 2 node minimum HA configuration
 */
public class HAManager {

    private ClusterCoordinator clusterCoordinator;
    private ScheduledExecutorService passiveNodeOutputSchedulerService;
    private ScheduledFuture passiveNodeOutputScheduledFuture;
    private boolean liveSyncEnabled;
    private int outputSyncInterval;
    private String localHost;
    private String localPort;
    private Timer syncAfterGracePeriodTimer;
    private int stateSyncGracePeriod;
    private boolean isActiveNode;
    private String nodeId;
    private String clusterId;
    private int sinkQueueCapacity;
    private int sourceQueueCapacity;
    private String activeNodeHost;
    private String activeNodePort;
    private HACoordinationSourceHandlerManager sourceHandlerManager;
    private List<Timer> retrySiddhiAppSyncTimerList;

    public static Map<String, Object> activeNodePropertiesMap = new HashMap<>();
    private static final Logger log = Logger.getLogger(HAManager.class);

    public HAManager(ClusterCoordinator clusterCoordinator, String nodeId, String clusterId,
                     DeploymentConfig deploymentConfig) {
        this.clusterCoordinator = clusterCoordinator;
        this.nodeId = nodeId;
        this.clusterId = clusterId;
        this.localHost = deploymentConfig.getLiveSync().getAdvertisedHost();
        this.localPort = String.valueOf(deploymentConfig.getLiveSync().getAdvertisedPort());
        this.liveSyncEnabled = deploymentConfig.getLiveSync().isEnabled();
        this.outputSyncInterval = deploymentConfig.getOutputSyncInterval();
        this.stateSyncGracePeriod = deploymentConfig.getStateSyncGracePeriod();
        this.sinkQueueCapacity = deploymentConfig.getSinkQueueCapacity();
        this.sourceQueueCapacity = deploymentConfig.getSourceQueueCapacity();
        this.retrySiddhiAppSyncTimerList = new LinkedList<>();
    }

    public void start() {
        sourceHandlerManager = new HACoordinationSourceHandlerManager(sourceQueueCapacity);
        HACoordinationSinkHandlerManager sinkHandlerManager = new HACoordinationSinkHandlerManager(sinkQueueCapacity);
        HACoordinationRecordTableHandlerManager recordTableHandlerManager =
                new HACoordinationRecordTableHandlerManager(sinkQueueCapacity);

        StreamProcessorDataHolder.setSinkHandlerManager(sinkHandlerManager);
        StreamProcessorDataHolder.setSourceHandlerManager(sourceHandlerManager);
        StreamProcessorDataHolder.setRecordTableHandlerManager(recordTableHandlerManager);
        SiddhiManager siddhiManager = StreamProcessorDataHolder.getSiddhiManager();

        siddhiManager.setSourceHandlerManager(StreamProcessorDataHolder.getSourceHandlerManager());
        siddhiManager.setSinkHandlerManager(StreamProcessorDataHolder.getSinkHandlerManager());
        siddhiManager.setRecordTableHandlerManager(StreamProcessorDataHolder.getRecordTableHandlerManager());

        clusterCoordinator.registerEventListener(new HAEventListener());

        //Give time for the cluster to normalize
        while (clusterCoordinator.getLeaderNode() == null) {
            try {
                Thread.sleep(100);
            } catch (InterruptedException e) {
                log.warn("Error in waiting for leader node");
            }
        }

        isActiveNode = clusterCoordinator.isLeaderNode();

        if (isActiveNode) {
            log.info("HA Deployment: Starting up as Active Node");
            isActiveNode = true;
            if (!liveSyncEnabled) {
                ScheduledExecutorService scheduledExecutorService = Executors.newSingleThreadScheduledExecutor();
                scheduledExecutorService.scheduleAtFixedRate(new ActiveNodeOutputSyncManager(
                                sinkHandlerManager, recordTableHandlerManager, clusterCoordinator), 0, outputSyncInterval,
                        TimeUnit.MILLISECONDS);
            }
        } else {
            log.info("HA Deployment: Starting up as Passive Node");
            Map<String, Object> activeNodeHostAndPortMap = clusterCoordinator.getLeaderNode().getPropertiesMap();

            //Not checking for null of Map since LeaderNode check is done. Leader node will have properties
            activeNodeHost = (String) activeNodeHostAndPortMap.get("host");
            activeNodePort = (String) activeNodeHostAndPortMap.get("port");

            if (liveSyncEnabled) {
                log.info("Passive Node: Live Sync enabled. State sync from Active node scheduled after "
                        + stateSyncGracePeriod / 1000 + " seconds");
                syncAfterGracePeriodTimer = liveSyncAfterGracePeriod(stateSyncGracePeriod);
            } else {
                log.info("Passive Node: Live Sync disabled. State sync from Active node scheduled after "
                        + stateSyncGracePeriod / 1000 + " seconds");
                syncAfterGracePeriodTimer = persistenceStoreSyncAfterGracePeriod(stateSyncGracePeriod);
            }

            passiveNodeOutputSchedulerService = Executors.newSingleThreadScheduledExecutor();
            passiveNodeOutputScheduledFuture = passiveNodeOutputSchedulerService.scheduleAtFixedRate(
                    new PassiveNodeOutputSyncManager(clusterCoordinator, sinkHandlerManager, recordTableHandlerManager,
                            activeNodeHost, activeNodePort, liveSyncEnabled), outputSyncInterval, outputSyncInterval,
                    TimeUnit.MILLISECONDS);
        }
        HAInfo haInfo = new HAInfo(nodeId, clusterId, isActiveNode);
        StreamProcessorDataHolder.setHaInfo(haInfo);
        StreamProcessorDataHolder.getInstance().getBundleContext().registerService(HAInfo.class.getName(), haInfo,
                null);
    }

    /**
     * Stops Publisher Syncing of Passive Node
     * Cancels scheduled state sync of Passive Node from Active Node
     * Updates Coordination Properties with Advertised Host and Port
     */
    void changeToActive() {
        isActiveNode = true;
        activeNodePropertiesMap.put("host", localHost);
        activeNodePropertiesMap.put("port", localPort);
        clusterCoordinator.setPropertiesMap(activeNodePropertiesMap);

        if (passiveNodeOutputScheduledFuture != null) {
            passiveNodeOutputScheduledFuture.cancel(false);
        }
        if (passiveNodeOutputSchedulerService != null) {
            passiveNodeOutputSchedulerService.shutdown();
        }
        if (syncAfterGracePeriodTimer != null) {
            syncAfterGracePeriodTimer.cancel();
            syncAfterGracePeriodTimer.purge();
        }
        if (retrySiddhiAppSyncTimerList.size() > 0) {
            for (Timer timer : retrySiddhiAppSyncTimerList) {
{'fraction_non_alphanumeric': 0.07415808759378817, 'fraction_numerical': 0.004187750828825685, 'mean_word_length': 3.4939884997386303, 'pattern_counts': {'":': 0, '<': 13, '<?xml version=': 0, '>': 15, 'https://': 0, 'lorem ipsum': 0, 'www.': 4, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3322623', 'n_tokens_mistral': 4366, 'n_tokens_neox': 4280, 'n_words': 965}