private void configureHosts() {
    final String hostConfiguration = (String) configuration.get(TRIDENT_CASSANDRA_CQL_HOSTS);
    final String[] hosts = hostConfiguration.split(",");
    final List<InetSocketAddress> sockets = new ArrayList<InetSocketAddress>();
    for (final String host : hosts) {
        if(StringUtils.contains(host, ":")) {
            final String hostParts [] = StringUtils.split(host, ":");
            sockets.add(new InetSocketAddress(hostParts[0], Integer.valueOf(hostParts[1])));
            LOG.debug("Configuring [" + host + "] with port [" + hostParts[1] + "]");
        } else {
            sockets.add(new InetSocketAddress(host, ProtocolOptions.DEFAULT_PORT));
            LOG.debug("Configuring [" + host + "] with port [" + ProtocolOptions.DEFAULT_PORT + "]");
        }
    }
    builder = builder.addContactPointsWithPorts(sockets);
}
 

@Test
public void testSaveToCassandra() throws Exception {
    LOGGER.debug("Connecting to Cassandra Quorum: " + conf.getStringList("cassandra.hosts").toString());

    SaveToCassandraActionExecutionFunction func = new SaveToCassandraActionExecutionFunction(
            getHostsStringFromList(conf.getStringList("cassandra.hosts")), ProtocolOptions.DEFAULT_PORT, 50,  BatchStatement.Type.UNLOGGED);

    List<StratioStreamingMessage> list = new ArrayList<StratioStreamingMessage>();
    message.setColumns(StreamsHelper.COLUMNS3);
    list.add(message);

    Exception ex = null;
    try {
        func.process(list);

    } catch (Exception e) {
        ex = e;
        ex.printStackTrace();
    }

    assertNull("Expected null value", ex);
}
 

/**
 * Uses a Cluster.Builder to create a Cassandra cluster reference using the given parameters
 *
 * @param contactPoints The contact points (hostname:port list of Cassandra nodes)
 * @param sslContext    The SSL context (used for secure connections)
 * @param username      The username for connection authentication
 * @param password      The password for connection authentication
 * @param compressionType Enable compression at transport-level requests and responses.
 * @return A reference to the Cluster object associated with the given Cassandra configuration
 */
protected Cluster createCluster(List<InetSocketAddress> contactPoints, SSLContext sslContext,
                                String username, String password, String compressionType) {
    Cluster.Builder builder = Cluster.builder().addContactPointsWithPorts(contactPoints);
    if (sslContext != null) {
        JdkSSLOptions sslOptions = JdkSSLOptions.builder()
                .withSSLContext(sslContext)
                .build();
        builder = builder.withSSL(sslOptions);
        if(ProtocolOptions.Compression.SNAPPY.equals(compressionType)) {
            builder = builder.withCompression(ProtocolOptions.Compression.SNAPPY);
        } else if(ProtocolOptions.Compression.LZ4.equals(compressionType)) {
            builder = builder.withCompression(ProtocolOptions.Compression.LZ4);
        }
    }
    if (username != null && password != null) {
        builder = builder.withCredentials(username, password);
    }
    return builder.build();
}
 

/**
 * Sets compression algorithm to use for the transport.
 *
 * @param compression Compression algorithm.
 */
public void setCompression(String compression) {
    this.compression = compression == null || compression.trim().isEmpty() ? null : compression.trim();

    try {
        if (this.compression != null)
            ProtocolOptions.Compression.valueOf(this.compression);
    }
    catch (Throwable e) {
        throw new IgniteException("Incorrect compression '" + compression + "' specified for Cassandra connection", e);
    }

    invalidate();
}
 

public CassandraMetricBatch(Metadata metadata, ProtocolOptions protocol, CodecRegistry codec,
    TokenAwarePolicy lbPolicy, int batchLimit) {
  this.protocol = protocol;
  this.codec = codec;
  this.metadata = metadata;
  this.policy = lbPolicy;
  metricQueries = new HashMap<>();
  this.batchLimit = batchLimit;

  metricQueries = new HashMap<>();
  dimensionQueries = new HashMap<>();
  dimensionMetricQueries = new HashMap<>();
  metricDimensionQueries = new HashMap<>();
  measurementQueries = new HashMap<>();
}
 

public static Cluster getInputCluster(String[] hosts, Configuration conf)
{
    int port = getInputNativePort(conf);
    Optional<AuthProvider> authProvider = getAuthProvider(conf);
    Optional<SSLOptions> sslOptions = getSSLOptions(conf);
    Optional<Integer> protocolVersion = getProtocolVersion(conf);
    LoadBalancingPolicy loadBalancingPolicy = getReadLoadBalancingPolicy(conf, hosts);
    SocketOptions socketOptions = getReadSocketOptions(conf);
    QueryOptions queryOptions = getReadQueryOptions(conf);
    PoolingOptions poolingOptions = getReadPoolingOptions(conf);
    
    Cluster.Builder builder = Cluster.builder()
                                     .addContactPoints(hosts)
                                     .withPort(port)
                                     .withCompression(ProtocolOptions.Compression.NONE);

    if (authProvider.isPresent())
        builder.withAuthProvider(authProvider.get());
    if (sslOptions.isPresent())
        builder.withSSL(sslOptions.get());

    if (protocolVersion.isPresent()) {
        builder.withProtocolVersion(protocolVersion.get());
    }
    builder.withLoadBalancingPolicy(loadBalancingPolicy)
           .withSocketOptions(socketOptions)
           .withQueryOptions(queryOptions)
           .withPoolingOptions(poolingOptions);

    return builder.build();
}
 

public ConstructorConfiguredCqlClientFactory(String hosts, String clusterName, ConsistencyLevel clusterConsistency,
                        ConsistencyLevel conditionalUpdateConsistency, ProtocolOptions.Compression compression) {
    this.hosts = hosts.split(",");
    this.clusterConsistencyLevel = clusterConsistency;
    if (conditionalUpdateConsistency != null){
        this.serialConsistencyLevel = conditionalUpdateConsistency;
    }
    if (clusterName != null) {
        this.clusterName = clusterName;
    }
    this.compression = compression;
}
 

public Cluster.Builder getClusterBuilder() {

        final List<InetSocketAddress> sockets = new ArrayList<InetSocketAddress>();
        for (String host : hosts) {
            if(StringUtils.contains(host, ":")) {
                String hostParts [] = StringUtils.split(host, ":");
                sockets.add(new InetSocketAddress(hostParts[0], Integer.valueOf(hostParts[1])));
                LOG.debug("Connecting to [" + host + "] with port [" + hostParts[1] + "]");
            } else {
                sockets.add(new InetSocketAddress(host, ProtocolOptions.DEFAULT_PORT));
                LOG.debug("Connecting to [" + host + "] with port [" + ProtocolOptions.DEFAULT_PORT + "]");
            }
        }

        Cluster.Builder builder = Cluster.builder().addContactPointsWithPorts(sockets).withCompression(compression);
        QueryOptions queryOptions = new QueryOptions();
        queryOptions.setConsistencyLevel(clusterConsistencyLevel);
        queryOptions.setSerialConsistencyLevel(serialConsistencyLevel);
        builder = builder.withQueryOptions(queryOptions);

        if (StringUtils.isNotEmpty(clusterName)) {
            builder = builder.withClusterName(clusterName);
        }

        return builder;

    }
 

@Test
public void testGetClusterBuilder() throws Exception {
    final Map<Object, Object> configuration = new HashMap<Object,Object>();
    configuration.put(MapConfiguredCqlClientFactory.TRIDENT_CASSANDRA_CQL_HOSTS, HOSTS);
    configuration.put(MapConfiguredCqlClientFactory.TRIDENT_CASSANDRA_CLUSTER_NAME, CLUSTER_NAME);
    configuration.put(MapConfiguredCqlClientFactory.TRIDENT_CASSANDRA_READ_TIMEOUT, READ_TIMEOUT);
    configuration.put(MapConfiguredCqlClientFactory.TRIDENT_CASSANDRA_CONNECT_TIMEOUT, CONNECT_TIMEOUT);
    configuration.put(MapConfiguredCqlClientFactory.TRIDENT_CASSANDRA_LOCAL_DATA_CENTER_NAME, DATA_CENTER_NAME);
    configuration.put(MapConfiguredCqlClientFactory.TRIDENT_CASSANDRA_CONSISTENCY, DEFAULT_CONSISTENCY_LEVEL.name());
    configuration.put(MapConfiguredCqlClientFactory.TRIDENT_CASSANDRA_SERIAL_CONSISTENCY, DEFAULT_SERIAL_CONSISTENCY_LEVEL.name());

    final CqlClientFactory factory =
            new MapConfiguredCqlClientFactory(configuration);

    final Cluster.Builder clusterBuilder = factory.getClusterBuilder();
    Assert.assertEquals(CLUSTER_NAME, clusterBuilder.getClusterName());
    final InetSocketAddress first = clusterBuilder.getContactPoints().get(0);
    final InetSocketAddress second = clusterBuilder.getContactPoints().get(1);
    Assert.assertEquals("localhost", first.getHostName());
    Assert.assertEquals(9042, first.getPort());
    Assert.assertEquals("remotehost", second.getHostName());
    Assert.assertEquals(1234, second.getPort());
    Assert.assertEquals(Integer.parseInt(CONNECT_TIMEOUT), clusterBuilder.getConfiguration().getSocketOptions().getConnectTimeoutMillis());
    Assert.assertEquals(Integer.parseInt(READ_TIMEOUT), clusterBuilder.getConfiguration().getSocketOptions().getReadTimeoutMillis());
    Assert.assertEquals(DEFAULT_CONSISTENCY_LEVEL, clusterBuilder.getConfiguration().getQueryOptions().getConsistencyLevel());
    Assert.assertEquals(DEFAULT_SERIAL_CONSISTENCY_LEVEL, clusterBuilder.getConfiguration().getQueryOptions().getSerialConsistencyLevel());
    Assert.assertEquals(ProtocolOptions.Compression.NONE, clusterBuilder.getConfiguration().getProtocolOptions().getCompression());
}
 

private Cluster createCluster(List<InetSocketAddress> contactPoints, SSLContext sslContext,
                              String username, String password, String compressionType,
                              Optional<Integer> readTimeoutMillisOptional, Optional<Integer> connectTimeoutMillisOptional) {
    Cluster.Builder builder = Cluster.builder().addContactPointsWithPorts(contactPoints);

    if (sslContext != null) {
        JdkSSLOptions sslOptions = JdkSSLOptions.builder()
                .withSSLContext(sslContext)
                .build();
        builder = builder.withSSL(sslOptions);
    }

    if (username != null && password != null) {
        builder = builder.withCredentials(username, password);
    }

    if(ProtocolOptions.Compression.SNAPPY.equals(compressionType)) {
        builder = builder.withCompression(ProtocolOptions.Compression.SNAPPY);
    } else if(ProtocolOptions.Compression.LZ4.equals(compressionType)) {
        builder = builder.withCompression(ProtocolOptions.Compression.LZ4);
    }

    SocketOptions socketOptions = new SocketOptions();
    readTimeoutMillisOptional.ifPresent(socketOptions::setReadTimeoutMillis);
    connectTimeoutMillisOptional.ifPresent(socketOptions::setConnectTimeoutMillis);

    builder.withSocketOptions(socketOptions);

    return builder.build();
}
 

@Test
public void shouldCreateClusterWithConfig() throws Exception {

	CassandraServiceInfo info = new CassandraServiceInfo("local",
			Collections.singletonList("127.0.0.1"), 9142);

	CassandraClusterConfig config = new CassandraClusterConfig();
	config.setCompression(ProtocolOptions.Compression.NONE);
	config.setPoolingOptions(new PoolingOptions().setPoolTimeoutMillis(1234));
	config.setQueryOptions(new QueryOptions());
	config.setProtocolVersion(ProtocolVersion.NEWEST_SUPPORTED);
	config.setLoadBalancingPolicy(new RoundRobinPolicy());
	config.setReconnectionPolicy(new ConstantReconnectionPolicy(1));
	config.setRetryPolicy(DowngradingConsistencyRetryPolicy.INSTANCE);
	config.setSocketOptions(new SocketOptions());

	Cluster cluster = creator.create(info, config);

	Configuration configuration = cluster.getConfiguration();

	assertThat(configuration.getProtocolOptions().getCompression(),
			is(config.getCompression()));
	assertThat(configuration.getQueryOptions(), is(config.getQueryOptions()));
	assertThat(configuration.getSocketOptions(), is(config.getSocketOptions()));

	Policies policies = configuration.getPolicies();
	assertThat(policies.getLoadBalancingPolicy(),
			is(config.getLoadBalancingPolicy()));
	assertThat(policies.getReconnectionPolicy(), is(config.getReconnectionPolicy()));
	assertThat(policies.getRetryPolicy(), is(config.getRetryPolicy()));
}
 

/**
	 * 描述: 初始化配置
	 * 时间: 2017年11月15日 上午11:25:07
	 * @author yi.zhang
	 * @param servers	服务地址
	 * @param keyspace	命名空间
	 * @param username	账号
	 * @param password	密码
	 */
	public void init(String servers,String keyspace,String username,String password) {
		try {
			// socket 链接配置
			SocketOptions socket = new SocketOptions();
			socket.setKeepAlive(true);
			socket.setReceiveBufferSize(1024* 1024);
			socket.setSendBufferSize(1024* 1024);
			socket.setConnectTimeoutMillis(5 * 1000);
			socket.setReadTimeoutMillis(1000);
			//设置连接池
			PoolingOptions pool = new PoolingOptions();
			// pool.setMaxRequestsPerConnection(HostDistance.LOCAL, 32);
			// pool.setMaxRequestsPerConnection(HostDistance.REMOTE, 32);
			// pool.setCoreConnectionsPerHost(HostDistance.LOCAL, 2);
			// pool.setCoreConnectionsPerHost(HostDistance.REMOTE, 2);
			// pool.setMaxConnectionsPerHost(HostDistance.LOCAL, 4);
			// pool.setMaxConnectionsPerHost(HostDistance.REMOTE, 4);
			pool.setHeartbeatIntervalSeconds(60);
			pool.setIdleTimeoutSeconds(120);
			pool.setPoolTimeoutMillis(5 * 1000);
			List<InetSocketAddress> saddress = new ArrayList<InetSocketAddress>();
			if (servers != null && !"".equals(servers)) {
				for (String server : servers.split(",")) {
					String[] address = server.split(":");
					String ip = address[0];
					int port = 9042;
					if (address != null && address.length > 1) {
						port = Integer.valueOf(address[1]);
					}
					saddress.add(new InetSocketAddress(ip, port));
				}
			}
			InetSocketAddress[] addresses = new InetSocketAddress[saddress.size()];
			saddress.toArray(addresses);
			
			Builder builder = Cluster.builder();
	        builder.withSocketOptions(socket);
	        // 设置压缩方式
	        builder.withCompression(ProtocolOptions.Compression.LZ4);
	        // 负载策略
//	        DCAwareRoundRobinPolicy loadBalance = DCAwareRoundRobinPolicy.builder().withLocalDc("localDc").withUsedHostsPerRemoteDc(2).allowRemoteDCsForLocalConsistencyLevel().build();
//	        builder.withLoadBalancingPolicy(loadBalance);
	        // 重试策略
	        builder.withRetryPolicy(DefaultRetryPolicy.INSTANCE);
			builder.withPoolingOptions(pool);
			builder.addContactPointsWithPorts(addresses);
			builder.withCredentials(username, password);
			Cluster cluster = builder.build();
			if (keyspace != null && !"".equals(keyspace)) {
				session = cluster.connect(keyspace);
			} else {
				session = cluster.connect();
			}
			mapping = new MappingManager(session);
		} catch (Exception e) {
			logger.error("-----Cassandra Config init Error-----", e);
		}
	}
 

/**
 * Creates Cassandra session wrapper if it wasn't created yet and returns it
 *
 * @param log logger
 * @return Cassandra session wrapper
 */
public synchronized CassandraSession session(IgniteLogger log) {
    if (ses != null)
        return ses;

    Cluster.Builder builder = Cluster.builder();

    if (user != null)
        builder = builder.withCredentials(user, pwd);

    if (port != null)
        builder = builder.withPort(port);

    if (contactPoints != null)
        builder = builder.addContactPoints(contactPoints);

    if (contactPointsWithPorts != null)
        builder = builder.addContactPointsWithPorts(contactPointsWithPorts);

    if (maxSchemaAgreementWaitSeconds != null)
        builder = builder.withMaxSchemaAgreementWaitSeconds(maxSchemaAgreementWaitSeconds);

    if (protoVer != null)
        builder = builder.withProtocolVersion(ProtocolVersion.fromInt(protoVer));

    if (compression != null) {
        try {
            builder = builder.withCompression(ProtocolOptions.Compression.valueOf(compression.trim().toLowerCase()));
        }
        catch (IllegalArgumentException e) {
            throw new IgniteException("Incorrect compression option '" + compression + "' specified for Cassandra connection", e);
        }
    }

    if (useSSL != null && useSSL)
        builder = builder.withSSL();

    if (sslOptions != null)
        builder = builder.withSSL(sslOptions);

    if (collectMetrix != null && !collectMetrix)
        builder = builder.withoutMetrics();

    if (jmxReporting != null && !jmxReporting)
        builder = builder.withoutJMXReporting();

    if (creds != null)
        builder = builder.withCredentials(creds.getUser(), creds.getPassword());

    if (loadBalancingPlc != null)
        builder = builder.withLoadBalancingPolicy(loadBalancingPlc);

    if (reconnectionPlc != null)
        builder = builder.withReconnectionPolicy(reconnectionPlc);

    if (retryPlc != null)
        builder = builder.withRetryPolicy(retryPlc);

    if (addrTranslator != null)
        builder = builder.withAddressTranslator(addrTranslator);

    if (speculativeExecutionPlc != null)
        builder = builder.withSpeculativeExecutionPolicy(speculativeExecutionPlc);

    if (authProvider != null)
        builder = builder.withAuthProvider(authProvider);

    if (poolingOptions != null)
        builder = builder.withPoolingOptions(poolingOptions);

    if (sockOptions != null)
        builder = builder.withSocketOptions(sockOptions);

    if (nettyOptions != null)
        builder = builder.withNettyOptions(nettyOptions);

    return ses = new CassandraSessionImpl(
        builder, fetchSize, readConsistency, writeConsistency, sessionExpirationTimeout, log);
}
 

public ProtocolOptions.Compression getCompression() {
    return compression;
}
 

public void setCompression(ProtocolOptions.Compression compression) {
    this.compression = compression;
}
 

public ProtocolOptions getProtocolOptions() {
  return cluster.getConfiguration().getProtocolOptions();
}
 

public ProtocolOptions.Compression compression()
{
    return ProtocolOptions.Compression.valueOf(compression);
}
 

public ConstructorConfiguredCqlClientFactory(String hosts) {
    this(hosts, null, ConsistencyLevel.QUORUM, QueryOptions.DEFAULT_SERIAL_CONSISTENCY_LEVEL, ProtocolOptions.Compression.NONE);
}
 

public ConstructorConfiguredCqlClientFactory(String hosts, ConsistencyLevel clusterConsistency) {
    this(hosts, null, clusterConsistency, QueryOptions.DEFAULT_SERIAL_CONSISTENCY_LEVEL, ProtocolOptions.Compression.NONE);
}
 

private void configureCompression() {
    final String compressionConfiguration = (String) configuration.get(TRIDENT_CASSANDRA_COMPRESSION);
    if (StringUtils.isNotEmpty(compressionConfiguration)) {
        builder = builder.withCompression(ProtocolOptions.Compression.valueOf(compressionConfiguration));
    }
}
 

public Integer getCassandraPort() {
    if (cassandraPort != null)
        return cassandraPort;
    else
        return ProtocolOptions.DEFAULT_PORT;
}
 

@Override
protected void doParse(Element element, ParserContext parserContext,
		BeanDefinitionBuilder builder) {
	super.doParse(element, parserContext, builder);

	Element optionsElement = DomUtils.getChildElementByTagName(element,
			ELEMENT_CASSANDRA_OPTIONS);

	BeanDefinitionBuilder beanBuilder = BeanDefinitionBuilder
			.genericBeanDefinition(CassandraClusterConfig.class.getName());

	if (optionsElement != null) {
		String compressionString = optionsElement.getAttribute(COMPRESSION_ATTRIBUTE);
		if (!StringUtils.isEmpty(compressionString)) {
			ProtocolOptions.Compression compression = ProtocolOptions.Compression
					.valueOf(compressionString);
			beanBuilder.addPropertyValue("compression", compression);
		}

		String retryPolicyString = optionsElement
				.getAttribute(RETRY_POLICY_ATTRIBUTE);
		if (!StringUtils.isEmpty(retryPolicyString)) {
			beanBuilder.addPropertyReference("retryPolicy", retryPolicyString);
		}

		String loadBalancingPolicyString = optionsElement
				.getAttribute(LOAD_BALANCING_POLICY_ATTRIBUTE);
		if (!StringUtils.isEmpty(loadBalancingPolicyString)) {
			beanBuilder.addPropertyReference("loadBalancingPolicy",
					loadBalancingPolicyString);
		}

		String socketOptionsString = optionsElement
				.getAttribute(SOCKET_OPTIONS_ATTRIBUTE);
		if (!StringUtils.isEmpty(socketOptionsString)) {
			beanBuilder.addPropertyReference("socketOptions", socketOptionsString);
		}

		String reconnectionPolicyString = optionsElement
				.getAttribute(RECONNECTION_POLICY_ATTRIBUTE);
		if (!StringUtils.isEmpty(reconnectionPolicyString)) {
			beanBuilder.addPropertyReference("reconnectionPolicy",
					reconnectionPolicyString);
		}

	}

	builder.addConstructorArgValue(beanBuilder.getBeanDefinition());
}