class 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);

            // Set up the client for secure (SSL/TLS communications) if configured to do so
            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;
            }

            // Create the cluster and connect to it
            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);
        }
    }

}