<commit_before>package com.thinkaurelius.titan.diskstorage.cassandra.astyanax;

import com.google.common.base.Preconditions;
import com.google.common.collect.ImmutableMap;
import com.netflix.astyanax.*;
import com.netflix.astyanax.connectionpool.NodeDiscoveryType;
import com.netflix.astyanax.connectionpool.exceptions.ConnectionException;
import com.netflix.astyanax.connectionpool.impl.*;
import com.netflix.astyanax.ddl.ColumnFamilyDefinition;
import com.netflix.astyanax.ddl.KeyspaceDefinition;
import com.netflix.astyanax.impl.AstyanaxConfigurationImpl;
import com.netflix.astyanax.model.ColumnFamily;
import com.netflix.astyanax.retry.RetryPolicy;
import com.netflix.astyanax.serializers.ByteBufferSerializer;
import com.netflix.astyanax.serializers.StringSerializer;
import com.netflix.astyanax.thrift.ThriftFamilyFactory;
import com.thinkaurelius.titan.diskstorage.PermanentStorageException;
import com.thinkaurelius.titan.diskstorage.StaticBuffer;
import com.thinkaurelius.titan.diskstorage.StorageException;
import com.thinkaurelius.titan.diskstorage.TemporaryStorageException;
import com.thinkaurelius.titan.diskstorage.cassandra.AbstractCassandraStoreManager;
import com.thinkaurelius.titan.diskstorage.cassandra.astyanax.locking.AstyanaxRecipeLocker;
import com.thinkaurelius.titan.diskstorage.keycolumnvalue.Entry;
import com.thinkaurelius.titan.diskstorage.keycolumnvalue.KCVMutation;
import com.thinkaurelius.titan.diskstorage.keycolumnvalue.StoreTransaction;

import org.apache.cassandra.dht.IPartitioner;
import org.apache.cassandra.dht.Token;
import org.apache.cassandra.exceptions.ConfigurationException;
import org.apache.cassandra.utils.FBUtilities;
import org.apache.commons.configuration.Configuration;
import org.apache.commons.lang.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.lang.reflect.Constructor;
import java.nio.ByteBuffer;
import java.util.HashMap;
import java.util.Map;

import static com.thinkaurelius.titan.diskstorage.cassandra.CassandraTransaction.getTx;

public class AstyanaxStoreManager extends AbstractCassandraStoreManager {

    private static final Logger log = LoggerFactory.getLogger(AstyanaxStoreManager.class);

    //################### ASTYANAX SPECIFIC CONFIGURATION OPTIONS ######################

    /**
     * Default name for the Cassandra cluster
     * <p/>
     * Value = {@value}
     */
    public static final String CLUSTER_DEFAULT = "Titan Cluster";
    public static final String CLUSTER_KEY = "cluster-name";

    /**
     * Maximum pooled connections per host.
     * <p/>
     * Value = {@value}
     */
    public static final int MAX_CONNECTIONS_PER_HOST_DEFAULT = 32;
    public static final String MAX_CONNECTIONS_PER_HOST_KEY = "max-connections-per-host";

    /**
     * Maximum open connections allowed in the pool (counting all hosts).
     * <p/>
     * Value = {@value}
     */
    public static final int MAX_CONNECTIONS_DEFAULT = -1;
    public static final String MAX_CONNECTIONS_KEY = "max-connections";

    /**
     * Maximum number of operations allowed per connection before the connection is closed.
     * <p/>
     * Value = {@value}
     */
    public static final int MAX_OPERATIONS_PER_CONNECTION_DEFAULT = 100 * 1000;
    public static final String MAX_OPERATIONS_PER_CONNECTION_KEY = "max-operations-per-connection";

    /**
     * Maximum pooled "cluster" connections per host.
     * <p/>
     * These connections are mostly idle and only used for DDL operations
     * (like creating keyspaces).  Titan doesn't need many of these connections
     * in ordinary operation.
     */
    public static final int MAX_CLUSTER_CONNECTIONS_PER_HOST_DEFAULT = 3;
    public static final String MAX_CLUSTER_CONNECTIONS_PER_HOST_KEY = "max-cluster-connections-per-host";

    /**
     * How Astyanax discovers Cassandra cluster nodes. This must be one of the
     * values of the Astyanax NodeDiscoveryType enum.
     * <p/>
     * Value = {@value}
     */
    public static final String NODE_DISCOVERY_TYPE_DEFAULT = "RING_DESCRIBE";
    public static final String NODE_DISCOVERY_TYPE_KEY = "node-discovery-type";

    /**
     * Astyanax's connection pooler implementation. This must be one of the
     * values of the Astyanax ConnectionPoolType enum.
     * <p/>
     * Value = {@value}
     */
    public static final String CONNECTION_POOL_TYPE_DEFAULT = "TOKEN_AWARE";
    public static final String CONNECTION_POOL_TYPE_KEY = "connection-pool-type";

    /**
     * This must be the fully-qualified classname (i.e. the complete package
     * name, a dot, and then the class name) of an implementation of Astyanax's
     * RetryPolicy interface. This string may be followed by a sequence of
     * integers, separated from the full classname and from each other by
     * commas; in this case, the integers are cast to native Java ints and
     * passed to the class constructor as arguments.
     * <p/>
     * Value = {@value}
     */
    public static final String RETRY_POLICY_DEFAULT = "com.netflix.astyanax.retry.BoundedExponentialBackoff,100,25000,8";
    public static final String RETRY_POLICY_KEY = "retry-policy";

    private final String clusterName;

    private final AstyanaxContext<Keyspace> keyspaceContext;
    private final AstyanaxContext<Cluster> clusterContext;

    private final RetryPolicy retryPolicy;

    private final Map<String, AstyanaxOrderedKeyColumnValueStore> openStores;

    public AstyanaxStoreManager(Configuration config) throws StorageException {
        super(config);

        // Check if we have non-default thrift frame size or max message size set and warn users
        // because there is nothing we can do in Astyanax to apply those, warning is good enough here
        // otherwise it would make bad user experience if we don't warn at all or crash on this.
        if (config.containsKey(THRIFT_FRAME_SIZE_MB))
            log.warn("Couldn't set custom Thrift Frame Size property, use 'cassandrathrift' instead.");

        this.clusterName = config.getString(CLUSTER_KEY, CLUSTER_DEFAULT);

        this.retryPolicy = getRetryPolicy(config.getString(RETRY_POLICY_KEY, RETRY_POLICY_DEFAULT));

        final int maxConnsPerHost =
                config.getInt(
                        MAX_CONNECTIONS_PER_HOST_KEY,
                        MAX_CONNECTIONS_PER_HOST_DEFAULT);

        final int maxClusterConnsPerHost =
                config.getInt(
                        MAX_CLUSTER_CONNECTIONS_PER_HOST_KEY,
                        MAX_CLUSTER_CONNECTIONS_PER_HOST_DEFAULT);

        this.clusterContext = createCluster(getContextBuilder(config, maxClusterConnsPerHost, "Cluster"));

        ensureKeyspaceExists(clusterContext.getClient());

        this.keyspaceContext = getContextBuilder(config, maxConnsPerHost, "Keyspace").buildKeyspace(ThriftFamilyFactory.getInstance());
        this.keyspaceContext.start();

        openStores = new HashMap<String, AstyanaxOrderedKeyColumnValueStore>(8);
    }

    @Override
    public Deployment getDeployment() {
        //return Deployment.REMOTE;
        //return LOCAL if over localhost else REMOTE
    }
<commit_msg>Make Astyanax always assume Deployment.REMOTE<commit_after>package com.thinkaurelius.titan.diskstorage.cassandra.astyanax;

import com.google.common.base.Preconditions;
import com.google.common.collect.ImmutableMap;
import com.netflix.astyanax.*;
import com.netflix.astyanax.connectionpool.NodeDiscoveryType;
import com.netflix.astyanax.connectionpool.exceptions.ConnectionException;
import com.netflix.astyanax.connectionpool.impl.*;
import com.netflix.astyanax.ddl.ColumnFamilyDefinition;
import com.netflix.astyanax.ddl.KeyspaceDefinition;
import com.netflix.astyanax.impl.AstyanaxConfigurationImpl;
import com.netflix.astyanax.model.ColumnFamily;
import com.netflix.astyanax.retry.RetryPolicy;
import com.netflix.astyanax.serializers.ByteBufferSerializer;
import com.netflix.astyanax.serializers.StringSerializer;
import com.netflix.astyanax.thrift.ThriftFamilyFactory;
import com.thinkaurelius.titan.diskstorage.PermanentStorageException;
import com.thinkaurelius.titan.diskstorage.StaticBuffer;
import com.thinkaurelius.titan.diskstorage.StorageException;
import com.thinkaurelius.titan.diskstorage.TemporaryStorageException;
import com.thinkaurelius.titan.diskstorage.cassandra.AbstractCassandraStoreManager;
import com.thinkaurelius.titan.diskstorage.cassandra.astyanax.locking.AstyanaxRecipeLocker;
import com.thinkaurelius.titan.diskstorage.keycolumnvalue.Entry;
import com.thinkaurelius.titan.diskstorage.keycolumnvalue.KCVMutation;
import com.thinkaurelius.titan.diskstorage.keycolumnvalue.StoreTransaction;

import org.apache.cassandra.dht.IPartitioner;
import org.apache.cassandra.dht.Token;
import org.apache.cassandra.exceptions.ConfigurationException;
import org.apache.cassandra.utils.FBUtilities;
import org.apache.commons.configuration.Configuration;
import org.apache.commons.lang.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.lang.reflect.Constructor;
import java.nio.ByteBuffer;
import java.util.HashMap;
import java.util.Map;

import static com.thinkaurelius.titan.diskstorage.cassandra.CassandraTransaction.getTx;

public class AstyanaxStoreManager extends AbstractCassandraStoreManager {

    private static final Logger log = LoggerFactory.getLogger(AstyanaxStoreManager.class);

    //################### ASTYANAX SPECIFIC CONFIGURATION OPTIONS ######################

    /**
     * Default name for the Cassandra cluster
     * <p/>
     * Value = {@value}
     */
    public static final String CLUSTER_DEFAULT = "Titan Cluster";
    public static final String CLUSTER_KEY = "cluster-name";

    /**
     * Maximum pooled connections per host.
     * <p/>
     * Value = {@value}
     */
    public static final int MAX_CONNECTIONS_PER_HOST_DEFAULT = 32;
    public static final String MAX_CONNECTIONS_PER_HOST_KEY = "max-connections-per-host";

    /**
     * Maximum open connections allowed in the pool (counting all hosts).
     * <p/>
     * Value = {@value}
     */
    public static final int MAX_CONNECTIONS_DEFAULT = -1;
    public static final String MAX_CONNECTIONS_KEY = "max-connections";

    /**
     * Maximum number of operations allowed per connection before the connection is closed.
     * <p/>
     * Value = {@value}
     */
    public static final int MAX_OPERATIONS_PER_CONNECTION_DEFAULT = 100 * 1000;
    public static final String MAX_OPERATIONS_PER_CONNECTION_KEY = "max-operations-per-connection";

    /**
     * Maximum pooled "cluster" connections per host.
     * <p/>
     * These connections are mostly idle and only used for DDL operations
     * (like creating keyspaces).  Titan doesn't need many of these connections
     * in ordinary operation.
     */
    public static final int MAX_CLUSTER_CONNECTIONS_PER_HOST_DEFAULT = 3;
    public static final String MAX_CLUSTER_CONNECTIONS_PER_HOST_KEY = "max-cluster-connections-per-host";

    /**
     * How Astyanax discovers Cassandra cluster nodes. This must be one of the
     * values of the Astyanax NodeDiscoveryType enum.
     * <p/>
     * Value = {@value}
     */
    public static final String NODE_DISCOVERY_TYPE_DEFAULT = "RING_DESCRIBE";
    public static final String NODE_DISCOVERY_TYPE_KEY = "node-discovery-type";

    /**
     * Astyanax's connection pooler implementation. This must be one of the
     * values of the Astyanax ConnectionPoolType enum.
     * <p/>
     * Value = {@value}
     */
    public static final String CONNECTION_POOL_TYPE_DEFAULT = "TOKEN_AWARE";
    public static final String CONNECTION_POOL_TYPE_KEY = "connection-pool-type";

    /**
     * This must be the fully-qualified classname (i.e. the complete package
     * name, a dot, and then the class name) of an implementation of Astyanax's
     * RetryPolicy interface. This string may be followed by a sequence of
     * integers, separated from the full classname and from each other by
     * commas; in this case, the integers are cast to native Java ints and
     * passed to the class constructor as arguments.
     * <p/>
     * Value = {@value}
     */
    public static final String RETRY_POLICY_DEFAULT = "com.netflix.astyanax.retry.BoundedExponentialBackoff,100,25000,8";
    public static final String RETRY_POLICY_KEY = "retry-policy";

    private final String clusterName;

    private final AstyanaxContext<Keyspace> keyspaceContext;
    private final AstyanaxContext<Cluster> clusterContext;

    private final RetryPolicy retryPolicy;

    private final Map<String, AstyanaxOrderedKeyColumnValueStore> openStores;

    public AstyanaxStoreManager(Configuration config) throws StorageException {
        super(config);

        // Check if we have non-default thrift frame size or max message size set and warn users
        // because there is nothing we can do in Astyanax to apply those, warning is good enough here
        // otherwise it would make bad user experience if we don't warn at all or crash on this.
        if (config.containsKey(THRIFT_FRAME_SIZE_MB))
            log.warn("Couldn't set custom Thrift Frame Size property, use 'cassandrathrift' instead.");

        this.clusterName = config.getString(CLUSTER_KEY, CLUSTER_DEFAULT);

        this.retryPolicy = getRetryPolicy(config.getString(RETRY_POLICY_KEY, RETRY_POLICY_DEFAULT));

        final int maxConnsPerHost =
                config.getInt(
                        MAX_CONNECTIONS_PER_HOST_KEY,
                        MAX_CONNECTIONS_PER_HOST_DEFAULT);

        final int maxClusterConnsPerHost =
                config.getInt(
                        MAX_CLUSTER_CONNECTIONS_PER_HOST_KEY,
                        MAX_CLUSTER_CONNECTIONS_PER_HOST_DEFAULT);

        this.clusterContext = createCluster(getContextBuilder(config, maxClusterConnsPerHost, "Cluster"));

        ensureKeyspaceExists(clusterContext.getClient());

        this.keyspaceContext = getContextBuilder(config, maxConnsPerHost, "Keyspace").buildKeyspace(ThriftFamilyFactory.getInstance());
        this.keyspaceContext.start();

        openStores = new HashMap<String, AstyanaxOrderedKeyColumnValueStore>(8);
    }

    @Override
    public Deployment getDeployment() {
        return Deployment.REMOTE; // TODO
    }

{'fraction_non_alphanumeric': 0.08469123714945101, 'fraction_numerical': 0.003217008182390377, 'mean_word_length': 3.8229342327150087, 'pattern_counts': {'":': 0, '<': 27, '<?xml version=': 0, '>': 27, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, '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': '7493609', 'n_tokens_mistral': 4233, 'n_tokens_neox': 4063, 'n_words': 1010}