@Test public void testConnectToCassandra() throws Exception { Cluster cluster = mock(Cluster.class); processor.setCluster(cluster); testRunner.setProperty(AbstractCassandraProcessor.CONSISTENCY_LEVEL, "ONE"); processor.connectToCassandra(testRunner.getProcessContext()); processor.stop(); assertNull(processor.getCluster()); processor.connectToCassandra(testRunner.getProcessContext()); assertEquals("cluster1", processor.getCluster().getMetadata().getClusterName()); }
protected void connectToCassandra(ProcessContext context) { if (cluster.get() == null) { ComponentLog log = getLogger(); final String contactPointList = context.getProperty(CONTACT_POINTS).getValue(); final String consistencyLevel = context.getProperty(CONSISTENCY_LEVEL).getValue(); List<InetSocketAddress> contactPoints = getContactPoints(contactPointList); final SSLContextService sslService = context.getProperty(PROP_SSL_CONTEXT_SERVICE).asControllerService(SSLContextService.class); final String rawClientAuth = context.getProperty(CLIENT_AUTH).getValue(); final SSLContext sslContext; if (sslService != null) { final SSLContextService.ClientAuth clientAuth; if (StringUtils.isBlank(rawClientAuth)) { clientAuth = SSLContextService.ClientAuth.REQUIRED; } else { try { clientAuth = SSLContextService.ClientAuth.valueOf(rawClientAuth); } catch (final IllegalArgumentException iae) { throw new ProviderCreationException(String.format("Unrecognized client auth '%s'. Possible values are [%s]", rawClientAuth, StringUtils.join(SslContextFactory.ClientAuth.values(), ", "))); } } sslContext = sslService.createSSLContext(clientAuth); } else { sslContext = null; } final String username, password; PropertyValue usernameProperty = context.getProperty(USERNAME); PropertyValue passwordProperty = context.getProperty(PASSWORD); if (usernameProperty != null && passwordProperty != null) { username = usernameProperty.getValue(); password = passwordProperty.getValue(); } else { username = null; password = null; } Cluster newCluster = createCluster(contactPoints, sslContext, username, password); PropertyValue keyspaceProperty = context.getProperty(KEYSPACE); final Session newSession; if (keyspaceProperty != null) { newSession = newCluster.connect(keyspaceProperty.getValue()); } else { newSession = newCluster.connect(); } newCluster.getConfiguration().getQueryOptions().setConsistencyLevel(ConsistencyLevel.valueOf(consistencyLevel)); Metadata metadata = newCluster.getMetadata(); log.info("Connected to Cassandra cluster: {}", new Object[]{metadata.getClusterName()}); cluster.set(newCluster); cassandraSession.set(newSession); } }
AbstractCassandraProcessor extends AbstractProcessor { protected void connectToCassandra(ProcessContext context) { if (cluster.get() == null) { ComponentLog log = getLogger(); final String contactPointList = context.getProperty(CONTACT_POINTS).getValue(); final String consistencyLevel = context.getProperty(CONSISTENCY_LEVEL).getValue(); List<InetSocketAddress> contactPoints = getContactPoints(contactPointList); final SSLContextService sslService = context.getProperty(PROP_SSL_CONTEXT_SERVICE).asControllerService(SSLContextService.class); final String rawClientAuth = context.getProperty(CLIENT_AUTH).getValue(); final SSLContext sslContext; if (sslService != null) { final SSLContextService.ClientAuth clientAuth; if (StringUtils.isBlank(rawClientAuth)) { clientAuth = SSLContextService.ClientAuth.REQUIRED; } else { try { clientAuth = SSLContextService.ClientAuth.valueOf(rawClientAuth); } catch (final IllegalArgumentException iae) { throw new ProviderCreationException(String.format("Unrecognized client auth '%s'. Possible values are [%s]", rawClientAuth, StringUtils.join(SslContextFactory.ClientAuth.values(), ", "))); } } sslContext = sslService.createSSLContext(clientAuth); } else { sslContext = null; } final String username, password; PropertyValue usernameProperty = context.getProperty(USERNAME); PropertyValue passwordProperty = context.getProperty(PASSWORD); if (usernameProperty != null && passwordProperty != null) { username = usernameProperty.getValue(); password = passwordProperty.getValue(); } else { username = null; password = null; } Cluster newCluster = createCluster(contactPoints, sslContext, username, password); PropertyValue keyspaceProperty = context.getProperty(KEYSPACE); final Session newSession; if (keyspaceProperty != null) { newSession = newCluster.connect(keyspaceProperty.getValue()); } else { newSession = newCluster.connect(); } newCluster.getConfiguration().getQueryOptions().setConsistencyLevel(ConsistencyLevel.valueOf(consistencyLevel)); Metadata metadata = newCluster.getMetadata(); log.info("Connected to Cassandra cluster: {}", new Object[]{metadata.getClusterName()}); cluster.set(newCluster); cassandraSession.set(newSession); } } }
AbstractCassandraProcessor extends AbstractProcessor { protected void connectToCassandra(ProcessContext context) { if (cluster.get() == null) { ComponentLog log = getLogger(); final String contactPointList = context.getProperty(CONTACT_POINTS).getValue(); final String consistencyLevel = context.getProperty(CONSISTENCY_LEVEL).getValue(); List<InetSocketAddress> contactPoints = getContactPoints(contactPointList); final SSLContextService sslService = context.getProperty(PROP_SSL_CONTEXT_SERVICE).asControllerService(SSLContextService.class); final String rawClientAuth = context.getProperty(CLIENT_AUTH).getValue(); final SSLContext sslContext; if (sslService != null) { final SSLContextService.ClientAuth clientAuth; if (StringUtils.isBlank(rawClientAuth)) { clientAuth = SSLContextService.ClientAuth.REQUIRED; } else { try { clientAuth = SSLContextService.ClientAuth.valueOf(rawClientAuth); } catch (final IllegalArgumentException iae) { throw new ProviderCreationException(String.format("Unrecognized client auth '%s'. Possible values are [%s]", rawClientAuth, StringUtils.join(SslContextFactory.ClientAuth.values(), ", "))); } } sslContext = sslService.createSSLContext(clientAuth); } else { sslContext = null; } final String username, password; PropertyValue usernameProperty = context.getProperty(USERNAME); PropertyValue passwordProperty = context.getProperty(PASSWORD); if (usernameProperty != null && passwordProperty != null) { username = usernameProperty.getValue(); password = passwordProperty.getValue(); } else { username = null; password = null; } Cluster newCluster = createCluster(contactPoints, sslContext, username, password); PropertyValue keyspaceProperty = context.getProperty(KEYSPACE); final Session newSession; if (keyspaceProperty != null) { newSession = newCluster.connect(keyspaceProperty.getValue()); } else { newSession = newCluster.connect(); } newCluster.getConfiguration().getQueryOptions().setConsistencyLevel(ConsistencyLevel.valueOf(consistencyLevel)); Metadata metadata = newCluster.getMetadata(); log.info("Connected to Cassandra cluster: {}", new Object[]{metadata.getClusterName()}); cluster.set(newCluster); cassandraSession.set(newSession); } }  }
AbstractCassandraProcessor extends AbstractProcessor { protected void connectToCassandra(ProcessContext context) { if (cluster.get() == null) { ComponentLog log = getLogger(); final String contactPointList = context.getProperty(CONTACT_POINTS).getValue(); final String consistencyLevel = context.getProperty(CONSISTENCY_LEVEL).getValue(); List<InetSocketAddress> contactPoints = getContactPoints(contactPointList); final SSLContextService sslService = context.getProperty(PROP_SSL_CONTEXT_SERVICE).asControllerService(SSLContextService.class); final String rawClientAuth = context.getProperty(CLIENT_AUTH).getValue(); final SSLContext sslContext; if (sslService != null) { final SSLContextService.ClientAuth clientAuth; if (StringUtils.isBlank(rawClientAuth)) { clientAuth = SSLContextService.ClientAuth.REQUIRED; } else { try { clientAuth = SSLContextService.ClientAuth.valueOf(rawClientAuth); } catch (final IllegalArgumentException iae) { throw new ProviderCreationException(String.format("Unrecognized client auth '%s'. Possible values are [%s]", rawClientAuth, StringUtils.join(SslContextFactory.ClientAuth.values(), ", "))); } } sslContext = sslService.createSSLContext(clientAuth); } else { sslContext = null; } final String username, password; PropertyValue usernameProperty = context.getProperty(USERNAME); PropertyValue passwordProperty = context.getProperty(PASSWORD); if (usernameProperty != null && passwordProperty != null) { username = usernameProperty.getValue(); password = passwordProperty.getValue(); } else { username = null; password = null; } Cluster newCluster = createCluster(contactPoints, sslContext, username, password); PropertyValue keyspaceProperty = context.getProperty(KEYSPACE); final Session newSession; if (keyspaceProperty != null) { newSession = newCluster.connect(keyspaceProperty.getValue()); } else { newSession = newCluster.connect(); } newCluster.getConfiguration().getQueryOptions().setConsistencyLevel(ConsistencyLevel.valueOf(consistencyLevel)); Metadata metadata = newCluster.getMetadata(); log.info("Connected to Cassandra cluster: {}", new Object[]{metadata.getClusterName()}); cluster.set(newCluster); cassandraSession.set(newSession); } }  void stop(); static Schema getUnionFieldType(String dataType); static Schema getSchemaForType(String dataType); static String getPrimitiveAvroTypeFromCassandraType(DataType dataType); static DataType getPrimitiveDataTypeFromString(String dataTypeName); List<InetSocketAddress> getContactPoints(String contactPointList); }
AbstractCassandraProcessor extends AbstractProcessor { protected void connectToCassandra(ProcessContext context) { if (cluster.get() == null) { ComponentLog log = getLogger(); final String contactPointList = context.getProperty(CONTACT_POINTS).getValue(); final String consistencyLevel = context.getProperty(CONSISTENCY_LEVEL).getValue(); List<InetSocketAddress> contactPoints = getContactPoints(contactPointList); final SSLContextService sslService = context.getProperty(PROP_SSL_CONTEXT_SERVICE).asControllerService(SSLContextService.class); final String rawClientAuth = context.getProperty(CLIENT_AUTH).getValue(); final SSLContext sslContext; if (sslService != null) { final SSLContextService.ClientAuth clientAuth; if (StringUtils.isBlank(rawClientAuth)) { clientAuth = SSLContextService.ClientAuth.REQUIRED; } else { try { clientAuth = SSLContextService.ClientAuth.valueOf(rawClientAuth); } catch (final IllegalArgumentException iae) { throw new ProviderCreationException(String.format("Unrecognized client auth '%s'. Possible values are [%s]", rawClientAuth, StringUtils.join(SslContextFactory.ClientAuth.values(), ", "))); } } sslContext = sslService.createSSLContext(clientAuth); } else { sslContext = null; } final String username, password; PropertyValue usernameProperty = context.getProperty(USERNAME); PropertyValue passwordProperty = context.getProperty(PASSWORD); if (usernameProperty != null && passwordProperty != null) { username = usernameProperty.getValue(); password = passwordProperty.getValue(); } else { username = null; password = null; } Cluster newCluster = createCluster(contactPoints, sslContext, username, password); PropertyValue keyspaceProperty = context.getProperty(KEYSPACE); final Session newSession; if (keyspaceProperty != null) { newSession = newCluster.connect(keyspaceProperty.getValue()); } else { newSession = newCluster.connect(); } newCluster.getConfiguration().getQueryOptions().setConsistencyLevel(ConsistencyLevel.valueOf(consistencyLevel)); Metadata metadata = newCluster.getMetadata(); log.info("Connected to Cassandra cluster: {}", new Object[]{metadata.getClusterName()}); cluster.set(newCluster); cassandraSession.set(newSession); } }  void stop(); static Schema getUnionFieldType(String dataType); static Schema getSchemaForType(String dataType); static String getPrimitiveAvroTypeFromCassandraType(DataType dataType); static DataType getPrimitiveDataTypeFromString(String dataTypeName); List<InetSocketAddress> getContactPoints(String contactPointList); static final int DEFAULT_CASSANDRA_PORT; static final PropertyDescriptor CONTACT_POINTS; static final PropertyDescriptor KEYSPACE; static final PropertyDescriptor PROP_SSL_CONTEXT_SERVICE; static final PropertyDescriptor CLIENT_AUTH; static final PropertyDescriptor USERNAME; static final PropertyDescriptor PASSWORD; static final PropertyDescriptor CONSISTENCY_LEVEL; static final PropertyDescriptor CHARSET; }