exceptionfactory commented on code in PR #11032: URL: https://github.com/apache/nifi/pull/11032#discussion_r3770370419
########## nifi-extension-bundles/nifi-cassandra-bundle/nifi-cassandra-services/src/main/java/org/apache/nifi/service/cassandra/CassandraSessionProvider.java: ########## @@ -0,0 +1,340 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.nifi.service.cassandra; + +import com.datastax.oss.driver.api.core.ConsistencyLevel; +import com.datastax.oss.driver.api.core.CqlIdentifier; +import com.datastax.oss.driver.api.core.CqlSession; +import com.datastax.oss.driver.api.core.CqlSessionBuilder; +import com.datastax.oss.driver.api.core.config.DefaultDriverOption; +import com.datastax.oss.driver.api.core.config.DriverConfigLoader; +import com.datastax.oss.driver.api.core.config.ProgrammaticDriverConfigLoaderBuilder; +import com.datastax.oss.driver.api.core.metadata.Metadata; +import com.datastax.oss.driver.api.core.metadata.Node; +import org.apache.nifi.annotation.documentation.CapabilityDescription; +import org.apache.nifi.annotation.documentation.Tags; +import org.apache.nifi.annotation.lifecycle.OnDisabled; +import org.apache.nifi.annotation.lifecycle.OnEnabled; +import org.apache.nifi.cassandra.CassandraClient; +import org.apache.nifi.cassandra.CassandraConnectionService; +import org.apache.nifi.cassandra.CompressionType; +import org.apache.nifi.components.PropertyDescriptor; +import org.apache.nifi.components.PropertyValue; +import org.apache.nifi.controller.AbstractControllerService; +import org.apache.nifi.controller.ConfigurationContext; +import org.apache.nifi.logging.ComponentLog; +import org.apache.nifi.processor.exception.ProcessException; +import org.apache.nifi.processor.util.StandardValidators; +import org.apache.nifi.reporting.InitializationException; +import org.apache.nifi.ssl.SSLContextService; + +import java.net.InetSocketAddress; +import java.time.Duration; +import java.util.ArrayList; +import java.util.List; +import java.util.Optional; +import java.util.concurrent.TimeUnit; +import javax.net.ssl.SSLContext; + +@Tags({ "cassandra", "dbcp", "database", "connection", "pooling" }) +@CapabilityDescription("Provides connection session for Cassandra processors to work with Apache Cassandra.") +public class CassandraSessionProvider extends AbstractControllerService implements CassandraConnectionService { + + public static final int DEFAULT_CASSANDRA_PORT = 9042; + + private static final String[] CONSISTENCY_LEVELS = { + ConsistencyLevel.ANY.name(), + ConsistencyLevel.ONE.name(), + ConsistencyLevel.TWO.name(), + ConsistencyLevel.THREE.name(), + ConsistencyLevel.QUORUM.name(), + ConsistencyLevel.ALL.name(), + ConsistencyLevel.LOCAL_ONE.name(), + ConsistencyLevel.LOCAL_QUORUM.name(), + ConsistencyLevel.EACH_QUORUM.name(), + ConsistencyLevel.SERIAL.name(), + ConsistencyLevel.LOCAL_SERIAL.name() + }; + + public static final PropertyDescriptor CONTACT_POINTS = new PropertyDescriptor.Builder() + .name("Cassandra Contact Points") + .description(""" + Contact points are addresses of Cassandra nodes. The list of contact points should be + comma-separated and in hostname:port format. Example node1:port,node2:port,.... + The default client port for Cassandra is 9042, but the port(s) must be explicitly specified. + """) + .required(true) + .addValidator(StandardValidators.HOSTNAME_PORT_LIST_VALIDATOR) + .build(); + + public static final PropertyDescriptor KEYSPACE = new PropertyDescriptor.Builder() + .name("Keyspace") + .description(""" + The Cassandra Keyspace to connect to. If no keyspace is specified, the query will need to include + keyspace name before any table reference. For query-native processors, or processors that support + the Table property, provide the table as <KEYSPACE>.<TABLE>. + """) + .required(false) + .addValidator(StandardValidators.NON_EMPTY_VALIDATOR) + .build(); + + public static final PropertyDescriptor PROP_SSL_CONTEXT_SERVICE = new PropertyDescriptor.Builder() + .name("SSL Context Service") + .description(""" + The SSL Context Service used to provide the client certificate information for establishing secure + TLS/SSL connections to the Cassandra cluster. + """) + .required(false) + .identifiesControllerService(SSLContextService.class) Review Comment: This should use the more generic `SSLContextProvider` interface. ```suggestion .identifiesControllerService(SSLContextProvider.class) ``` ########## nifi-extension-bundles/nifi-cassandra-bundle/nifi-cassandra-services/src/main/java/org/apache/nifi/service/cassandra/CassandraSessionProvider.java: ########## @@ -0,0 +1,340 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.nifi.service.cassandra; + +import com.datastax.oss.driver.api.core.ConsistencyLevel; +import com.datastax.oss.driver.api.core.CqlIdentifier; +import com.datastax.oss.driver.api.core.CqlSession; +import com.datastax.oss.driver.api.core.CqlSessionBuilder; +import com.datastax.oss.driver.api.core.config.DefaultDriverOption; +import com.datastax.oss.driver.api.core.config.DriverConfigLoader; +import com.datastax.oss.driver.api.core.config.ProgrammaticDriverConfigLoaderBuilder; +import com.datastax.oss.driver.api.core.metadata.Metadata; +import com.datastax.oss.driver.api.core.metadata.Node; +import org.apache.nifi.annotation.documentation.CapabilityDescription; +import org.apache.nifi.annotation.documentation.Tags; +import org.apache.nifi.annotation.lifecycle.OnDisabled; +import org.apache.nifi.annotation.lifecycle.OnEnabled; +import org.apache.nifi.cassandra.CassandraClient; +import org.apache.nifi.cassandra.CassandraConnectionService; +import org.apache.nifi.cassandra.CompressionType; +import org.apache.nifi.components.PropertyDescriptor; +import org.apache.nifi.components.PropertyValue; +import org.apache.nifi.controller.AbstractControllerService; +import org.apache.nifi.controller.ConfigurationContext; +import org.apache.nifi.logging.ComponentLog; +import org.apache.nifi.processor.exception.ProcessException; +import org.apache.nifi.processor.util.StandardValidators; +import org.apache.nifi.reporting.InitializationException; +import org.apache.nifi.ssl.SSLContextService; + +import java.net.InetSocketAddress; +import java.time.Duration; +import java.util.ArrayList; +import java.util.List; +import java.util.Optional; +import java.util.concurrent.TimeUnit; +import javax.net.ssl.SSLContext; + +@Tags({ "cassandra", "dbcp", "database", "connection", "pooling" }) +@CapabilityDescription("Provides connection session for Cassandra processors to work with Apache Cassandra.") +public class CassandraSessionProvider extends AbstractControllerService implements CassandraConnectionService { + + public static final int DEFAULT_CASSANDRA_PORT = 9042; + + private static final String[] CONSISTENCY_LEVELS = { + ConsistencyLevel.ANY.name(), + ConsistencyLevel.ONE.name(), + ConsistencyLevel.TWO.name(), + ConsistencyLevel.THREE.name(), + ConsistencyLevel.QUORUM.name(), + ConsistencyLevel.ALL.name(), + ConsistencyLevel.LOCAL_ONE.name(), + ConsistencyLevel.LOCAL_QUORUM.name(), + ConsistencyLevel.EACH_QUORUM.name(), + ConsistencyLevel.SERIAL.name(), + ConsistencyLevel.LOCAL_SERIAL.name() + }; + + public static final PropertyDescriptor CONTACT_POINTS = new PropertyDescriptor.Builder() + .name("Cassandra Contact Points") + .description(""" + Contact points are addresses of Cassandra nodes. The list of contact points should be + comma-separated and in hostname:port format. Example node1:port,node2:port,.... + The default client port for Cassandra is 9042, but the port(s) must be explicitly specified. + """) + .required(true) + .addValidator(StandardValidators.HOSTNAME_PORT_LIST_VALIDATOR) + .build(); + + public static final PropertyDescriptor KEYSPACE = new PropertyDescriptor.Builder() + .name("Keyspace") + .description(""" + The Cassandra Keyspace to connect to. If no keyspace is specified, the query will need to include + keyspace name before any table reference. For query-native processors, or processors that support + the Table property, provide the table as <KEYSPACE>.<TABLE>. + """) + .required(false) + .addValidator(StandardValidators.NON_EMPTY_VALIDATOR) + .build(); + + public static final PropertyDescriptor PROP_SSL_CONTEXT_SERVICE = new PropertyDescriptor.Builder() + .name("SSL Context Service") + .description(""" + The SSL Context Service used to provide the client certificate information for establishing secure + TLS/SSL connections to the Cassandra cluster. + """) + .required(false) + .identifiesControllerService(SSLContextService.class) + .build(); + + public static final PropertyDescriptor USERNAME = new PropertyDescriptor.Builder() + .name("Username") + .description("Username to access the Cassandra cluster") + .required(false) + .addValidator(StandardValidators.NON_EMPTY_VALIDATOR) + .build(); + + public static final PropertyDescriptor PASSWORD = new PropertyDescriptor.Builder() + .name("Password") + .description("Password to access the Cassandra cluster") + .required(false) + .sensitive(true) + .addValidator(StandardValidators.NON_EMPTY_VALIDATOR) + .build(); + + public static final PropertyDescriptor CONSISTENCY_LEVEL = new PropertyDescriptor.Builder() + .name("Consistency Level") + .description("The strategy for how many replicas must respond before results are returned.") + .required(true) + .allowableValues(CONSISTENCY_LEVELS) + .defaultValue(ConsistencyLevel.QUORUM.name()) + .build(); + + public static final PropertyDescriptor COMPRESSION_TYPE = new PropertyDescriptor.Builder() + .name("Compression Type") + .description(""" + Specifies the compression type used for transport-level requests and responses.Note that Snappy + compression is not supported when using Protocol V5. + """) + .required(false) + .allowableValues(CompressionType.class) + .defaultValue(CompressionType.NONE.getValue()) + .build(); + + public static final PropertyDescriptor READ_TIMEOUT = new PropertyDescriptor.Builder() + .name("Read Timeout") + .description("Read timeout. 0 means no timeout. If no value is set, the underlying default will be used.") + .required(false) Review Comment: A default value should be provided ########## nifi-extension-bundles/nifi-cassandra-bundle/nifi-cassandra-services/src/main/java/org/apache/nifi/service/cassandra/CassandraSessionProvider.java: ########## @@ -0,0 +1,340 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.nifi.service.cassandra; + +import com.datastax.oss.driver.api.core.ConsistencyLevel; +import com.datastax.oss.driver.api.core.CqlIdentifier; +import com.datastax.oss.driver.api.core.CqlSession; +import com.datastax.oss.driver.api.core.CqlSessionBuilder; +import com.datastax.oss.driver.api.core.config.DefaultDriverOption; +import com.datastax.oss.driver.api.core.config.DriverConfigLoader; +import com.datastax.oss.driver.api.core.config.ProgrammaticDriverConfigLoaderBuilder; +import com.datastax.oss.driver.api.core.metadata.Metadata; +import com.datastax.oss.driver.api.core.metadata.Node; +import org.apache.nifi.annotation.documentation.CapabilityDescription; +import org.apache.nifi.annotation.documentation.Tags; +import org.apache.nifi.annotation.lifecycle.OnDisabled; +import org.apache.nifi.annotation.lifecycle.OnEnabled; +import org.apache.nifi.cassandra.CassandraClient; +import org.apache.nifi.cassandra.CassandraConnectionService; +import org.apache.nifi.cassandra.CompressionType; +import org.apache.nifi.components.PropertyDescriptor; +import org.apache.nifi.components.PropertyValue; +import org.apache.nifi.controller.AbstractControllerService; +import org.apache.nifi.controller.ConfigurationContext; +import org.apache.nifi.logging.ComponentLog; +import org.apache.nifi.processor.exception.ProcessException; +import org.apache.nifi.processor.util.StandardValidators; +import org.apache.nifi.reporting.InitializationException; +import org.apache.nifi.ssl.SSLContextService; + +import java.net.InetSocketAddress; +import java.time.Duration; +import java.util.ArrayList; +import java.util.List; +import java.util.Optional; +import java.util.concurrent.TimeUnit; +import javax.net.ssl.SSLContext; + +@Tags({ "cassandra", "dbcp", "database", "connection", "pooling" }) +@CapabilityDescription("Provides connection session for Cassandra processors to work with Apache Cassandra.") +public class CassandraSessionProvider extends AbstractControllerService implements CassandraConnectionService { + + public static final int DEFAULT_CASSANDRA_PORT = 9042; + + private static final String[] CONSISTENCY_LEVELS = { + ConsistencyLevel.ANY.name(), + ConsistencyLevel.ONE.name(), + ConsistencyLevel.TWO.name(), + ConsistencyLevel.THREE.name(), + ConsistencyLevel.QUORUM.name(), + ConsistencyLevel.ALL.name(), + ConsistencyLevel.LOCAL_ONE.name(), + ConsistencyLevel.LOCAL_QUORUM.name(), + ConsistencyLevel.EACH_QUORUM.name(), + ConsistencyLevel.SERIAL.name(), + ConsistencyLevel.LOCAL_SERIAL.name() + }; + + public static final PropertyDescriptor CONTACT_POINTS = new PropertyDescriptor.Builder() + .name("Cassandra Contact Points") + .description(""" + Contact points are addresses of Cassandra nodes. The list of contact points should be + comma-separated and in hostname:port format. Example node1:port,node2:port,.... + The default client port for Cassandra is 9042, but the port(s) must be explicitly specified. + """) + .required(true) + .addValidator(StandardValidators.HOSTNAME_PORT_LIST_VALIDATOR) + .build(); + + public static final PropertyDescriptor KEYSPACE = new PropertyDescriptor.Builder() + .name("Keyspace") + .description(""" + The Cassandra Keyspace to connect to. If no keyspace is specified, the query will need to include + keyspace name before any table reference. For query-native processors, or processors that support + the Table property, provide the table as <KEYSPACE>.<TABLE>. + """) + .required(false) + .addValidator(StandardValidators.NON_EMPTY_VALIDATOR) + .build(); + + public static final PropertyDescriptor PROP_SSL_CONTEXT_SERVICE = new PropertyDescriptor.Builder() + .name("SSL Context Service") + .description(""" + The SSL Context Service used to provide the client certificate information for establishing secure + TLS/SSL connections to the Cassandra cluster. + """) + .required(false) + .identifiesControllerService(SSLContextService.class) + .build(); + + public static final PropertyDescriptor USERNAME = new PropertyDescriptor.Builder() + .name("Username") + .description("Username to access the Cassandra cluster") + .required(false) + .addValidator(StandardValidators.NON_EMPTY_VALIDATOR) + .build(); + + public static final PropertyDescriptor PASSWORD = new PropertyDescriptor.Builder() + .name("Password") + .description("Password to access the Cassandra cluster") + .required(false) + .sensitive(true) + .addValidator(StandardValidators.NON_EMPTY_VALIDATOR) + .build(); + + public static final PropertyDescriptor CONSISTENCY_LEVEL = new PropertyDescriptor.Builder() + .name("Consistency Level") + .description("The strategy for how many replicas must respond before results are returned.") + .required(true) + .allowableValues(CONSISTENCY_LEVELS) + .defaultValue(ConsistencyLevel.QUORUM.name()) + .build(); + + public static final PropertyDescriptor COMPRESSION_TYPE = new PropertyDescriptor.Builder() + .name("Compression Type") + .description(""" + Specifies the compression type used for transport-level requests and responses.Note that Snappy + compression is not supported when using Protocol V5. + """) + .required(false) + .allowableValues(CompressionType.class) + .defaultValue(CompressionType.NONE.getValue()) + .build(); + + public static final PropertyDescriptor READ_TIMEOUT = new PropertyDescriptor.Builder() + .name("Read Timeout") + .description("Read timeout. 0 means no timeout. If no value is set, the underlying default will be used.") + .required(false) + .addValidator(StandardValidators.TIME_PERIOD_VALIDATOR) + .build(); + + public static final PropertyDescriptor CONNECT_TIMEOUT = new PropertyDescriptor.Builder() + .name("Connect Timeout") + .description("Connection timeout. 0 means no timeout. If no value is set, the underlying default will be used.") + .required(false) Review Comment: A default value should be provided. ########## nifi-extension-bundles/nifi-cassandra-bundle/nifi-cassandra-services/src/main/java/org/apache/nifi/service/cassandra/CassandraSessionProvider.java: ########## @@ -0,0 +1,340 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.nifi.service.cassandra; + +import com.datastax.oss.driver.api.core.ConsistencyLevel; +import com.datastax.oss.driver.api.core.CqlIdentifier; +import com.datastax.oss.driver.api.core.CqlSession; +import com.datastax.oss.driver.api.core.CqlSessionBuilder; +import com.datastax.oss.driver.api.core.config.DefaultDriverOption; +import com.datastax.oss.driver.api.core.config.DriverConfigLoader; +import com.datastax.oss.driver.api.core.config.ProgrammaticDriverConfigLoaderBuilder; +import com.datastax.oss.driver.api.core.metadata.Metadata; +import com.datastax.oss.driver.api.core.metadata.Node; +import org.apache.nifi.annotation.documentation.CapabilityDescription; +import org.apache.nifi.annotation.documentation.Tags; +import org.apache.nifi.annotation.lifecycle.OnDisabled; +import org.apache.nifi.annotation.lifecycle.OnEnabled; +import org.apache.nifi.cassandra.CassandraClient; +import org.apache.nifi.cassandra.CassandraConnectionService; +import org.apache.nifi.cassandra.CompressionType; +import org.apache.nifi.components.PropertyDescriptor; +import org.apache.nifi.components.PropertyValue; +import org.apache.nifi.controller.AbstractControllerService; +import org.apache.nifi.controller.ConfigurationContext; +import org.apache.nifi.logging.ComponentLog; +import org.apache.nifi.processor.exception.ProcessException; +import org.apache.nifi.processor.util.StandardValidators; +import org.apache.nifi.reporting.InitializationException; +import org.apache.nifi.ssl.SSLContextService; + +import java.net.InetSocketAddress; +import java.time.Duration; +import java.util.ArrayList; +import java.util.List; +import java.util.Optional; +import java.util.concurrent.TimeUnit; +import javax.net.ssl.SSLContext; + +@Tags({ "cassandra", "dbcp", "database", "connection", "pooling" }) +@CapabilityDescription("Provides connection session for Cassandra processors to work with Apache Cassandra.") +public class CassandraSessionProvider extends AbstractControllerService implements CassandraConnectionService { + + public static final int DEFAULT_CASSANDRA_PORT = 9042; + + private static final String[] CONSISTENCY_LEVELS = { + ConsistencyLevel.ANY.name(), + ConsistencyLevel.ONE.name(), + ConsistencyLevel.TWO.name(), + ConsistencyLevel.THREE.name(), + ConsistencyLevel.QUORUM.name(), + ConsistencyLevel.ALL.name(), + ConsistencyLevel.LOCAL_ONE.name(), + ConsistencyLevel.LOCAL_QUORUM.name(), + ConsistencyLevel.EACH_QUORUM.name(), + ConsistencyLevel.SERIAL.name(), + ConsistencyLevel.LOCAL_SERIAL.name() + }; + + public static final PropertyDescriptor CONTACT_POINTS = new PropertyDescriptor.Builder() + .name("Cassandra Contact Points") + .description(""" + Contact points are addresses of Cassandra nodes. The list of contact points should be + comma-separated and in hostname:port format. Example node1:port,node2:port,.... + The default client port for Cassandra is 9042, but the port(s) must be explicitly specified. + """) + .required(true) + .addValidator(StandardValidators.HOSTNAME_PORT_LIST_VALIDATOR) + .build(); + + public static final PropertyDescriptor KEYSPACE = new PropertyDescriptor.Builder() + .name("Keyspace") + .description(""" + The Cassandra Keyspace to connect to. If no keyspace is specified, the query will need to include + keyspace name before any table reference. For query-native processors, or processors that support + the Table property, provide the table as <KEYSPACE>.<TABLE>. + """) + .required(false) + .addValidator(StandardValidators.NON_EMPTY_VALIDATOR) + .build(); + + public static final PropertyDescriptor PROP_SSL_CONTEXT_SERVICE = new PropertyDescriptor.Builder() + .name("SSL Context Service") + .description(""" + The SSL Context Service used to provide the client certificate information for establishing secure + TLS/SSL connections to the Cassandra cluster. + """) + .required(false) + .identifiesControllerService(SSLContextService.class) + .build(); + + public static final PropertyDescriptor USERNAME = new PropertyDescriptor.Builder() + .name("Username") + .description("Username to access the Cassandra cluster") + .required(false) + .addValidator(StandardValidators.NON_EMPTY_VALIDATOR) + .build(); + + public static final PropertyDescriptor PASSWORD = new PropertyDescriptor.Builder() + .name("Password") + .description("Password to access the Cassandra cluster") + .required(false) + .sensitive(true) + .addValidator(StandardValidators.NON_EMPTY_VALIDATOR) + .build(); + + public static final PropertyDescriptor CONSISTENCY_LEVEL = new PropertyDescriptor.Builder() + .name("Consistency Level") + .description("The strategy for how many replicas must respond before results are returned.") + .required(true) + .allowableValues(CONSISTENCY_LEVELS) + .defaultValue(ConsistencyLevel.QUORUM.name()) + .build(); + + public static final PropertyDescriptor COMPRESSION_TYPE = new PropertyDescriptor.Builder() + .name("Compression Type") + .description(""" + Specifies the compression type used for transport-level requests and responses.Note that Snappy + compression is not supported when using Protocol V5. + """) + .required(false) + .allowableValues(CompressionType.class) + .defaultValue(CompressionType.NONE.getValue()) + .build(); + + public static final PropertyDescriptor READ_TIMEOUT = new PropertyDescriptor.Builder() + .name("Read Timeout") + .description("Read timeout. 0 means no timeout. If no value is set, the underlying default will be used.") + .required(false) + .addValidator(StandardValidators.TIME_PERIOD_VALIDATOR) + .build(); + + public static final PropertyDescriptor CONNECT_TIMEOUT = new PropertyDescriptor.Builder() + .name("Connect Timeout") + .description("Connection timeout. 0 means no timeout. If no value is set, the underlying default will be used.") + .required(false) + .addValidator(StandardValidators.TIME_PERIOD_VALIDATOR) + .build(); + + public static final PropertyDescriptor LOCAL_DATACENTER = new PropertyDescriptor.Builder() + .name("Local Datacenter") + .description("The local datacenter name for the Cassandra cluster (required by driver 4.x).") + .required(true) + .addValidator(StandardValidators.NON_EMPTY_VALIDATOR) + .build(); + + private CqlSession cassandraSession; + private CassandraClient cassandraClient; + private String clusterAddress; Review Comment: These should be defined after all static variables ########## nifi-extension-bundles/nifi-cassandra-bundle/nifi-cassandra-services/src/test/java/org/apache/nifi/service/cassandra/TestCassandraSessionProvider.java: ########## @@ -0,0 +1,108 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.nifi.service.cassandra; + +import org.apache.nifi.cassandra.CassandraClient; +import org.apache.nifi.components.PropertyDescriptor; +import org.apache.nifi.controller.ConfigurationContext; +import org.apache.nifi.reporting.InitializationException; +import org.apache.nifi.util.NoOpProcessor; +import org.apache.nifi.util.TestRunner; +import org.apache.nifi.util.TestRunners; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; + +import java.util.List; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.Mockito.doNothing; +import static org.mockito.Mockito.doReturn; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.spy; +import static org.mockito.Mockito.times; +import static org.mockito.Mockito.verify; + +class TestCassandraSessionProvider { + + private static final String SERVICE_ID = "cassandra-session-provider"; + + private static TestRunner runner; + private static CassandraSessionProvider sessionProvider; + private static CassandraClient mockClient; + + @BeforeEach + void setup() throws InitializationException { + sessionProvider = spy(new CassandraSessionProvider()); + doNothing().when(sessionProvider).connectToCassandra(any(ConfigurationContext.class)); + mockClient = mock(CassandraClient.class); + doReturn(mockClient).when(sessionProvider).getClient(); + + runner = TestRunners.newTestRunner(NoOpProcessor.class); + runner.setValidateExpressionUsage(false); + runner.addControllerService(SERVICE_ID, sessionProvider); + } + + @Test + void testGetPropertyDescriptors() { Review Comment: This test is not necessary and should be removed ########## nifi-extension-bundles/nifi-cassandra-bundle/nifi-cassandra-services/src/main/java/org/apache/nifi/service/cassandra/CassandraSessionProvider.java: ########## @@ -0,0 +1,340 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.nifi.service.cassandra; + +import com.datastax.oss.driver.api.core.ConsistencyLevel; +import com.datastax.oss.driver.api.core.CqlIdentifier; +import com.datastax.oss.driver.api.core.CqlSession; +import com.datastax.oss.driver.api.core.CqlSessionBuilder; +import com.datastax.oss.driver.api.core.config.DefaultDriverOption; +import com.datastax.oss.driver.api.core.config.DriverConfigLoader; +import com.datastax.oss.driver.api.core.config.ProgrammaticDriverConfigLoaderBuilder; +import com.datastax.oss.driver.api.core.metadata.Metadata; +import com.datastax.oss.driver.api.core.metadata.Node; +import org.apache.nifi.annotation.documentation.CapabilityDescription; +import org.apache.nifi.annotation.documentation.Tags; +import org.apache.nifi.annotation.lifecycle.OnDisabled; +import org.apache.nifi.annotation.lifecycle.OnEnabled; +import org.apache.nifi.cassandra.CassandraClient; +import org.apache.nifi.cassandra.CassandraConnectionService; +import org.apache.nifi.cassandra.CompressionType; +import org.apache.nifi.components.PropertyDescriptor; +import org.apache.nifi.components.PropertyValue; +import org.apache.nifi.controller.AbstractControllerService; +import org.apache.nifi.controller.ConfigurationContext; +import org.apache.nifi.logging.ComponentLog; +import org.apache.nifi.processor.exception.ProcessException; +import org.apache.nifi.processor.util.StandardValidators; +import org.apache.nifi.reporting.InitializationException; +import org.apache.nifi.ssl.SSLContextService; + +import java.net.InetSocketAddress; +import java.time.Duration; +import java.util.ArrayList; +import java.util.List; +import java.util.Optional; +import java.util.concurrent.TimeUnit; +import javax.net.ssl.SSLContext; + +@Tags({ "cassandra", "dbcp", "database", "connection", "pooling" }) +@CapabilityDescription("Provides connection session for Cassandra processors to work with Apache Cassandra.") +public class CassandraSessionProvider extends AbstractControllerService implements CassandraConnectionService { + + public static final int DEFAULT_CASSANDRA_PORT = 9042; + + private static final String[] CONSISTENCY_LEVELS = { + ConsistencyLevel.ANY.name(), + ConsistencyLevel.ONE.name(), + ConsistencyLevel.TWO.name(), + ConsistencyLevel.THREE.name(), + ConsistencyLevel.QUORUM.name(), + ConsistencyLevel.ALL.name(), + ConsistencyLevel.LOCAL_ONE.name(), + ConsistencyLevel.LOCAL_QUORUM.name(), + ConsistencyLevel.EACH_QUORUM.name(), + ConsistencyLevel.SERIAL.name(), + ConsistencyLevel.LOCAL_SERIAL.name() + }; + + public static final PropertyDescriptor CONTACT_POINTS = new PropertyDescriptor.Builder() + .name("Cassandra Contact Points") + .description(""" + Contact points are addresses of Cassandra nodes. The list of contact points should be + comma-separated and in hostname:port format. Example node1:port,node2:port,.... + The default client port for Cassandra is 9042, but the port(s) must be explicitly specified. + """) + .required(true) + .addValidator(StandardValidators.HOSTNAME_PORT_LIST_VALIDATOR) + .build(); + + public static final PropertyDescriptor KEYSPACE = new PropertyDescriptor.Builder() + .name("Keyspace") + .description(""" + The Cassandra Keyspace to connect to. If no keyspace is specified, the query will need to include + keyspace name before any table reference. For query-native processors, or processors that support + the Table property, provide the table as <KEYSPACE>.<TABLE>. + """) + .required(false) + .addValidator(StandardValidators.NON_EMPTY_VALIDATOR) + .build(); + + public static final PropertyDescriptor PROP_SSL_CONTEXT_SERVICE = new PropertyDescriptor.Builder() + .name("SSL Context Service") + .description(""" + The SSL Context Service used to provide the client certificate information for establishing secure + TLS/SSL connections to the Cassandra cluster. + """) + .required(false) + .identifiesControllerService(SSLContextService.class) + .build(); + + public static final PropertyDescriptor USERNAME = new PropertyDescriptor.Builder() + .name("Username") + .description("Username to access the Cassandra cluster") + .required(false) Review Comment: Should the Username be required? ########## nifi-extension-bundles/nifi-cassandra-bundle/nifi-cassandra-services/src/test/java/org/apache/nifi/service/cassandra/TestCassandraSessionProvider.java: ########## @@ -0,0 +1,108 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.nifi.service.cassandra; + +import org.apache.nifi.cassandra.CassandraClient; +import org.apache.nifi.components.PropertyDescriptor; +import org.apache.nifi.controller.ConfigurationContext; +import org.apache.nifi.reporting.InitializationException; +import org.apache.nifi.util.NoOpProcessor; +import org.apache.nifi.util.TestRunner; +import org.apache.nifi.util.TestRunners; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; + +import java.util.List; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.Mockito.doNothing; +import static org.mockito.Mockito.doReturn; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.spy; +import static org.mockito.Mockito.times; +import static org.mockito.Mockito.verify; + +class TestCassandraSessionProvider { + + private static final String SERVICE_ID = "cassandra-session-provider"; + + private static TestRunner runner; + private static CassandraSessionProvider sessionProvider; + private static CassandraClient mockClient; + + @BeforeEach + void setup() throws InitializationException { + sessionProvider = spy(new CassandraSessionProvider()); + doNothing().when(sessionProvider).connectToCassandra(any(ConfigurationContext.class)); + mockClient = mock(CassandraClient.class); + doReturn(mockClient).when(sessionProvider).getClient(); + + runner = TestRunners.newTestRunner(NoOpProcessor.class); + runner.setValidateExpressionUsage(false); + runner.addControllerService(SERVICE_ID, sessionProvider); + } + + @Test + void testGetPropertyDescriptors() { + List<PropertyDescriptor> properties = sessionProvider.getPropertyDescriptors(); + + assertEquals(10, properties.size(), "Property count mismatch"); + assertTrue(properties.contains(CassandraSessionProvider.CONSISTENCY_LEVEL)); + assertTrue(properties.contains(CassandraSessionProvider.CONTACT_POINTS)); + assertTrue(properties.contains(CassandraSessionProvider.KEYSPACE)); + assertTrue(properties.contains(CassandraSessionProvider.PASSWORD)); + assertTrue(properties.contains(CassandraSessionProvider.PROP_SSL_CONTEXT_SERVICE)); + assertTrue(properties.contains(CassandraSessionProvider.USERNAME)); + assertTrue(properties.contains(CassandraSessionProvider.LOCAL_DATACENTER)); + } + + @Test + void testDefaultsBeforeEnabling() { + assertNotNull(sessionProvider.getClient(), "Client should return mocked client"); + } + + @Test + void testEnableServiceWithValidProperties() throws InitializationException { + + runner.setProperty(sessionProvider, CassandraSessionProvider.CONTACT_POINTS, "localhost:9042"); + runner.setProperty(sessionProvider, CassandraSessionProvider.LOCAL_DATACENTER, "datacenter1"); + runner.setProperty(sessionProvider, CassandraSessionProvider.CONSISTENCY_LEVEL, "QUORUM"); + + runner.enableControllerService(sessionProvider); + runner.assertValid(sessionProvider); + assertNotNull(sessionProvider.getClient(), "Cassandra client should be mocked after enabling service"); + } + + @Test + void testEnableServiceMissingRequiredProperties() { + runner.setProperty(sessionProvider, CassandraSessionProvider.CONTACT_POINTS, "localhost:9042"); + runner.assertNotValid(sessionProvider); + } + + @Test + void testConnectToCassandraIsCalledOnEnable() throws InitializationException { + runner.setProperty(sessionProvider, CassandraSessionProvider.CONTACT_POINTS, "localhost:9042"); + runner.setProperty(sessionProvider, CassandraSessionProvider.LOCAL_DATACENTER, "datacenter1"); + runner.enableControllerService(sessionProvider); + + verify(sessionProvider, times(1)).connectToCassandra(any(ConfigurationContext.class)); + } Review Comment: This test appears to just exercise the spied instance, and should be removed ########## nifi-extension-bundles/nifi-cassandra-bundle/nifi-cassandra-services/pom.xml: ########## @@ -0,0 +1,70 @@ +<?xml version="1.0" encoding="UTF-8"?> +<!-- + Licensed to the Apache Software Foundation (ASF) under one or more + contributor license agreements. See the NOTICE file distributed with + this work for additional information regarding copyright ownership. + The ASF licenses this file to You under the Apache License, Version 2.0 + (the "License"); you may not use this file except in compliance with + the License. You may obtain a copy of the License at + http://www.apache.org/licenses/LICENSE-2.0 + Unless required by applicable law or agreed to in writing, software + distributed under the License is distributed on an "AS IS" BASIS, + WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + See the License for the specific language governing permissions and + limitations under the License. +--> +<project xmlns="http://maven.apache.org/POM/4.0.0"; xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"; xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd";> + + <parent> + <groupId>org.apache.nifi</groupId> + <artifactId>nifi-cassandra-bundle</artifactId> + <version>2.10.0-SNAPSHOT</version> + </parent> + <modelVersion>4.0.0</modelVersion> + + <artifactId>nifi-cassandra-services</artifactId> + <packaging>jar</packaging> + + <dependencies> + <dependency> + <groupId>org.apache.nifi</groupId> + <artifactId>nifi-utils</artifactId> + <version>2.10.0-SNAPSHOT</version> + </dependency> + <dependency> + <groupId>org.apache.nifi</groupId> + <artifactId>nifi-cassandra-services-api</artifactId> + <scope>provided</scope> + <version>2.10.0-SNAPSHOT</version> + </dependency> + <dependency> + <groupId>org.apache.cassandra</groupId> + <artifactId>java-driver-core</artifactId> + </dependency> + <dependency> + <groupId>org.apache.nifi</groupId> + <artifactId>nifi-ssl-context-service-api</artifactId> + <version>2.10.0-SNAPSHOT</version> + </dependency> + <dependency> + <groupId>org.apache.nifi</groupId> + <artifactId>nifi-framework-api</artifactId> + <version>2.10.0-SNAPSHOT</version> + </dependency> Review Comment: This should be removed ########## nifi-extension-bundles/nifi-cassandra-bundle/nifi-cassandra-services/src/main/java/org/apache/nifi/service/cassandra/CassandraSessionProvider.java: ########## @@ -0,0 +1,340 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.nifi.service.cassandra; + +import com.datastax.oss.driver.api.core.ConsistencyLevel; +import com.datastax.oss.driver.api.core.CqlIdentifier; +import com.datastax.oss.driver.api.core.CqlSession; +import com.datastax.oss.driver.api.core.CqlSessionBuilder; +import com.datastax.oss.driver.api.core.config.DefaultDriverOption; +import com.datastax.oss.driver.api.core.config.DriverConfigLoader; +import com.datastax.oss.driver.api.core.config.ProgrammaticDriverConfigLoaderBuilder; +import com.datastax.oss.driver.api.core.metadata.Metadata; +import com.datastax.oss.driver.api.core.metadata.Node; +import org.apache.nifi.annotation.documentation.CapabilityDescription; +import org.apache.nifi.annotation.documentation.Tags; +import org.apache.nifi.annotation.lifecycle.OnDisabled; +import org.apache.nifi.annotation.lifecycle.OnEnabled; +import org.apache.nifi.cassandra.CassandraClient; +import org.apache.nifi.cassandra.CassandraConnectionService; +import org.apache.nifi.cassandra.CompressionType; +import org.apache.nifi.components.PropertyDescriptor; +import org.apache.nifi.components.PropertyValue; +import org.apache.nifi.controller.AbstractControllerService; +import org.apache.nifi.controller.ConfigurationContext; +import org.apache.nifi.logging.ComponentLog; +import org.apache.nifi.processor.exception.ProcessException; +import org.apache.nifi.processor.util.StandardValidators; +import org.apache.nifi.reporting.InitializationException; +import org.apache.nifi.ssl.SSLContextService; + +import java.net.InetSocketAddress; +import java.time.Duration; +import java.util.ArrayList; +import java.util.List; +import java.util.Optional; +import java.util.concurrent.TimeUnit; +import javax.net.ssl.SSLContext; + +@Tags({ "cassandra", "dbcp", "database", "connection", "pooling" }) +@CapabilityDescription("Provides connection session for Cassandra processors to work with Apache Cassandra.") +public class CassandraSessionProvider extends AbstractControllerService implements CassandraConnectionService { + + public static final int DEFAULT_CASSANDRA_PORT = 9042; + + private static final String[] CONSISTENCY_LEVELS = { + ConsistencyLevel.ANY.name(), + ConsistencyLevel.ONE.name(), + ConsistencyLevel.TWO.name(), + ConsistencyLevel.THREE.name(), + ConsistencyLevel.QUORUM.name(), + ConsistencyLevel.ALL.name(), + ConsistencyLevel.LOCAL_ONE.name(), + ConsistencyLevel.LOCAL_QUORUM.name(), + ConsistencyLevel.EACH_QUORUM.name(), + ConsistencyLevel.SERIAL.name(), + ConsistencyLevel.LOCAL_SERIAL.name() + }; + + public static final PropertyDescriptor CONTACT_POINTS = new PropertyDescriptor.Builder() + .name("Cassandra Contact Points") + .description(""" + Contact points are addresses of Cassandra nodes. The list of contact points should be + comma-separated and in hostname:port format. Example node1:port,node2:port,.... + The default client port for Cassandra is 9042, but the port(s) must be explicitly specified. + """) + .required(true) + .addValidator(StandardValidators.HOSTNAME_PORT_LIST_VALIDATOR) + .build(); + + public static final PropertyDescriptor KEYSPACE = new PropertyDescriptor.Builder() + .name("Keyspace") + .description(""" + The Cassandra Keyspace to connect to. If no keyspace is specified, the query will need to include + keyspace name before any table reference. For query-native processors, or processors that support + the Table property, provide the table as <KEYSPACE>.<TABLE>. + """) + .required(false) + .addValidator(StandardValidators.NON_EMPTY_VALIDATOR) + .build(); + + public static final PropertyDescriptor PROP_SSL_CONTEXT_SERVICE = new PropertyDescriptor.Builder() + .name("SSL Context Service") + .description(""" + The SSL Context Service used to provide the client certificate information for establishing secure + TLS/SSL connections to the Cassandra cluster. + """) + .required(false) + .identifiesControllerService(SSLContextService.class) + .build(); + + public static final PropertyDescriptor USERNAME = new PropertyDescriptor.Builder() + .name("Username") + .description("Username to access the Cassandra cluster") + .required(false) + .addValidator(StandardValidators.NON_EMPTY_VALIDATOR) + .build(); + + public static final PropertyDescriptor PASSWORD = new PropertyDescriptor.Builder() + .name("Password") + .description("Password to access the Cassandra cluster") + .required(false) + .sensitive(true) + .addValidator(StandardValidators.NON_EMPTY_VALIDATOR) + .build(); + + public static final PropertyDescriptor CONSISTENCY_LEVEL = new PropertyDescriptor.Builder() + .name("Consistency Level") + .description("The strategy for how many replicas must respond before results are returned.") + .required(true) + .allowableValues(CONSISTENCY_LEVELS) + .defaultValue(ConsistencyLevel.QUORUM.name()) + .build(); + + public static final PropertyDescriptor COMPRESSION_TYPE = new PropertyDescriptor.Builder() + .name("Compression Type") + .description(""" + Specifies the compression type used for transport-level requests and responses.Note that Snappy + compression is not supported when using Protocol V5. + """) + .required(false) + .allowableValues(CompressionType.class) + .defaultValue(CompressionType.NONE.getValue()) + .build(); + + public static final PropertyDescriptor READ_TIMEOUT = new PropertyDescriptor.Builder() + .name("Read Timeout") + .description("Read timeout. 0 means no timeout. If no value is set, the underlying default will be used.") + .required(false) + .addValidator(StandardValidators.TIME_PERIOD_VALIDATOR) + .build(); + + public static final PropertyDescriptor CONNECT_TIMEOUT = new PropertyDescriptor.Builder() + .name("Connect Timeout") + .description("Connection timeout. 0 means no timeout. If no value is set, the underlying default will be used.") + .required(false) + .addValidator(StandardValidators.TIME_PERIOD_VALIDATOR) + .build(); + + public static final PropertyDescriptor LOCAL_DATACENTER = new PropertyDescriptor.Builder() + .name("Local Datacenter") + .description("The local datacenter name for the Cassandra cluster (required by driver 4.x).") Review Comment: The reference to the driver version should be removed ########## nifi-extension-bundles/nifi-cassandra-bundle/nifi-cassandra-services/src/main/java/org/apache/nifi/service/cassandra/CassandraSessionProvider.java: ########## @@ -0,0 +1,340 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.nifi.service.cassandra; + +import com.datastax.oss.driver.api.core.ConsistencyLevel; +import com.datastax.oss.driver.api.core.CqlIdentifier; +import com.datastax.oss.driver.api.core.CqlSession; +import com.datastax.oss.driver.api.core.CqlSessionBuilder; +import com.datastax.oss.driver.api.core.config.DefaultDriverOption; +import com.datastax.oss.driver.api.core.config.DriverConfigLoader; +import com.datastax.oss.driver.api.core.config.ProgrammaticDriverConfigLoaderBuilder; +import com.datastax.oss.driver.api.core.metadata.Metadata; +import com.datastax.oss.driver.api.core.metadata.Node; +import org.apache.nifi.annotation.documentation.CapabilityDescription; +import org.apache.nifi.annotation.documentation.Tags; +import org.apache.nifi.annotation.lifecycle.OnDisabled; +import org.apache.nifi.annotation.lifecycle.OnEnabled; +import org.apache.nifi.cassandra.CassandraClient; +import org.apache.nifi.cassandra.CassandraConnectionService; +import org.apache.nifi.cassandra.CompressionType; +import org.apache.nifi.components.PropertyDescriptor; +import org.apache.nifi.components.PropertyValue; +import org.apache.nifi.controller.AbstractControllerService; +import org.apache.nifi.controller.ConfigurationContext; +import org.apache.nifi.logging.ComponentLog; +import org.apache.nifi.processor.exception.ProcessException; +import org.apache.nifi.processor.util.StandardValidators; +import org.apache.nifi.reporting.InitializationException; +import org.apache.nifi.ssl.SSLContextService; + +import java.net.InetSocketAddress; +import java.time.Duration; +import java.util.ArrayList; +import java.util.List; +import java.util.Optional; +import java.util.concurrent.TimeUnit; +import javax.net.ssl.SSLContext; + +@Tags({ "cassandra", "dbcp", "database", "connection", "pooling" }) +@CapabilityDescription("Provides connection session for Cassandra processors to work with Apache Cassandra.") +public class CassandraSessionProvider extends AbstractControllerService implements CassandraConnectionService { + + public static final int DEFAULT_CASSANDRA_PORT = 9042; + + private static final String[] CONSISTENCY_LEVELS = { + ConsistencyLevel.ANY.name(), + ConsistencyLevel.ONE.name(), + ConsistencyLevel.TWO.name(), + ConsistencyLevel.THREE.name(), + ConsistencyLevel.QUORUM.name(), + ConsistencyLevel.ALL.name(), + ConsistencyLevel.LOCAL_ONE.name(), + ConsistencyLevel.LOCAL_QUORUM.name(), + ConsistencyLevel.EACH_QUORUM.name(), + ConsistencyLevel.SERIAL.name(), + ConsistencyLevel.LOCAL_SERIAL.name() + }; + + public static final PropertyDescriptor CONTACT_POINTS = new PropertyDescriptor.Builder() + .name("Cassandra Contact Points") + .description(""" + Contact points are addresses of Cassandra nodes. The list of contact points should be + comma-separated and in hostname:port format. Example node1:port,node2:port,.... + The default client port for Cassandra is 9042, but the port(s) must be explicitly specified. + """) + .required(true) + .addValidator(StandardValidators.HOSTNAME_PORT_LIST_VALIDATOR) + .build(); + + public static final PropertyDescriptor KEYSPACE = new PropertyDescriptor.Builder() + .name("Keyspace") + .description(""" + The Cassandra Keyspace to connect to. If no keyspace is specified, the query will need to include + keyspace name before any table reference. For query-native processors, or processors that support + the Table property, provide the table as <KEYSPACE>.<TABLE>. + """) + .required(false) + .addValidator(StandardValidators.NON_EMPTY_VALIDATOR) + .build(); + + public static final PropertyDescriptor PROP_SSL_CONTEXT_SERVICE = new PropertyDescriptor.Builder() + .name("SSL Context Service") + .description(""" + The SSL Context Service used to provide the client certificate information for establishing secure + TLS/SSL connections to the Cassandra cluster. + """) + .required(false) + .identifiesControllerService(SSLContextService.class) + .build(); + + public static final PropertyDescriptor USERNAME = new PropertyDescriptor.Builder() + .name("Username") + .description("Username to access the Cassandra cluster") + .required(false) + .addValidator(StandardValidators.NON_EMPTY_VALIDATOR) + .build(); + + public static final PropertyDescriptor PASSWORD = new PropertyDescriptor.Builder() + .name("Password") + .description("Password to access the Cassandra cluster") + .required(false) + .sensitive(true) + .addValidator(StandardValidators.NON_EMPTY_VALIDATOR) + .build(); + + public static final PropertyDescriptor CONSISTENCY_LEVEL = new PropertyDescriptor.Builder() + .name("Consistency Level") + .description("The strategy for how many replicas must respond before results are returned.") + .required(true) + .allowableValues(CONSISTENCY_LEVELS) + .defaultValue(ConsistencyLevel.QUORUM.name()) + .build(); + + public static final PropertyDescriptor COMPRESSION_TYPE = new PropertyDescriptor.Builder() + .name("Compression Type") + .description(""" + Specifies the compression type used for transport-level requests and responses.Note that Snappy + compression is not supported when using Protocol V5. + """) + .required(false) Review Comment: A property with a default value should be required ########## nifi-extension-bundles/nifi-cassandra-bundle/nifi-cassandra-services/src/main/java/org/apache/nifi/service/cassandra/CassandraSessionProvider.java: ########## @@ -0,0 +1,340 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.nifi.service.cassandra; + +import com.datastax.oss.driver.api.core.ConsistencyLevel; +import com.datastax.oss.driver.api.core.CqlIdentifier; +import com.datastax.oss.driver.api.core.CqlSession; +import com.datastax.oss.driver.api.core.CqlSessionBuilder; +import com.datastax.oss.driver.api.core.config.DefaultDriverOption; +import com.datastax.oss.driver.api.core.config.DriverConfigLoader; +import com.datastax.oss.driver.api.core.config.ProgrammaticDriverConfigLoaderBuilder; +import com.datastax.oss.driver.api.core.metadata.Metadata; +import com.datastax.oss.driver.api.core.metadata.Node; +import org.apache.nifi.annotation.documentation.CapabilityDescription; +import org.apache.nifi.annotation.documentation.Tags; +import org.apache.nifi.annotation.lifecycle.OnDisabled; +import org.apache.nifi.annotation.lifecycle.OnEnabled; +import org.apache.nifi.cassandra.CassandraClient; +import org.apache.nifi.cassandra.CassandraConnectionService; +import org.apache.nifi.cassandra.CompressionType; +import org.apache.nifi.components.PropertyDescriptor; +import org.apache.nifi.components.PropertyValue; +import org.apache.nifi.controller.AbstractControllerService; +import org.apache.nifi.controller.ConfigurationContext; +import org.apache.nifi.logging.ComponentLog; +import org.apache.nifi.processor.exception.ProcessException; +import org.apache.nifi.processor.util.StandardValidators; +import org.apache.nifi.reporting.InitializationException; +import org.apache.nifi.ssl.SSLContextService; + +import java.net.InetSocketAddress; +import java.time.Duration; +import java.util.ArrayList; +import java.util.List; +import java.util.Optional; +import java.util.concurrent.TimeUnit; +import javax.net.ssl.SSLContext; + +@Tags({ "cassandra", "dbcp", "database", "connection", "pooling" }) +@CapabilityDescription("Provides connection session for Cassandra processors to work with Apache Cassandra.") +public class CassandraSessionProvider extends AbstractControllerService implements CassandraConnectionService { + + public static final int DEFAULT_CASSANDRA_PORT = 9042; + + private static final String[] CONSISTENCY_LEVELS = { + ConsistencyLevel.ANY.name(), + ConsistencyLevel.ONE.name(), + ConsistencyLevel.TWO.name(), + ConsistencyLevel.THREE.name(), + ConsistencyLevel.QUORUM.name(), + ConsistencyLevel.ALL.name(), + ConsistencyLevel.LOCAL_ONE.name(), + ConsistencyLevel.LOCAL_QUORUM.name(), + ConsistencyLevel.EACH_QUORUM.name(), + ConsistencyLevel.SERIAL.name(), + ConsistencyLevel.LOCAL_SERIAL.name() + }; + + public static final PropertyDescriptor CONTACT_POINTS = new PropertyDescriptor.Builder() + .name("Cassandra Contact Points") + .description(""" + Contact points are addresses of Cassandra nodes. The list of contact points should be + comma-separated and in hostname:port format. Example node1:port,node2:port,.... + The default client port for Cassandra is 9042, but the port(s) must be explicitly specified. + """) + .required(true) + .addValidator(StandardValidators.HOSTNAME_PORT_LIST_VALIDATOR) + .build(); + + public static final PropertyDescriptor KEYSPACE = new PropertyDescriptor.Builder() + .name("Keyspace") + .description(""" + The Cassandra Keyspace to connect to. If no keyspace is specified, the query will need to include + keyspace name before any table reference. For query-native processors, or processors that support + the Table property, provide the table as <KEYSPACE>.<TABLE>. + """) + .required(false) + .addValidator(StandardValidators.NON_EMPTY_VALIDATOR) + .build(); + + public static final PropertyDescriptor PROP_SSL_CONTEXT_SERVICE = new PropertyDescriptor.Builder() + .name("SSL Context Service") + .description(""" + The SSL Context Service used to provide the client certificate information for establishing secure + TLS/SSL connections to the Cassandra cluster. + """) + .required(false) + .identifiesControllerService(SSLContextService.class) + .build(); + + public static final PropertyDescriptor USERNAME = new PropertyDescriptor.Builder() + .name("Username") + .description("Username to access the Cassandra cluster") + .required(false) + .addValidator(StandardValidators.NON_EMPTY_VALIDATOR) + .build(); + + public static final PropertyDescriptor PASSWORD = new PropertyDescriptor.Builder() + .name("Password") + .description("Password to access the Cassandra cluster") + .required(false) + .sensitive(true) + .addValidator(StandardValidators.NON_EMPTY_VALIDATOR) + .build(); + + public static final PropertyDescriptor CONSISTENCY_LEVEL = new PropertyDescriptor.Builder() + .name("Consistency Level") + .description("The strategy for how many replicas must respond before results are returned.") + .required(true) + .allowableValues(CONSISTENCY_LEVELS) + .defaultValue(ConsistencyLevel.QUORUM.name()) + .build(); + + public static final PropertyDescriptor COMPRESSION_TYPE = new PropertyDescriptor.Builder() + .name("Compression Type") + .description(""" + Specifies the compression type used for transport-level requests and responses.Note that Snappy + compression is not supported when using Protocol V5. + """) + .required(false) + .allowableValues(CompressionType.class) + .defaultValue(CompressionType.NONE.getValue()) + .build(); + + public static final PropertyDescriptor READ_TIMEOUT = new PropertyDescriptor.Builder() + .name("Read Timeout") + .description("Read timeout. 0 means no timeout. If no value is set, the underlying default will be used.") + .required(false) + .addValidator(StandardValidators.TIME_PERIOD_VALIDATOR) + .build(); + + public static final PropertyDescriptor CONNECT_TIMEOUT = new PropertyDescriptor.Builder() + .name("Connect Timeout") + .description("Connection timeout. 0 means no timeout. If no value is set, the underlying default will be used.") + .required(false) + .addValidator(StandardValidators.TIME_PERIOD_VALIDATOR) + .build(); + + public static final PropertyDescriptor LOCAL_DATACENTER = new PropertyDescriptor.Builder() + .name("Local Datacenter") + .description("The local datacenter name for the Cassandra cluster (required by driver 4.x).") + .required(true) + .addValidator(StandardValidators.NON_EMPTY_VALIDATOR) + .build(); + + private CqlSession cassandraSession; + private CassandraClient cassandraClient; + private String clusterAddress; + + private static final List<PropertyDescriptor> PROPERTY_DESCRIPTORS = List.of( + CONTACT_POINTS, + CONSISTENCY_LEVEL, + COMPRESSION_TYPE, + KEYSPACE, + USERNAME, + PASSWORD, + PROP_SSL_CONTEXT_SERVICE, + READ_TIMEOUT, + CONNECT_TIMEOUT, + LOCAL_DATACENTER + ); + + @Override + public List<PropertyDescriptor> getSupportedPropertyDescriptors() { + return PROPERTY_DESCRIPTORS; + } + + @OnEnabled + public void onEnabled(final ConfigurationContext context) throws InitializationException { + try { + connectToCassandra(context); + } catch (Exception e) { + throw new InitializationException("Failed to connect to Cassandra cluster. Please check your configuration.", e); Review Comment: ```suggestion throw new InitializationException("Failed to connect to Cassandra cluster", e); ``` ########## nifi-extension-bundles/nifi-cassandra-bundle/nifi-cassandra-services/src/test/java/org/apache/nifi/service/cassandra/TestCassandraSessionProvider.java: ########## @@ -0,0 +1,108 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.nifi.service.cassandra; + +import org.apache.nifi.cassandra.CassandraClient; +import org.apache.nifi.components.PropertyDescriptor; +import org.apache.nifi.controller.ConfigurationContext; +import org.apache.nifi.reporting.InitializationException; +import org.apache.nifi.util.NoOpProcessor; +import org.apache.nifi.util.TestRunner; +import org.apache.nifi.util.TestRunners; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; + +import java.util.List; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.Mockito.doNothing; +import static org.mockito.Mockito.doReturn; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.spy; +import static org.mockito.Mockito.times; +import static org.mockito.Mockito.verify; + +class TestCassandraSessionProvider { Review Comment: ```suggestion class CassandraSessionProviderTest { ``` ########## pom.xml: ########## @@ -499,7 +503,12 @@ <dependency> <groupId>org.xerial.snappy</groupId> <artifactId>snappy-java</artifactId> - <version>1.1.10.8</version> + <version>${snappy-java.version}</version> + </dependency> + <dependency> + <groupId>org.apache.cassandra</groupId> + <artifactId>java-driver-core</artifactId> Review Comment: This should be moved down to the Cassandra bundle ########## nifi-extension-bundles/nifi-cassandra-bundle/nifi-cassandra-services/pom.xml: ########## @@ -0,0 +1,70 @@ +<?xml version="1.0" encoding="UTF-8"?> +<!-- + Licensed to the Apache Software Foundation (ASF) under one or more + contributor license agreements. See the NOTICE file distributed with + this work for additional information regarding copyright ownership. + The ASF licenses this file to You under the Apache License, Version 2.0 + (the "License"); you may not use this file except in compliance with + the License. You may obtain a copy of the License at + http://www.apache.org/licenses/LICENSE-2.0 + Unless required by applicable law or agreed to in writing, software + distributed under the License is distributed on an "AS IS" BASIS, + WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + See the License for the specific language governing permissions and + limitations under the License. +--> +<project xmlns="http://maven.apache.org/POM/4.0.0"; xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"; xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd";> + + <parent> + <groupId>org.apache.nifi</groupId> + <artifactId>nifi-cassandra-bundle</artifactId> + <version>2.10.0-SNAPSHOT</version> + </parent> + <modelVersion>4.0.0</modelVersion> + + <artifactId>nifi-cassandra-services</artifactId> + <packaging>jar</packaging> + + <dependencies> + <dependency> + <groupId>org.apache.nifi</groupId> + <artifactId>nifi-utils</artifactId> + <version>2.10.0-SNAPSHOT</version> + </dependency> + <dependency> + <groupId>org.apache.nifi</groupId> + <artifactId>nifi-cassandra-services-api</artifactId> + <scope>provided</scope> + <version>2.10.0-SNAPSHOT</version> + </dependency> + <dependency> + <groupId>org.apache.cassandra</groupId> + <artifactId>java-driver-core</artifactId> + </dependency> + <dependency> + <groupId>org.apache.nifi</groupId> + <artifactId>nifi-ssl-context-service-api</artifactId> Review Comment: This should be provided ########## nifi-extension-bundles/nifi-cassandra-bundle/nifi-cassandra-services/src/main/java/org/apache/nifi/service/cassandra/CassandraSessionProvider.java: ########## @@ -0,0 +1,340 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.nifi.service.cassandra; + +import com.datastax.oss.driver.api.core.ConsistencyLevel; +import com.datastax.oss.driver.api.core.CqlIdentifier; +import com.datastax.oss.driver.api.core.CqlSession; +import com.datastax.oss.driver.api.core.CqlSessionBuilder; +import com.datastax.oss.driver.api.core.config.DefaultDriverOption; +import com.datastax.oss.driver.api.core.config.DriverConfigLoader; +import com.datastax.oss.driver.api.core.config.ProgrammaticDriverConfigLoaderBuilder; +import com.datastax.oss.driver.api.core.metadata.Metadata; +import com.datastax.oss.driver.api.core.metadata.Node; +import org.apache.nifi.annotation.documentation.CapabilityDescription; +import org.apache.nifi.annotation.documentation.Tags; +import org.apache.nifi.annotation.lifecycle.OnDisabled; +import org.apache.nifi.annotation.lifecycle.OnEnabled; +import org.apache.nifi.cassandra.CassandraClient; +import org.apache.nifi.cassandra.CassandraConnectionService; +import org.apache.nifi.cassandra.CompressionType; +import org.apache.nifi.components.PropertyDescriptor; +import org.apache.nifi.components.PropertyValue; +import org.apache.nifi.controller.AbstractControllerService; +import org.apache.nifi.controller.ConfigurationContext; +import org.apache.nifi.logging.ComponentLog; +import org.apache.nifi.processor.exception.ProcessException; +import org.apache.nifi.processor.util.StandardValidators; +import org.apache.nifi.reporting.InitializationException; +import org.apache.nifi.ssl.SSLContextService; + +import java.net.InetSocketAddress; +import java.time.Duration; +import java.util.ArrayList; +import java.util.List; +import java.util.Optional; +import java.util.concurrent.TimeUnit; +import javax.net.ssl.SSLContext; + +@Tags({ "cassandra", "dbcp", "database", "connection", "pooling" }) +@CapabilityDescription("Provides connection session for Cassandra processors to work with Apache Cassandra.") +public class CassandraSessionProvider extends AbstractControllerService implements CassandraConnectionService { + + public static final int DEFAULT_CASSANDRA_PORT = 9042; + + private static final String[] CONSISTENCY_LEVELS = { + ConsistencyLevel.ANY.name(), + ConsistencyLevel.ONE.name(), + ConsistencyLevel.TWO.name(), + ConsistencyLevel.THREE.name(), + ConsistencyLevel.QUORUM.name(), + ConsistencyLevel.ALL.name(), + ConsistencyLevel.LOCAL_ONE.name(), + ConsistencyLevel.LOCAL_QUORUM.name(), + ConsistencyLevel.EACH_QUORUM.name(), + ConsistencyLevel.SERIAL.name(), + ConsistencyLevel.LOCAL_SERIAL.name() + }; + + public static final PropertyDescriptor CONTACT_POINTS = new PropertyDescriptor.Builder() + .name("Cassandra Contact Points") + .description(""" + Contact points are addresses of Cassandra nodes. The list of contact points should be + comma-separated and in hostname:port format. Example node1:port,node2:port,.... + The default client port for Cassandra is 9042, but the port(s) must be explicitly specified. + """) + .required(true) + .addValidator(StandardValidators.HOSTNAME_PORT_LIST_VALIDATOR) + .build(); + + public static final PropertyDescriptor KEYSPACE = new PropertyDescriptor.Builder() + .name("Keyspace") + .description(""" + The Cassandra Keyspace to connect to. If no keyspace is specified, the query will need to include + keyspace name before any table reference. For query-native processors, or processors that support + the Table property, provide the table as <KEYSPACE>.<TABLE>. + """) + .required(false) + .addValidator(StandardValidators.NON_EMPTY_VALIDATOR) + .build(); + + public static final PropertyDescriptor PROP_SSL_CONTEXT_SERVICE = new PropertyDescriptor.Builder() + .name("SSL Context Service") + .description(""" + The SSL Context Service used to provide the client certificate information for establishing secure + TLS/SSL connections to the Cassandra cluster. + """) + .required(false) + .identifiesControllerService(SSLContextService.class) + .build(); + + public static final PropertyDescriptor USERNAME = new PropertyDescriptor.Builder() + .name("Username") + .description("Username to access the Cassandra cluster") + .required(false) + .addValidator(StandardValidators.NON_EMPTY_VALIDATOR) + .build(); + + public static final PropertyDescriptor PASSWORD = new PropertyDescriptor.Builder() + .name("Password") + .description("Password to access the Cassandra cluster") + .required(false) + .sensitive(true) + .addValidator(StandardValidators.NON_EMPTY_VALIDATOR) + .build(); + + public static final PropertyDescriptor CONSISTENCY_LEVEL = new PropertyDescriptor.Builder() + .name("Consistency Level") + .description("The strategy for how many replicas must respond before results are returned.") + .required(true) + .allowableValues(CONSISTENCY_LEVELS) + .defaultValue(ConsistencyLevel.QUORUM.name()) + .build(); + + public static final PropertyDescriptor COMPRESSION_TYPE = new PropertyDescriptor.Builder() + .name("Compression Type") + .description(""" + Specifies the compression type used for transport-level requests and responses.Note that Snappy + compression is not supported when using Protocol V5. + """) + .required(false) + .allowableValues(CompressionType.class) + .defaultValue(CompressionType.NONE.getValue()) + .build(); + + public static final PropertyDescriptor READ_TIMEOUT = new PropertyDescriptor.Builder() + .name("Read Timeout") + .description("Read timeout. 0 means no timeout. If no value is set, the underlying default will be used.") + .required(false) + .addValidator(StandardValidators.TIME_PERIOD_VALIDATOR) + .build(); + + public static final PropertyDescriptor CONNECT_TIMEOUT = new PropertyDescriptor.Builder() + .name("Connect Timeout") + .description("Connection timeout. 0 means no timeout. If no value is set, the underlying default will be used.") + .required(false) + .addValidator(StandardValidators.TIME_PERIOD_VALIDATOR) + .build(); + + public static final PropertyDescriptor LOCAL_DATACENTER = new PropertyDescriptor.Builder() + .name("Local Datacenter") + .description("The local datacenter name for the Cassandra cluster (required by driver 4.x).") + .required(true) + .addValidator(StandardValidators.NON_EMPTY_VALIDATOR) + .build(); + + private CqlSession cassandraSession; + private CassandraClient cassandraClient; + private String clusterAddress; + + private static final List<PropertyDescriptor> PROPERTY_DESCRIPTORS = List.of( + CONTACT_POINTS, + CONSISTENCY_LEVEL, + COMPRESSION_TYPE, + KEYSPACE, + USERNAME, + PASSWORD, + PROP_SSL_CONTEXT_SERVICE, + READ_TIMEOUT, + CONNECT_TIMEOUT, + LOCAL_DATACENTER + ); + + @Override + public List<PropertyDescriptor> getSupportedPropertyDescriptors() { + return PROPERTY_DESCRIPTORS; + } + + @OnEnabled + public void onEnabled(final ConfigurationContext context) throws InitializationException { + try { + connectToCassandra(context); + } catch (Exception e) { + throw new InitializationException("Failed to connect to Cassandra cluster. Please check your configuration.", e); + } + } + + @OnDisabled + public void onDisabled() { + if (cassandraSession != null) { + cassandraSession.close(); + cassandraSession = null; + } + cassandraClient = null; + } + + @Override + public CassandraClient getClient() { + if (cassandraClient == null) { + throw new ProcessException(""" + Cassandra Client is not initialized. The Controller Service failed to connect to the cluster. + """); + } Review Comment: This check and exception seem unnecessary ########## nifi-extension-bundles/nifi-cassandra-bundle/nifi-cassandra-services/pom.xml: ########## @@ -0,0 +1,70 @@ +<?xml version="1.0" encoding="UTF-8"?> +<!-- + Licensed to the Apache Software Foundation (ASF) under one or more + contributor license agreements. See the NOTICE file distributed with + this work for additional information regarding copyright ownership. + The ASF licenses this file to You under the Apache License, Version 2.0 + (the "License"); you may not use this file except in compliance with + the License. You may obtain a copy of the License at + http://www.apache.org/licenses/LICENSE-2.0 + Unless required by applicable law or agreed to in writing, software + distributed under the License is distributed on an "AS IS" BASIS, + WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + See the License for the specific language governing permissions and + limitations under the License. +--> +<project xmlns="http://maven.apache.org/POM/4.0.0"; xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"; xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd";> + + <parent> + <groupId>org.apache.nifi</groupId> + <artifactId>nifi-cassandra-bundle</artifactId> + <version>2.10.0-SNAPSHOT</version> + </parent> + <modelVersion>4.0.0</modelVersion> + + <artifactId>nifi-cassandra-services</artifactId> + <packaging>jar</packaging> + + <dependencies> + <dependency> + <groupId>org.apache.nifi</groupId> + <artifactId>nifi-utils</artifactId> + <version>2.10.0-SNAPSHOT</version> + </dependency> + <dependency> + <groupId>org.apache.nifi</groupId> + <artifactId>nifi-cassandra-services-api</artifactId> + <scope>provided</scope> + <version>2.10.0-SNAPSHOT</version> + </dependency> + <dependency> + <groupId>org.apache.cassandra</groupId> + <artifactId>java-driver-core</artifactId> + </dependency> + <dependency> + <groupId>org.apache.nifi</groupId> + <artifactId>nifi-ssl-context-service-api</artifactId> + <version>2.10.0-SNAPSHOT</version> + </dependency> + <dependency> + <groupId>org.apache.nifi</groupId> + <artifactId>nifi-framework-api</artifactId> + <version>2.10.0-SNAPSHOT</version> + </dependency> + <dependency> + <groupId>org.apache.nifi</groupId> + <artifactId>nifi-mock</artifactId> + <version>2.10.0-SNAPSHOT</version> + <scope>test</scope> + </dependency> + <dependency> + <groupId>at.yawk.lz4</groupId> + <artifactId>lz4-java</artifactId> + <version>1.10.4</version> + </dependency> Review Comment: This should be inherited from the parent pom.xml -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
