<commit_before>package net.onrc.onos.core.datastore.hazelcast;

import java.io.FileNotFoundException;
import java.util.Collection;
import java.util.List;

import net.onrc.onos.core.datagrid.HazelcastDatagrid;
import net.onrc.onos.core.datastore.IKVClient;
import net.onrc.onos.core.datastore.IKVTable;
import net.onrc.onos.core.datastore.IKVTable.IKVEntry;
import net.onrc.onos.core.datastore.IKVTableID;
import net.onrc.onos.core.datastore.IMultiEntryOperation;
import net.onrc.onos.core.datastore.IMultiEntryOperation.OPERATION;
import net.onrc.onos.core.datastore.IMultiEntryOperation.STATUS;
import net.onrc.onos.core.datastore.ObjectDoesntExistException;
import net.onrc.onos.core.datastore.ObjectExistsException;
import net.onrc.onos.core.datastore.WrongVersionException;
import net.onrc.onos.core.datastore.hazelcast.HZTable.VersionedValue;
import net.onrc.onos.core.datastore.internal.IModifiableMultiEntryOperation;
import net.onrc.onos.core.datastore.utils.ByteArrayUtil;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.hazelcast.client.HazelcastClient;
import com.hazelcast.client.config.ClientConfig;
import com.hazelcast.config.Config;
import com.hazelcast.config.FileSystemXmlConfig;
import com.hazelcast.config.MapConfig;
import com.hazelcast.config.SerializationConfig;
import com.hazelcast.core.Hazelcast;
import com.hazelcast.core.HazelcastInstance;
import com.hazelcast.core.IAtomicLong;
import com.hazelcast.core.IMap;

/**
 * Hazelcast implementation of datastore IKVClient.
 */
public final class HZClient implements IKVClient {
    private static final Logger log = LoggerFactory.getLogger(HZClient.class);

    static final long VERSION_NONEXISTENT = 0L;

    private static final String MAP_PREFIX = "datastore://";

    private static final String BASE_CONFIG_FILENAME =
            System.getProperty("net.onrc.onos.core.datastore.hazelcast.baseConfig", "conf/hazelcast.xml");
    private static final String HAZELCAST_DEFAULT_XML = "conf/hazelcast.default.xml";

    // XXX Remove this mode at some point
    private static boolean useClientMode = Boolean.parseBoolean(
            System.getProperty("net.onrc.onos.core.datastore.hazelcast.clientMode", "true"));

    // Note: xml configuration will overwrite this value if present
    private static int backupCount = Integer.parseInt(
            System.getProperty("net.onrc.onos.core.datastore.hazelcast.backupCount", "3"));

    private final HazelcastInstance hazelcastInstance;

    private static final HZClient THE_INSTANCE = new HZClient();

    /**
     * Get DataStoreClient implemented on Hazelcast.
     *
     * @return HZClient
     */
    public static HZClient getClient() {
        return THE_INSTANCE;
    }

    /**
     * Default constructor.
     * <p/>
     * Get or create the Hazelcast Instance to use for datastore.
     */
    private HZClient() {
        // Try to get the existing HZ instance in JVM if possible.
        HazelcastInstance instance = Hazelcast.getHazelcastInstanceByName(HazelcastDatagrid.ONOS_HAZELCAST_INSTANCE);
        if (instance == null) {
            log.error("Failed to get the Hazelcast instance in JVM. "
                    + "Probably DataStoreClient was requested before "
                    + "IDatagridService was started "
                    + "or running as part of unit tests. "
                    + "Creating instance on it's own.");
            instance = getFallbackHZinstance(BASE_CONFIG_FILENAME);
        }
        hazelcastInstance = instance;
    }

    /**
     * Get or create the hazelcast instance to use for datastore, when existing
     * Hazelcast instance cannot be retrieved.
     * <p/>
     *
     * @param hazelcastConfigFileName Hazelcast configuration to use when creating a
     * @return HazelcastInstance to use for datastore
     */
    private static HazelcastInstance getFallbackHZinstance(final String hazelcastConfigFileName) {
        Config baseHzConfig = null;
        try {
            baseHzConfig = new FileSystemXmlConfig(hazelcastConfigFileName);
        } catch (FileNotFoundException e) {
            log.error("Error opening Hazelcast XML configuration. File not found: " + hazelcastConfigFileName, e);
            // Fallback mechanism to support running unit test without setup.
            log.error("Falling back to default Hazelcast XML {}", HAZELCAST_DEFAULT_XML);
            try {
                baseHzConfig = new FileSystemXmlConfig(HAZELCAST_DEFAULT_XML);
            } catch (FileNotFoundException e2) {
                log.error("Error opening fall back Hazelcast XML configuration. "
                        + "File not found: " + HAZELCAST_DEFAULT_XML, e2);

                // intentionally throwing Exception "e" thrown from non-fallback
                // Hazelcast configuration loading.
                throw new Error("Cannot find Hazelcast configuration: " + hazelcastConfigFileName, e);
            }
        }

        // use xml config if present, if not use System.property
        MapConfig mapConfig = baseHzConfig.getMapConfigs().get(MAP_PREFIX + "*");
        if (mapConfig != null) {
            backupCount = mapConfig.getBackupCount();
        }

        HazelcastInstance instance = null;
        // TODO Client mode should be removed at some point.
        // we can get HZ instance used by ONOS using getHazelcastInstanceByName
        if (useClientMode) {
            log.info("Configuring Hazelcast datastore as Client mode");
            ClientConfig clientConfig = new ClientConfig();
            final int port = baseHzConfig.getNetworkConfig().getPort();

            String server = System.getProperty("net.onrc.onos.core.datastore.hazelcast.client.server", "localhost");
            clientConfig.addAddress(server + ":" + port);

            // client mode connection limit.
            // set to 0 for fast fall back to Instance mode.
            String sAttempts = System.getProperty("net.onrc.onos.core.datastore.hazelcast.client.attemptLimit");
            if (sAttempts != null) {
                clientConfig.setConnectionAttemptLimit(Integer.parseInt(sAttempts));
            }

            // copy group config from base Hazelcast configuration
            clientConfig.getGroupConfig().setName(baseHzConfig.getGroupConfig().getName());
            clientConfig.getGroupConfig().setPassword(baseHzConfig.getGroupConfig().getPassword());

            // TODO We probably need to figure out what else need to be
            // derived from baseConfig

            registerSerializer(clientConfig.getSerializationConfig());

            log.info("Starting Hazelcast datastore client for [{}]", clientConfig.getAddressList());

            try {
                instance = HazelcastClient.newHazelcastClient(clientConfig);
                if (!instance.getCluster().getMembers().isEmpty()) {
                    log.debug("Members in cluster: " + instance.getCluster().getMembers());
                    return instance;
                }
                log.info("Failed to find cluster member, falling back to Instance mode");
            } catch (IllegalStateException e) {
                log.info("Failed to initialize HazelcastClient, falling back to Instance mode");
            }
            useClientMode = false;
            instance = null;
        }
        log.info("Configuring Hazelcast datastore as Instance mode");

        // To run 2 Hazelcast instance in 1 JVM,
        // we probably need to something like below
        //int port = hazelcastConfig.getNetworkConfig().getPort();
        //hazelcastConfig.getNetworkConfig().setPort(port+1);

        registerSerializer(baseHzConfig.getSerializationConfig());

        return Hazelcast.newHazelcastInstance(baseHzConfig);
    }

    /**
     * Register serializer for VersionedValue class used to imitate value version.
     *
     * @param config SerializationConfig to add VersionedValueSerializableFactory.
     */
    private static void registerSerializer(final SerializationConfig config) {
        // TODO remove this function at some point.
        // This method is no longer required, if equibalent to the following
        // is defined in hazelcast.xml
        config.addDataSerializableFactoryClass(
                VersionedValueSerializableFactory.FACTORY_ID,
                VersionedValueSerializableFactory.class);
    }

    @Override
    public IKVTable getTable(final String tableName) {
        IMap<byte[], VersionedValue> map = hazelcastInstance.getMap(MAP_PREFIX + tableName);

        if (!useClientMode) {
            // config only available in Instance Mode
            // Client Mode must rely on hazelcast.xml to be properly configured.
            MapConfig config = hazelcastInstance.getConfig().getMapConfig(MAP_PREFIX + tableName);<commit_msg>Give access to HazelcastInstance<commit_after>package net.onrc.onos.core.datastore.hazelcast;

import java.io.FileNotFoundException;
import java.util.Collection;
import java.util.List;

import net.onrc.onos.core.datagrid.HazelcastDatagrid;
import net.onrc.onos.core.datastore.IKVClient;
import net.onrc.onos.core.datastore.IKVTable;
import net.onrc.onos.core.datastore.IKVTable.IKVEntry;
import net.onrc.onos.core.datastore.IKVTableID;
import net.onrc.onos.core.datastore.IMultiEntryOperation;
import net.onrc.onos.core.datastore.IMultiEntryOperation.OPERATION;
import net.onrc.onos.core.datastore.IMultiEntryOperation.STATUS;
import net.onrc.onos.core.datastore.ObjectDoesntExistException;
import net.onrc.onos.core.datastore.ObjectExistsException;
import net.onrc.onos.core.datastore.WrongVersionException;
import net.onrc.onos.core.datastore.hazelcast.HZTable.VersionedValue;
import net.onrc.onos.core.datastore.internal.IModifiableMultiEntryOperation;
import net.onrc.onos.core.datastore.utils.ByteArrayUtil;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.hazelcast.client.HazelcastClient;
import com.hazelcast.client.config.ClientConfig;
import com.hazelcast.config.Config;
import com.hazelcast.config.FileSystemXmlConfig;
import com.hazelcast.config.MapConfig;
import com.hazelcast.config.SerializationConfig;
import com.hazelcast.core.Hazelcast;
import com.hazelcast.core.HazelcastInstance;
import com.hazelcast.core.IAtomicLong;
import com.hazelcast.core.IMap;

/**
 * Hazelcast implementation of datastore IKVClient.
 */
public final class HZClient implements IKVClient {
    private static final Logger log = LoggerFactory.getLogger(HZClient.class);

    static final long VERSION_NONEXISTENT = 0L;

    private static final String MAP_PREFIX = "datastore://";

    private static final String BASE_CONFIG_FILENAME =
            System.getProperty("net.onrc.onos.core.datastore.hazelcast.baseConfig", "conf/hazelcast.xml");
    private static final String HAZELCAST_DEFAULT_XML = "conf/hazelcast.default.xml";

    // XXX Remove this mode at some point
    private static boolean useClientMode = Boolean.parseBoolean(
            System.getProperty("net.onrc.onos.core.datastore.hazelcast.clientMode", "true"));

    // Note: xml configuration will overwrite this value if present
    private static int backupCount = Integer.parseInt(
            System.getProperty("net.onrc.onos.core.datastore.hazelcast.backupCount", "3"));

    private final HazelcastInstance hazelcastInstance;

    private static final HZClient THE_INSTANCE = new HZClient();

    /**
     * Get DataStoreClient implemented on Hazelcast.
     *
     * @return HZClient
     */
    public static HZClient getClient() {
        return THE_INSTANCE;
    }

    /**
     * Default constructor.
     * <p/>
     * Get or create the Hazelcast Instance to use for datastore.
     */
    private HZClient() {
        // Try to get the existing HZ instance in JVM if possible.
        HazelcastInstance instance = Hazelcast.getHazelcastInstanceByName(HazelcastDatagrid.ONOS_HAZELCAST_INSTANCE);
        if (instance == null) {
            log.error("Failed to get the Hazelcast instance in JVM. "
                    + "Probably DataStoreClient was requested before "
                    + "IDatagridService was started "
                    + "or running as part of unit tests. "
                    + "Creating instance on it's own.");
            instance = getFallbackHZinstance(BASE_CONFIG_FILENAME);
        }
        hazelcastInstance = instance;
    }

    /**
     * Get or create the hazelcast instance to use for datastore, when existing
     * Hazelcast instance cannot be retrieved.
     * <p/>
     *
     * @param hazelcastConfigFileName Hazelcast configuration to use when creating a
     * @return HazelcastInstance to use for datastore
     */
    private static HazelcastInstance getFallbackHZinstance(final String hazelcastConfigFileName) {
        Config baseHzConfig = null;
        try {
            baseHzConfig = new FileSystemXmlConfig(hazelcastConfigFileName);
        } catch (FileNotFoundException e) {
            log.error("Error opening Hazelcast XML configuration. File not found: " + hazelcastConfigFileName, e);
            // Fallback mechanism to support running unit test without setup.
            log.error("Falling back to default Hazelcast XML {}", HAZELCAST_DEFAULT_XML);
            try {
                baseHzConfig = new FileSystemXmlConfig(HAZELCAST_DEFAULT_XML);
            } catch (FileNotFoundException e2) {
                log.error("Error opening fall back Hazelcast XML configuration. "
                        + "File not found: " + HAZELCAST_DEFAULT_XML, e2);

                // intentionally throwing Exception "e" thrown from non-fallback
                // Hazelcast configuration loading.
                throw new Error("Cannot find Hazelcast configuration: " + hazelcastConfigFileName, e);
            }
        }

        // use xml config if present, if not use System.property
        MapConfig mapConfig = baseHzConfig.getMapConfigs().get(MAP_PREFIX + "*");
        if (mapConfig != null) {
            backupCount = mapConfig.getBackupCount();
        }

        HazelcastInstance instance = null;
        // TODO Client mode should be removed at some point.
        // we can get HZ instance used by ONOS using getHazelcastInstanceByName
        if (useClientMode) {
            log.info("Configuring Hazelcast datastore as Client mode");
            ClientConfig clientConfig = new ClientConfig();
            final int port = baseHzConfig.getNetworkConfig().getPort();

            String server = System.getProperty("net.onrc.onos.core.datastore.hazelcast.client.server", "localhost");
            clientConfig.addAddress(server + ":" + port);

            // client mode connection limit.
            // set to 0 for fast fall back to Instance mode.
            String sAttempts = System.getProperty("net.onrc.onos.core.datastore.hazelcast.client.attemptLimit");
            if (sAttempts != null) {
                clientConfig.setConnectionAttemptLimit(Integer.parseInt(sAttempts));
            }

            // copy group config from base Hazelcast configuration
            clientConfig.getGroupConfig().setName(baseHzConfig.getGroupConfig().getName());
            clientConfig.getGroupConfig().setPassword(baseHzConfig.getGroupConfig().getPassword());

            // TODO We probably need to figure out what else need to be
            // derived from baseConfig

            registerSerializer(clientConfig.getSerializationConfig());

            log.info("Starting Hazelcast datastore client for [{}]", clientConfig.getAddressList());

            try {
                instance = HazelcastClient.newHazelcastClient(clientConfig);
                if (!instance.getCluster().getMembers().isEmpty()) {
                    log.debug("Members in cluster: " + instance.getCluster().getMembers());
                    return instance;
                }
                log.info("Failed to find cluster member, falling back to Instance mode");
            } catch (IllegalStateException e) {
                log.info("Failed to initialize HazelcastClient, falling back to Instance mode");
            }
            useClientMode = false;
            instance = null;
        }
        log.info("Configuring Hazelcast datastore as Instance mode");

        // To run 2 Hazelcast instance in 1 JVM,
        // we probably need to something like below
        //int port = hazelcastConfig.getNetworkConfig().getPort();
        //hazelcastConfig.getNetworkConfig().setPort(port+1);

        registerSerializer(baseHzConfig.getSerializationConfig());

        return Hazelcast.newHazelcastInstance(baseHzConfig);
    }

    /**
     * Register serializer for VersionedValue class used to imitate value version.
     *
     * @param config SerializationConfig to add VersionedValueSerializableFactory.
     */
    private static void registerSerializer(final SerializationConfig config) {
        // TODO remove this function at some point.
        // This method is no longer required, if equibalent to the following
        // is defined in hazelcast.xml
        config.addDataSerializableFactoryClass(
                VersionedValueSerializableFactory.FACTORY_ID,
                VersionedValueSerializableFactory.class);
    }


    /**
     * Gets the HazelcastInstance object.
     *
     * @return HazelcastInstance
     */
    HazelcastInstance getHZInstance() {
        return hazelcastInstance;
    }

    @Override
    public IKVTable getTable(final String tableName) {
        IMap<byte[], VersionedValue> map = hazelcastInstance.getMap(MAP_PREFIX + tableName);

        if (!useClientMode) {
            // config only available in Instance Mode
            // Client Mode must rely on hazelcast.xml to be properly configured.
            MapConfig config = hazelcastInstance.getConfig().getMapConfig(MAP_PREFIX + tableName);
{'fraction_non_alphanumeric': 0.08536790705505977, 'fraction_numerical': 0.0011225234326766571, 'mean_word_length': 2.7894512973202894, 'pattern_counts': {'":': 2, '<': 9, '<?xml version=': 0, '>': 9, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 12}, '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': '5712036', 'n_tokens_mistral': 4960, 'n_tokens_neox': 4871, 'n_words': 1192}