Java 类com.datastax.driver.core.policies.DCAwareRoundRobinPolicy 实例源码

@Override
public void startComponent() {
  if (cluster == null) {
    // Configure and build up the Cassandra cluster.
    cluster = Cluster.builder()
            .withClusterName(clusterName)
            .withPort(port)
            .withRetryPolicy(DefaultRetryPolicy.INSTANCE)
            // TokenAware requires query has routing info (e.g. BoundStatement with all PK value bound).
            .withLoadBalancingPolicy(new TokenAwarePolicy(DCAwareRoundRobinPolicy.builder().build()))
            .addContactPoints(contactPoints.toArray(new String[contactPoints.size()]))
            .build();

    // Register any codecs.
    cluster.getConfiguration().getCodecRegistry()
            .register(new CassandraEnumCodec<>(AccessMode.class, AccessMode.getValueMap()))
            .register(new CassandraEnumCodec<>(Direction.class, Direction.getValueMap()))
            .register(new CassandraEnumCodec<>(SourceEntity.Type.class, SourceEntity.Type.getValueMap()));

    // Create a session.
    manager = new MappingManager(cluster.connect());
  }
}

static Session connect() {
    String contactPoint = "localhost";
    String keySpace = "ks1";

    if(session == null) {

        DCAwareRoundRobinPolicy dcAwarePolicy = new DCAwareRoundRobinPolicy.Builder().build();
        LoadBalancingPolicy policy = new TokenAwarePolicy(dcAwarePolicy);

        cluster = Cluster.builder().addContactPoint(contactPoint)
                .withLoadBalancingPolicy(policy).build();
        cluster.init();
        for (Host host : cluster.getMetadata().getAllHosts()) {
            System.out.printf("Address: %s, Rack: %s, Datacenter: %s, Tokens: %s\n", host.getAddress(),
                    host.getDatacenter(), host.getRack(), host.getTokens());
        }
    }
    return session;
}

private void setup()
throws IOException, KeyStoreException, NoSuchAlgorithmException, KeyManagementException,
              CertificateException, UnrecoverableKeyException  {
// Connect to Cassandra
Cluster.Builder clusterBuilder = Cluster.builder()
    .addContactPoint(host)
    .withPort(port)
    .withLoadBalancingPolicy(new TokenAwarePolicy(DCAwareRoundRobinPolicy.builder().build()));
if (null != username)
    clusterBuilder = clusterBuilder.withCredentials(username, password);
       if (null != truststorePath)
           clusterBuilder = clusterBuilder.withSSL(createSSLOptions());

cluster = clusterBuilder.build();
       if (null == cluster) {
           throw new IOException("Could not create cluster");
       }
session = cluster.connect();
   }

static Cluster buildCluster(Cassandra3Storage cassandra) {
  Cluster.Builder builder = Cluster.builder();
  List<InetSocketAddress> contactPoints = parseContactPoints(cassandra);
  int defaultPort = findConnectPort(contactPoints);
  builder.addContactPointsWithPorts(contactPoints);
  builder.withPort(defaultPort); // This ends up protocolOptions.port
  if (cassandra.username != null && cassandra.password != null) {
    builder.withCredentials(cassandra.username, cassandra.password);
  }
  builder.withRetryPolicy(ZipkinRetryPolicy.INSTANCE);
  builder.withLoadBalancingPolicy(new TokenAwarePolicy(new LatencyAwarePolicy.Builder(
      cassandra.localDc != null
          ? DCAwareRoundRobinPolicy.builder().withLocalDc(cassandra.localDc).build()
          : new RoundRobinPolicy()
      // This can select remote, but LatencyAwarePolicy will prefer local
  ).build()));
  builder.withPoolingOptions(new PoolingOptions().setMaxConnectionsPerHost(
      HostDistance.LOCAL, cassandra.maxConnections
  ));
  return builder.build();
}

static Cluster buildCluster(CassandraStorage cassandra) {
  Cluster.Builder builder = Cluster.builder();
  List<InetSocketAddress> contactPoints = parseContactPoints(cassandra);
  int defaultPort = findConnectPort(contactPoints);
  builder.addContactPointsWithPorts(contactPoints);
  builder.withPort(defaultPort); // This ends up protocolOptions.port
  if (cassandra.username != null && cassandra.password != null) {
    builder.withCredentials(cassandra.username, cassandra.password);
  }
  builder.withRetryPolicy(ZipkinRetryPolicy.INSTANCE);
  builder.withLoadBalancingPolicy(new TokenAwarePolicy(new LatencyAwarePolicy.Builder(
      cassandra.localDc != null
          ? DCAwareRoundRobinPolicy.builder().withLocalDc(cassandra.localDc).build()
          : new RoundRobinPolicy()
      // This can select remote, but LatencyAwarePolicy will prefer local
  ).build()));
  builder.withPoolingOptions(new PoolingOptions().setMaxConnectionsPerHost(
      HostDistance.LOCAL, cassandra.maxConnections
  ));
  return builder.build();
}

public GenericRepair(String inputFile) throws IOException {
    properties =  new HashMap<String, String>();
    loadProperties(inputFile);
    cluster = Cluster.builder()
            .addContactPoint(properties.get("contact_point"))
            .withQueryOptions(new QueryOptions().setFetchSize(Integer.parseInt(properties.get("fetch_size"))))
            .withCredentials(properties.get("username"), properties.get("password"))
            .withSocketOptions(new SocketOptions().setConnectTimeoutMillis(1000000).setReadTimeoutMillis(1000000))
            .withLoadBalancingPolicy(DCAwareRoundRobinPolicy.builder().withLocalDc(properties.get("local_dc"))
                    .build()).build();
    session = cluster.connect();
    scanConsistencyLevel = ConsistencyLevel.valueOf(properties.get("scan_consistency"));
    fetchConsistencyLevel = ConsistencyLevel.ALL;
    this.keyspace = properties.get("keyspace");
    this.tableName = properties.get("table_name");
    this.partitionKey = getPartitionKey();
    this.individualFetchStatement = initializeIndividualFetchStatement();
    this.partitionKeyColumnsToDataTypeName = new DataType.Name[partitionKey.split(",").length];
    queue = new LinkedBlockingQueue<>(Integer.parseInt(properties.get("queue_size")));
    sleepForFailedFetchStatement = Integer.parseInt(properties.get("sleep_millisconds"));
    boundStatement = new BoundStatement(session.prepare("select distinct "+partitionKey+" from "+keyspace+"."+tableName));;
}

/**
 * Get a Cassandra cluster using hosts and port.
 */
private Cluster getCluster(List<String> hosts, int port, String username, String password,
                           String localDc, String consistencyLevel) {
  Cluster.Builder builder = Cluster.builder()
      .addContactPoints(hosts.toArray(new String[0]))
      .withPort(port);

  if (username != null) {
    builder.withAuthProvider(new PlainTextAuthProvider(username, password));
  }

  if (localDc != null) {
    builder.withLoadBalancingPolicy(
        new TokenAwarePolicy(new DCAwareRoundRobinPolicy.Builder().withLocalDc(localDc).build()));
  } else {
    builder.withLoadBalancingPolicy(new TokenAwarePolicy(new RoundRobinPolicy()));
  }

  if (consistencyLevel != null) {
    builder.withQueryOptions(
        new QueryOptions().setConsistencyLevel(ConsistencyLevel.valueOf(consistencyLevel)));
  }

  return builder.build();
}

private void setup()
    throws IOException, KeyStoreException, NoSuchAlgorithmException, KeyManagementException,
           CertificateException, UnrecoverableKeyException  {
    // Connect to Cassandra
    PoolingOptions pOpts = new PoolingOptions();
    pOpts.setCoreConnectionsPerHost(HostDistance.LOCAL, 4);
    pOpts.setMaxConnectionsPerHost(HostDistance.LOCAL, 4);
    Cluster.Builder clusterBuilder = Cluster.builder()
        .addContactPoint(host)
        .withPort(port)
        .withPoolingOptions(pOpts)
        .withLoadBalancingPolicy(new TokenAwarePolicy( DCAwareRoundRobinPolicy.builder().build()));
    if (null != username)
        clusterBuilder = clusterBuilder.withCredentials(username, password);
    if (null != truststorePath)
        clusterBuilder = clusterBuilder.withSSL(createSSLOptions());

    cluster = clusterBuilder.build();
    if (null == cluster) {
        throw new IOException("Could not create cluster");
    }
    session = cluster.connect();
}

protected void createCache(Map<String, String> mapParams) throws Exception {
    final Cluster.Builder bluePrint = Cluster.builder().withClusterName("BluePrint")
            .withQueryOptions(new QueryOptions().setConsistencyLevel(ConsistencyLevel.LOCAL_QUORUM))
            .withRetryPolicy(DefaultRetryPolicy.INSTANCE)
            .withLoadBalancingPolicy(new TokenAwarePolicy(new DCAwareRoundRobinPolicy()))
            .addContactPoint(mapParams.get("cassandra.server.ip.address")).withPort(9042);

    cache1 = mache(String.class, CassandraTestEntity.class)
            .cachedBy(guava())
            .storedIn(cassandra()
                    .withCluster(bluePrint)
                    .withKeyspace(mapParams.get("keyspace.name"))
                    .withSchemaOptions(SchemaOptions.CREATE_SCHEMA_IF_NEEDED)
                    .build())
            .withMessaging(kafka()
                    .withKafkaMqConfig(KafkaMqConfigBuilder.builder()
                            .withZkHost(mapParams.get("kafka.connection"))
                            .build())
                    .withTopic(mapParams.get("kafka.topic"))
                    .build())
            .macheUp();
}

@Override
public LoadBalancingPolicy build() {
    DCAwareRoundRobinPolicy.Builder builder = DCAwareRoundRobinPolicy.builder();

    if (allowRemoteDCsForLocalConsistencyLevel == Boolean.TRUE) {
        builder.allowRemoteDCsForLocalConsistencyLevel();
    }

    if (localDC != null) {
        builder.withLocalDc(localDC);
    }

    if (usedHostsPerRemoteDC != null) {
        builder.withUsedHostsPerRemoteDc(usedHostsPerRemoteDC);
    }

    return builder.build();
}

public static void main(String[] args){
    Cluster cluster;
    Session session;
    cluster = Cluster
            .builder()
            .addContactPoint("127.0.0.1")
            .withRetryPolicy(DefaultRetryPolicy.INSTANCE) //Other option: DowngradingConsistencyRetryPolicy
            .withLoadBalancingPolicy(new TokenAwarePolicy(DCAwareRoundRobinPolicy.builder().build()))
            .build();
    session = cluster.connect("demo");


    PreparedStatement statement = session.prepare("INSERT INTO user (id, name) VALUES (?, ?)");
    Statement boundStatement = statement
            .bind(1, "user 1")
            .enableTracing();

    long startTime = System.currentTimeMillis();
    ResultSet resultSet = session.execute(boundStatement);
    long duration = System.currentTimeMillis() - startTime;
    System.out.format("Time taken: %d", duration);

    ExecutionInfo executionInfo = resultSet.getExecutionInfo();
    printQueryTrace(executionInfo.getQueryTrace());
    cluster.close();

}

@Test
public void loadBalancing_settingLocalDcIgnoresOtherDatacenters() {
  DCAwareRoundRobinPolicy policy = toDCAwareRoundRobinPolicy(
      Cassandra3Storage.builder().localDc("bar").build());

  Host foo = mock(Host.class);
  when(foo.getDatacenter()).thenReturn("foo");
  Host bar = mock(Host.class);
  when(bar.getDatacenter()).thenReturn("bar");
  policy.init(mock(Cluster.class), asList(foo, bar));

  assertThat(policy.distance(foo)).isEqualTo(HostDistance.IGNORED);
  assertThat(policy.distance(bar)).isEqualTo(HostDistance.LOCAL);
}

DCAwareRoundRobinPolicy toDCAwareRoundRobinPolicy(Cassandra3Storage storage) {
  return (DCAwareRoundRobinPolicy) ((LatencyAwarePolicy) ((TokenAwarePolicy) buildCluster(storage)
      .getConfiguration()
      .getPolicies()
      .getLoadBalancingPolicy())
      .getChildPolicy()).getChildPolicy();
}

@Test
public void loadBalancing_settingLocalDcIgnoresOtherDatacenters() {
  DCAwareRoundRobinPolicy policy = toDCAwareRoundRobinPolicy(
      CassandraStorage.builder().localDc("bar").build());

  Host foo = mock(Host.class);
  when(foo.getDatacenter()).thenReturn("foo");
  Host bar = mock(Host.class);
  when(bar.getDatacenter()).thenReturn("bar");
  policy.init(mock(Cluster.class), asList(foo, bar));

  assertThat(policy.distance(foo)).isEqualTo(HostDistance.IGNORED);
  assertThat(policy.distance(bar)).isEqualTo(HostDistance.LOCAL);
}

DCAwareRoundRobinPolicy toDCAwareRoundRobinPolicy(CassandraStorage storage) {
  return (DCAwareRoundRobinPolicy) ((LatencyAwarePolicy) ((TokenAwarePolicy) buildCluster(storage)
      .getConfiguration()
      .getPolicies()
      .getLoadBalancingPolicy())
      .getChildPolicy()).getChildPolicy();
}

protected Cluster getCluster()
{
    Builder cb = Cluster.builder();
    cb.addContactPoints(contactPoints);
    cb.withPort(getPort());

    if (getDataCenter() != null)
    {
        cb.withLoadBalancingPolicy(new DCAwareRoundRobinPolicy(getDataCenter()));
    }

    enrichCluster(cb);
    return cb.build();
}

public JavaDriverClient(StressSettings settings, String host, int port, EncryptionOptions.ClientEncryptionOptions encryptionOptions)
{
    this.host = host;
    this.port = port;
    this.username = settings.mode.username;
    this.password = settings.mode.password;
    this.authProvider = settings.mode.authProvider;
    this.encryptionOptions = encryptionOptions;
    if (settings.node.isWhiteList)
        whitelist = new WhiteListPolicy(new DCAwareRoundRobinPolicy(), settings.node.resolveAll(settings.port.nativePort));
    else
        whitelist = null;
}

private void configureLoadBalancingPolicy() {
    final String dataCenterNameConfiguration = (String) configuration.get(TRIDENT_CASSANDRA_LOCAL_DATA_CENTER_NAME);
    if (StringUtils.isNotEmpty(dataCenterNameConfiguration)) {
        final LoadBalancingPolicy loadBalancingPolicy = new DCAwareRoundRobinPolicy(dataCenterNameConfiguration);
        builder = builder.withLoadBalancingPolicy(loadBalancingPolicy);
    }
}

private Cluster cluster() {
    return Cluster.builder()
                  .addContactPoints(contactPoint).withPort(port)
                  .withCredentials(username, password)
                  .withSocketOptions(new SocketOptions().setKeepAlive(true)
                                                        .setReadTimeoutMillis(readTimeoutMillis))
                  .withLoadBalancingPolicy(DCAwareRoundRobinPolicy.builder()
                                                                  .withLocalDc(localDatacenter)
                                                                  .build())
                  .build();
}

@Test
public void testLoadBalancingPolicyParsing() throws Exception
{
    String lbPolicyStr = "RoundRobinPolicy()";
    System.out.println(lbPolicyStr);
    assertTrue(Utils.parseLbPolicy(lbPolicyStr) instanceof RoundRobinPolicy);
    System.out.println("====================");
    lbPolicyStr = "TokenAwarePolicy(RoundRobinPolicy())";
    System.out.println(lbPolicyStr);
    assertTrue(Utils.parseLbPolicy(lbPolicyStr) instanceof TokenAwarePolicy);
    System.out.println("====================");
    lbPolicyStr = "DCAwareRoundRobinPolicy(\"dc1\")";
    System.out.println(lbPolicyStr);
    assertTrue(Utils.parseLbPolicy(lbPolicyStr) instanceof DCAwareRoundRobinPolicy);
    System.out.println("====================");
    lbPolicyStr = "TokenAwarePolicy(DCAwareRoundRobinPolicy(\"dc1\"))";
    System.out.println(lbPolicyStr);
    assertTrue(Utils.parseLbPolicy(lbPolicyStr) instanceof TokenAwarePolicy);       
    System.out.println("====================");
    lbPolicyStr = "TokenAwarePolicy";
    System.out.println(lbPolicyStr);
    assertTrue(Utils.parseLbPolicy(lbPolicyStr)==null);
    System.out.println("====================");
    lbPolicyStr = "LatencyAwarePolicy(TokenAwarePolicy(RoundRobinPolicy()),(double) 10.5,(long) 1,(long) 10,(long)1,10)";       
    System.out.println(lbPolicyStr);
    assertTrue(Utils.parseLbPolicy(lbPolicyStr) instanceof LatencyAwarePolicy);
    System.out.println("====================");             

}

public JavaDriverClient(StressSettings settings, String host, int port, EncryptionOptions.ClientEncryptionOptions encryptionOptions)
{
    this.protocolVersion = settings.mode.protocolVersion;
    this.host = host;
    this.port = port;
    this.username = settings.mode.username;
    this.password = settings.mode.password;
    this.authProvider = settings.mode.authProvider;
    this.encryptionOptions = encryptionOptions;
    if (settings.node.isWhiteList)
        whitelist = new WhiteListPolicy(DCAwareRoundRobinPolicy.builder().build(), settings.node.resolveAll(settings.port.nativePort));
    else
        whitelist = null;
    connectionsPerHost = settings.mode.connectionsPerHost == null ? 8 : settings.mode.connectionsPerHost;

    int maxThreadCount = 0;
    if (settings.rate.auto)
        maxThreadCount = settings.rate.maxThreads;
    else
        maxThreadCount = settings.rate.threadCount;

    //Always allow enough pending requests so every thread can have a request pending
    //See https://issues.apache.org/jira/browse/CASSANDRA-7217
    int requestsPerConnection = (maxThreadCount / connectionsPerHost) + connectionsPerHost;

    maxPendingPerConnection = settings.mode.maxPendingPerConnection == null ? Math.max(128, requestsPerConnection ) : settings.mode.maxPendingPerConnection;
}

@Test
public void buildsPolicyWithNoParams() throws Exception {
    final DCAwareRoundRobinPolicyFactory factory = new DCAwareRoundRobinPolicyFactory();

    final LoadBalancingPolicy policy = factory.build();

    assertThat(policy).isExactlyInstanceOf(DCAwareRoundRobinPolicy.class);
}

@Test
public void buildsPolicyWithAllParams() throws Exception {
    final DCAwareRoundRobinPolicyFactory factory = new DCAwareRoundRobinPolicyFactory();
    factory.setLocalDC("dc1");
    factory.setUsedHostsPerRemoteDC(1);
    factory.setAllowRemoteDCsForLocalConsistencyLevel(true);

    final LoadBalancingPolicy policy = factory.build();

    assertThat(policy).isExactlyInstanceOf(DCAwareRoundRobinPolicy.class);
}

public JavaDriverClient(StressSettings settings, String host, int port, EncryptionOptions.ClientEncryptionOptions encryptionOptions)
{
    this.host = host;
    this.port = port;
    this.encryptionOptions = encryptionOptions;
    if (settings.node.isWhiteList)
        whitelist = new WhiteListPolicy(new DCAwareRoundRobinPolicy(), settings.node.resolveAll(settings.port.nativePort));
    else
        whitelist = null;
}

protected static Cluster connect(final String node, final int port, final String clusterName) {
    final Cluster cluster = Cluster.builder()
            .addContactPoints(node.split(","))
            .withPort(port)
            .withClusterName(clusterName)
            .withLoadBalancingPolicy(new DCAwareRoundRobinPolicy()) //uses the DC of the seed node it connects to!! So one needs to give it the right seed
            //.withLoadBalancingPolicy(new RoundRobinPolicy())
             .build();
    final Metadata metadata = cluster.getMetadata();
    logger.info(String.format("Connected to cluster: %s\n",
            metadata.getClusterName()));
    return cluster;
}

public JavaDriverClient(StressSettings settings, String host, int port, EncryptionOptions.ClientEncryptionOptions encryptionOptions)
{
    this.host = host;
    this.port = port;
    this.username = settings.mode.username;
    this.password = settings.mode.password;
    this.authProvider = settings.mode.authProvider;
    this.encryptionOptions = encryptionOptions;
    if (settings.node.isWhiteList)
        whitelist = new WhiteListPolicy(new DCAwareRoundRobinPolicy(), settings.node.resolveAll(settings.port.nativePort));
    else
        whitelist = null;
}

private void connect(String nodes, String dataCenter, String username, String password) {
    Builder builder = Cluster.builder();
    if (nodes == null || nodes.isEmpty()) {
        throw new RuntimeException(Const.CASS_NODES + " is not defined");
    }
    if (dataCenter != null && !dataCenter.isEmpty()) {
        DCAwareRoundRobinPolicy policy = DCAwareRoundRobinPolicy.builder()
                .withLocalDc(dataCenter)
                .build();
        builder.withLoadBalancingPolicy(policy);
    }
    String[] nodeParts = nodes.split(",");
    for (String node : nodeParts) {
        node = node.trim();
        if (!node.isEmpty()) {
            LOGGER.info("Adding Cassandra node {}", node);
            builder.addContactPoint(node);
        }
    }
    if (username != null && !username.isEmpty() && password != null && !password.isEmpty()) {
        builder.withCredentials(username, password);
    }
    cluster = builder.build();
    Metadata metadata = cluster.getMetadata();
    LOGGER.info("Connected to cluster: {}", metadata.getClusterName());
    for (Host host : metadata.getAllHosts()) {
        LOGGER.info("Datacenter: {} Host: {} Rack: {}", host.getDatacenter(), host.getAddress(), host.getRack());
    }
}

public CassandraSessionManaged build(Environment environment, String localDc) {
    PoolingOptions poolingOptions = new PoolingOptions();
    poolingOptions.setConnectionsPerHost(HostDistance.LOCAL,  3, 5)
        .setConnectionsPerHost(HostDistance.REMOTE, 1, 2);
    final DCAwareRoundRobinPolicy.Builder builder = DCAwareRoundRobinPolicy.builder();
    if (localDc != null) {
        builder.withLocalDc(localDc);
    }
    QueryOptions queryOptions = new QueryOptions();
    queryOptions.setConsistencyLevel(ConsistencyLevel.LOCAL_ONE).setSerialConsistencyLevel(ConsistencyLevel.LOCAL_SERIAL);
    final Cluster cluster = Cluster
        .builder()
        .withRetryPolicy(DefaultRetryPolicy.INSTANCE)
        .withReconnectionPolicy(new ExponentialReconnectionPolicy(10L, 1000L))
        .withQueryOptions(queryOptions)
        .withLoadBalancingPolicy(new TokenAwarePolicy(builder.build()))
        .addContactPoints(getContactPoints().stream().toArray(String[]::new))
        .withPort(getPort())
        .withSpeculativeExecutionPolicy(new ConstantSpeculativeExecutionPolicy(1000, 2))
        .withPoolingOptions(poolingOptions)
        .build();

    cluster.getConfiguration().getCodecRegistry()
        .register(InstantCodec.instance);

    Session session = cluster.connect(getKeySpace());

    CassandraSessionManaged cassandraSessionManaged = new CassandraSessionManaged(cluster, session);
    environment.lifecycle().manage(cassandraSessionManaged);

    return cassandraSessionManaged;
}

private static void connect() {
    Set<InetSocketAddress> dbHosts = ioconfig.getUniqueBinaryTransportHostsAsInetSocketAddresses();

    int readTimeoutMaxRetries = ioconfig.getReadTimeoutMaxRetries();
    int writeTimeoutMaxRetries = ioconfig.getWriteTimeoutMaxRetries();
    int unavailableMaxRetries = ioconfig.getUnavailableMaxRetries();

    CodecRegistry codecRegistry = new CodecRegistry();

    cluster = Cluster.builder()
            .withLoadBalancingPolicy(new TokenAwarePolicy(DCAwareRoundRobinPolicy.builder().withLocalDc(ioconfig.getDatacenterName()).build(), false))
            .withPoolingOptions(getPoolingOptions())
            .withRetryPolicy(new RetryNTimes(readTimeoutMaxRetries, writeTimeoutMaxRetries, unavailableMaxRetries))
            .withCodecRegistry(codecRegistry)
            .withSocketOptions(getSocketOptions())
            .addContactPointsWithPorts(dbHosts)
            .build();

    QueryLogger queryLogger = QueryLogger.builder()
            .withConstantThreshold(5000)
            .build();

    cluster.register(queryLogger);

    if ( LOG.isDebugEnabled() ) {
        logDebugConnectionInfo();
    }

    try {
        session = cluster.connect( CassandraModel.QUOTED_KEYSPACE );
    }
    catch (NoHostAvailableException e){
        // TODO: figure out how to bubble this up
        throw new RuntimeException(e);
    }
}

@Inject
public CassandraCluster(final PersisterConfig config) {

  this.dbConfig = config.getCassandraDbConfiguration();

  QueryOptions qo = new QueryOptions();
  qo.setConsistencyLevel(ConsistencyLevel.valueOf(dbConfig.getConsistencyLevel()));
  qo.setDefaultIdempotence(true);

  String[] contactPoints = dbConfig.getContactPoints();
  int retries = dbConfig.getMaxWriteRetries();
  Builder builder = Cluster.builder().addContactPoints(contactPoints).withPort(dbConfig.getPort());
  builder
      .withSocketOptions(new SocketOptions().setConnectTimeoutMillis(dbConfig.getConnectionTimeout())
          .setReadTimeoutMillis(dbConfig.getReadTimeout()));
  builder.withQueryOptions(qo).withRetryPolicy(new MonascaRetryPolicy(retries, retries, retries));

  lbPolicy = new TokenAwarePolicy(
      DCAwareRoundRobinPolicy.builder().withLocalDc(dbConfig.getLocalDataCenter()).build());
  builder.withLoadBalancingPolicy(lbPolicy);

  String user = dbConfig.getUser();
  if (user != null && !user.isEmpty()) {
    builder.withAuthProvider(new PlainTextAuthProvider(dbConfig.getUser(), dbConfig.getPassword()));
  }
  cluster = builder.build();

  PoolingOptions poolingOptions = cluster.getConfiguration().getPoolingOptions();

  poolingOptions.setConnectionsPerHost(HostDistance.LOCAL, dbConfig.getMaxConnections(),
      dbConfig.getMaxConnections()).setConnectionsPerHost(HostDistance.REMOTE,
          dbConfig.getMaxConnections(), dbConfig.getMaxConnections());

  poolingOptions.setMaxRequestsPerConnection(HostDistance.LOCAL, dbConfig.getMaxRequests())
      .setMaxRequestsPerConnection(HostDistance.REMOTE, dbConfig.getMaxRequests());

  metricsSession = cluster.connect(dbConfig.getKeySpace());

  measurementInsertStmt = metricsSession.prepare(MEASUREMENT_INSERT_CQL).setIdempotent(true);
  measurementUpdateStmt = metricsSession.prepare(MEASUREMENT_UPDATE_CQL).setIdempotent(true);
  metricInsertStmt = metricsSession.prepare(METRICS_INSERT_CQL).setIdempotent(true);
  metricUpdateStmt = metricsSession.prepare(METRICS_UPDATE_CQL).setIdempotent(true);
  dimensionStmt = metricsSession.prepare(DIMENSION_INSERT_CQL).setIdempotent(true);
  dimensionMetricStmt = metricsSession.prepare(DIMENSION_METRIC_INSERT_CQL).setIdempotent(true);
  metricDimensionStmt = metricsSession.prepare(METRIC_DIMENSION_INSERT_CQL).setIdempotent(true);

  retrieveMetricIdStmt = metricsSession.prepare(RETRIEVE_METRIC_ID_CQL).setIdempotent(true);
  retrieveMetricDimensionStmt = metricsSession.prepare(RETRIEVE_METRIC_DIMENSION_CQL)
      .setIdempotent(true);

  alarmsSession = cluster.connect(dbConfig.getKeySpace());

  alarmHistoryInsertStmt = alarmsSession.prepare(INSERT_ALARM_STATE_HISTORY_SQL).setIdempotent(true);

  metricIdCache = CacheBuilder.newBuilder()
      .maximumSize(config.getCassandraDbConfiguration().getDefinitionMaxCacheSize()).build();

  dimensionCache = CacheBuilder.newBuilder()
      .maximumSize(config.getCassandraDbConfiguration().getDefinitionMaxCacheSize()).build();

  metricDimensionCache = CacheBuilder.newBuilder()
      .maximumSize(config.getCassandraDbConfiguration().getDefinitionMaxCacheSize()).build();

  logger.info("loading cached definitions from db");

  ExecutorService executor = Executors.newFixedThreadPool(250);

  //a majority of the ids are for metrics not actively receiving msgs anymore
  //loadMetricIdCache(executor);

  loadDimensionCache();

  loadMetricDimensionCache(executor);

  executor.shutdown();
}

public synchronized Cluster buildCluster(){

        ConsistencyLevel defaultConsistencyLevel;
        try {
            defaultConsistencyLevel = cassandraConfig.getDataStaxReadCl();
        } catch (IllegalArgumentException e){

            logger.error("Unable to parse provided consistency level in property: {}, defaulting to: {}",
                CassandraFig.READ_CL,
                ConsistencyLevel.LOCAL_QUORUM);

            defaultConsistencyLevel = ConsistencyLevel.LOCAL_QUORUM;
        }


        LoadBalancingPolicy loadBalancingPolicy;
        if( !cassandraConfig.getLocalDataCenter().isEmpty() ){

            loadBalancingPolicy = new DCAwareRoundRobinPolicy.Builder()
                .withLocalDc( cassandraConfig.getLocalDataCenter() ).build();
        }else{
            loadBalancingPolicy = new DCAwareRoundRobinPolicy.Builder().build();
        }

        final PoolingOptions poolingOptions = new PoolingOptions()
            .setCoreConnectionsPerHost(HostDistance.LOCAL, cassandraConfig.getConnections())
            .setMaxConnectionsPerHost(HostDistance.LOCAL, cassandraConfig.getConnections())
            .setIdleTimeoutSeconds( cassandraConfig.getPoolTimeout() / 1000 )
            .setPoolTimeoutMillis( cassandraConfig.getPoolTimeout());

        // purposely add a couple seconds to the driver's lower level socket timeouts vs. cassandra timeouts
        final SocketOptions socketOptions = new SocketOptions()
            .setConnectTimeoutMillis( cassandraConfig.getTimeout())
            .setReadTimeoutMillis( cassandraConfig.getTimeout())
            .setKeepAlive(true);

        final QueryOptions queryOptions = new QueryOptions()
            .setConsistencyLevel(defaultConsistencyLevel)
            .setMetadataEnabled(true); // choose whether to have the driver store metadata such as schema info

        Cluster.Builder datastaxCluster = Cluster.builder()
            .withClusterName(cassandraConfig.getClusterName())
            .addContactPoints(cassandraConfig.getHosts().split(","))
            .withMaxSchemaAgreementWaitSeconds(45)
            .withCompression(ProtocolOptions.Compression.LZ4)
            .withLoadBalancingPolicy(loadBalancingPolicy)
            .withPoolingOptions(poolingOptions)
            .withQueryOptions(queryOptions)
            .withSocketOptions(socketOptions)
            .withReconnectionPolicy(Policies.defaultReconnectionPolicy())
            // client side timestamp generation is IMPORTANT; otherwise successive writes are left up to the server
            // to determine the ts and bad network delays, clock sync, etc. can result in bad behaviors
            .withTimestampGenerator(new AtomicMonotonicTimestampGenerator())
            .withProtocolVersion(getProtocolVersion(cassandraConfig.getVersion()));

        // only add auth credentials if they were provided
        if ( !cassandraConfig.getUsername().isEmpty() && !cassandraConfig.getPassword().isEmpty() ){
            datastaxCluster.withCredentials(
                cassandraConfig.getUsername(),
                cassandraConfig.getPassword()
            );
        }


        return datastaxCluster.build();

    }