[GitHub] [nifi] sjyang18 commented on a change in pull request #4253: NIFI-7406: PutAzureCosmosRecord Processor
sjyang18 commented on a change in pull request #4253: URL: https://github.com/apache/nifi/pull/4253#discussion_r503423692 ## File path: nifi-nar-bundles/nifi-azure-bundle/pom.xml ## @@ -29,6 +29,8 @@ 8.4.0 1.5.0 2.10.3 +4.2.0 +1.6.0 Review comment: com.azure:azure-identity (https://mvnrepository.com/artifact/com.azure/azure-identity/1.0.6) has also dependency on com.azure:azure-core, which is 1.5.0 and is upgradable to 1.9.0. Thus, it is also not breaking with 1.6.0 azure-core version. 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. For queries about this service, please contact Infrastructure at: us...@infra.apache.org
[GitHub] [nifi] sjyang18 commented on a change in pull request #4253: NIFI-7406: PutAzureCosmosRecord Processor
sjyang18 commented on a change in pull request #4253:
URL: https://github.com/apache/nifi/pull/4253#discussion_r502721645
##
File path:
nifi-nar-bundles/nifi-azure-bundle/nifi-azure-processors/src/test/java/org/apache/nifi/processors/azure/cosmos/document/MockTestBase.java
##
@@ -0,0 +1,74 @@
+/*
+ * 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.processors.azure.cosmos.document;
+
+import static org.mockito.Mockito.mock;
+
+import java.util.Random;
+
+import com.azure.cosmos.ConsistencyLevel;
+import com.azure.cosmos.CosmosClient;
+
+import org.apache.nifi.reporting.InitializationException;
+import
org.apache.nifi.services.azure.cosmos.document.AzureCosmosDBConnectionControllerService;
+import org.apache.nifi.util.TestRunner;
+
+public class MockTestBase {
+
+protected static final String MOCK_DB_NAME = "MOCK_DB_NAME";
+protected static final String MOCK_CONTAINER_ID = "MOCK_CONTAINER_ID";
+protected static final String MOCK_URI = "MOCK_URI";
+protected static final String MOCK_DB_ACCESS_KEY = "MOCK_DB_ACCESS_KEY";
+public static final String MOCK_QUERY = "select * from c";
+
+public static final String MOCK_PARTITION_FIELD_NAME = "category";
+protected TestRunner testRunner;
+
+protected void setBasicMockProperties(boolean withConnectionService)
throws InitializationException {
+if(testRunner != null) {
+
testRunner.setProperty(AbstractAzureCosmosDBProcessor.DATABASE_NAME,
MOCK_DB_NAME);
+
testRunner.setProperty(AbstractAzureCosmosDBProcessor.CONTAINER_ID,MOCK_CONTAINER_ID);
+
testRunner.setProperty(AbstractAzureCosmosDBProcessor.PARTITION_KEY,MOCK_PARTITION_FIELD_NAME);
+if(withConnectionService) {
+// setup connnection controller service
+AzureCosmosDBConnectionControllerService service = new
MockConnectionService();
+testRunner.addControllerService("connService", service);
+testRunner.setProperty(service, AzureCosmosDBUtils.URI,
MOCK_URI);
+testRunner.setProperty(service,
AzureCosmosDBUtils.DB_ACCESS_KEY, MOCK_DB_ACCESS_KEY);
+
+// now, after enabling and setting the service, it should be
valid
+testRunner.enableControllerService(service);
+
testRunner.setProperty(AbstractAzureCosmosDBProcessor.CONNECTION_SERVICE,
"connService");
Review comment:
assertValid() are called in the actual test cases. This is not the test
case, but the common method used in the actual test cases.
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.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
[GitHub] [nifi] sjyang18 commented on a change in pull request #4253: NIFI-7406: PutAzureCosmosRecord Processor
sjyang18 commented on a change in pull request #4253: URL: https://github.com/apache/nifi/pull/4253#discussion_r502678996 ## File path: nifi-nar-bundles/nifi-azure-bundle/pom.xml ## @@ -29,6 +29,8 @@ 8.4.0 1.5.0 2.10.3 +4.2.0 +1.6.0 Review comment: If we stay with azure core 1.5.0, this will not work with Cosmos 4.2.0, which has the better performance than Cosmos 4.1.0. I tested out integration tests with azure core version 1.6.0, and it shows the same test result as 1.5.0. No test failures with azure-datalake processors. 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. For queries about this service, please contact Infrastructure at: us...@infra.apache.org
[GitHub] [nifi] sjyang18 commented on a change in pull request #4253: NIFI-7406: PutAzureCosmosRecord Processor
sjyang18 commented on a change in pull request #4253:
URL: https://github.com/apache/nifi/pull/4253#discussion_r421705645
##
File path:
nifi-nar-bundles/nifi-azure-bundle/nifi-azure-processors/src/main/java/org/apache/nifi/processors/azure/cosmos/document/AbstractAzureCosmosDBProcessor.java
##
@@ -0,0 +1,337 @@
+/*
+ * 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.processors.azure.cosmos.document;
+
+import java.io.ByteArrayInputStream;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+
+import com.azure.cosmos.ConnectionPolicy;
+import com.azure.cosmos.ConsistencyLevel;
+import com.azure.cosmos.CosmosClient;
+import com.azure.cosmos.CosmosClientBuilder;
+import com.azure.cosmos.CosmosClientException;
+import com.azure.cosmos.CosmosContainer;
+import com.azure.cosmos.CosmosDatabase;
+import com.azure.cosmos.models.CosmosContainerProperties;
+
+import org.apache.nifi.annotation.lifecycle.OnScheduled;
+import org.apache.nifi.annotation.lifecycle.OnStopped;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.components.ValidationContext;
+import org.apache.nifi.components.ValidationResult;
+import org.apache.nifi.logging.ComponentLog;
+
+import org.apache.nifi.flowfile.FlowFile;
+import org.apache.nifi.processor.AbstractProcessor;
+import org.apache.nifi.processor.ProcessContext;
+import org.apache.nifi.processor.ProcessSession;
+import org.apache.nifi.processor.Relationship;
+import org.apache.nifi.processor.util.StandardValidators;
+import org.apache.nifi.services.azure.cosmos.AzureCosmosDBConnectionService;
+
+public abstract class AbstractAzureCosmosDBProcessor extends AbstractProcessor
{
+static final String CONSISTENCY_STRONG = "STRONG";
+static final String CONSISTENCY_BOUNDED_STALENESS= "BOUNDED_STALENESS";
+static final String CONSISTENCY_SESSION = "SESSION";
+static final String CONSISTENCY_CONSISTENT_PREFIX = "CONSISTENT_PREFIX";
+static final String CONSISTENCY_EVENTUAL = "EVENTUAL";
+
+static final Relationship REL_SUCCESS = new
Relationship.Builder().name("success")
+.description("All FlowFiles that are written to Azure Cosmos DB are routed
to this relationship").build();
+
+static final Relationship REL_FAILURE = new
Relationship.Builder().name("failure")
+.description("All FlowFiles that cannot be written to Azure Cosmos DB are
routed to this relationship").build();
+
+static final Relationship REL_ORIGINAL = new Relationship.Builder()
+.name("original")
+.description("All input FlowFiles that are part of a successful
query execution go here.")
+.build();
+
+static final PropertyDescriptor CONNECTION_SERVICE = new
PropertyDescriptor.Builder()
+.name("azure-cosmos-db-connection-service")
+.displayName("Azure Cosmos DB Connection Service")
+.description("If configured, this property will use the assigned for
retrieving connection string info.")
+.required(false)
+.identifiesControllerService(AzureCosmosDBConnectionService.class)
+.build();
+
+
+static final PropertyDescriptor URI = new PropertyDescriptor.Builder()
+.name("azure-cosmos-db-uri")
+.displayName("Azure Cosmos DB URI")
+.description("Azure Cosmos DB URI, typically of the form:
https://{databaseaccount}.documents.azure.com:443/. Note this host URL is for
Azure Cosmos DB with CORE SQL API")
+.required(false)
+.addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+.build();
+
+static final PropertyDescriptor DB_ACCESS_KEY = new
PropertyDescriptor.Builder()
+.name("azure-cosmos-db-key")
+.displayName("Azure Cosmos DB Access Key")
+.description("Azure Cosmos DB Access Key from Azure Portal
(Settings->Keys)")
Review comment:
I tried to give the general statement so that user may figure out if
there will be changes in Azure portal.
This is an automated message from the Apache Git Service.
To respond to the message, please log on to
[GitHub] [nifi] sjyang18 commented on a change in pull request #4253: NIFI-7406: PutAzureCosmosRecord Processor
sjyang18 commented on a change in pull request #4253:
URL: https://github.com/apache/nifi/pull/4253#discussion_r421704182
##
File path:
nifi-nar-bundles/nifi-azure-bundle/nifi-azure-processors/src/main/java/org/apache/nifi/processors/azure/cosmos/document/AbstractAzureCosmosDBProcessor.java
##
@@ -0,0 +1,337 @@
+/*
+ * 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.processors.azure.cosmos.document;
+
+import java.io.ByteArrayInputStream;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+
+import com.azure.cosmos.ConnectionPolicy;
+import com.azure.cosmos.ConsistencyLevel;
+import com.azure.cosmos.CosmosClient;
+import com.azure.cosmos.CosmosClientBuilder;
+import com.azure.cosmos.CosmosClientException;
+import com.azure.cosmos.CosmosContainer;
+import com.azure.cosmos.CosmosDatabase;
+import com.azure.cosmos.models.CosmosContainerProperties;
+
+import org.apache.nifi.annotation.lifecycle.OnScheduled;
+import org.apache.nifi.annotation.lifecycle.OnStopped;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.components.ValidationContext;
+import org.apache.nifi.components.ValidationResult;
+import org.apache.nifi.logging.ComponentLog;
+
+import org.apache.nifi.flowfile.FlowFile;
+import org.apache.nifi.processor.AbstractProcessor;
+import org.apache.nifi.processor.ProcessContext;
+import org.apache.nifi.processor.ProcessSession;
+import org.apache.nifi.processor.Relationship;
+import org.apache.nifi.processor.util.StandardValidators;
+import org.apache.nifi.services.azure.cosmos.AzureCosmosDBConnectionService;
+
+public abstract class AbstractAzureCosmosDBProcessor extends AbstractProcessor
{
+static final String CONSISTENCY_STRONG = "STRONG";
+static final String CONSISTENCY_BOUNDED_STALENESS= "BOUNDED_STALENESS";
+static final String CONSISTENCY_SESSION = "SESSION";
+static final String CONSISTENCY_CONSISTENT_PREFIX = "CONSISTENT_PREFIX";
+static final String CONSISTENCY_EVENTUAL = "EVENTUAL";
+
+static final Relationship REL_SUCCESS = new
Relationship.Builder().name("success")
+.description("All FlowFiles that are written to Azure Cosmos DB are routed
to this relationship").build();
+
+static final Relationship REL_FAILURE = new
Relationship.Builder().name("failure")
+.description("All FlowFiles that cannot be written to Azure Cosmos DB are
routed to this relationship").build();
+
+static final Relationship REL_ORIGINAL = new Relationship.Builder()
+.name("original")
+.description("All input FlowFiles that are part of a successful
query execution go here.")
+.build();
+
+static final PropertyDescriptor CONNECTION_SERVICE = new
PropertyDescriptor.Builder()
+.name("azure-cosmos-db-connection-service")
+.displayName("Azure Cosmos DB Connection Service")
+.description("If configured, this property will use the assigned for
retrieving connection string info.")
+.required(false)
+.identifiesControllerService(AzureCosmosDBConnectionService.class)
+.build();
+
+
+static final PropertyDescriptor URI = new PropertyDescriptor.Builder()
+.name("azure-cosmos-db-uri")
+.displayName("Azure Cosmos DB URI")
+.description("Azure Cosmos DB URI, typically of the form:
https://{databaseaccount}.documents.azure.com:443/. Note this host URL is for
Azure Cosmos DB with CORE SQL API")
+.required(false)
+.addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+.build();
+
+static final PropertyDescriptor DB_ACCESS_KEY = new
PropertyDescriptor.Builder()
+.name("azure-cosmos-db-key")
+.displayName("Azure Cosmos DB Access Key")
+.description("Azure Cosmos DB Access Key from Azure Portal
(Settings->Keys)")
+.required(false)
+.addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+.sensitive(true)
+.build();
+
+static final PropertyDescriptor DATABASE_NAME = new
PropertyDescriptor.Builder()
+.name("azure-cosmos-db-name")
+.displayName("Azure Cosmos DB
[GitHub] [nifi] sjyang18 commented on a change in pull request #4253: NIFI-7406: PutAzureCosmosRecord Processor
sjyang18 commented on a change in pull request #4253:
URL: https://github.com/apache/nifi/pull/4253#discussion_r421704036
##
File path:
nifi-nar-bundles/nifi-azure-bundle/nifi-azure-processors/src/main/java/org/apache/nifi/processors/azure/cosmos/document/AbstractAzureCosmosDBProcessor.java
##
@@ -0,0 +1,337 @@
+/*
+ * 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.processors.azure.cosmos.document;
+
+import java.io.ByteArrayInputStream;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+
+import com.azure.cosmos.ConnectionPolicy;
+import com.azure.cosmos.ConsistencyLevel;
+import com.azure.cosmos.CosmosClient;
+import com.azure.cosmos.CosmosClientBuilder;
+import com.azure.cosmos.CosmosClientException;
+import com.azure.cosmos.CosmosContainer;
+import com.azure.cosmos.CosmosDatabase;
+import com.azure.cosmos.models.CosmosContainerProperties;
+
+import org.apache.nifi.annotation.lifecycle.OnScheduled;
+import org.apache.nifi.annotation.lifecycle.OnStopped;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.components.ValidationContext;
+import org.apache.nifi.components.ValidationResult;
+import org.apache.nifi.logging.ComponentLog;
+
+import org.apache.nifi.flowfile.FlowFile;
+import org.apache.nifi.processor.AbstractProcessor;
+import org.apache.nifi.processor.ProcessContext;
+import org.apache.nifi.processor.ProcessSession;
+import org.apache.nifi.processor.Relationship;
+import org.apache.nifi.processor.util.StandardValidators;
+import org.apache.nifi.services.azure.cosmos.AzureCosmosDBConnectionService;
+
+public abstract class AbstractAzureCosmosDBProcessor extends AbstractProcessor
{
+static final String CONSISTENCY_STRONG = "STRONG";
+static final String CONSISTENCY_BOUNDED_STALENESS= "BOUNDED_STALENESS";
+static final String CONSISTENCY_SESSION = "SESSION";
+static final String CONSISTENCY_CONSISTENT_PREFIX = "CONSISTENT_PREFIX";
+static final String CONSISTENCY_EVENTUAL = "EVENTUAL";
+
+static final Relationship REL_SUCCESS = new
Relationship.Builder().name("success")
+.description("All FlowFiles that are written to Azure Cosmos DB are routed
to this relationship").build();
+
+static final Relationship REL_FAILURE = new
Relationship.Builder().name("failure")
+.description("All FlowFiles that cannot be written to Azure Cosmos DB are
routed to this relationship").build();
+
+static final Relationship REL_ORIGINAL = new Relationship.Builder()
+.name("original")
+.description("All input FlowFiles that are part of a successful
query execution go here.")
+.build();
+
+static final PropertyDescriptor CONNECTION_SERVICE = new
PropertyDescriptor.Builder()
+.name("azure-cosmos-db-connection-service")
+.displayName("Azure Cosmos DB Connection Service")
+.description("If configured, this property will use the assigned for
retrieving connection string info.")
+.required(false)
+.identifiesControllerService(AzureCosmosDBConnectionService.class)
+.build();
+
+
+static final PropertyDescriptor URI = new PropertyDescriptor.Builder()
+.name("azure-cosmos-db-uri")
+.displayName("Azure Cosmos DB URI")
+.description("Azure Cosmos DB URI, typically of the form:
https://{databaseaccount}.documents.azure.com:443/. Note this host URL is for
Azure Cosmos DB with CORE SQL API")
+.required(false)
+.addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+.build();
+
+static final PropertyDescriptor DB_ACCESS_KEY = new
PropertyDescriptor.Builder()
+.name("azure-cosmos-db-key")
+.displayName("Azure Cosmos DB Access Key")
+.description("Azure Cosmos DB Access Key from Azure Portal
(Settings->Keys)")
+.required(false)
+.addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+.sensitive(true)
+.build();
+
+static final PropertyDescriptor DATABASE_NAME = new
PropertyDescriptor.Builder()
+.name("azure-cosmos-db-name")
+.displayName("Azure Cosmos DB
[GitHub] [nifi] sjyang18 commented on a change in pull request #4253: NIFI-7406: PutAzureCosmosRecord Processor
sjyang18 commented on a change in pull request #4253:
URL: https://github.com/apache/nifi/pull/4253#discussion_r421165541
##
File path:
nifi-nar-bundles/nifi-azure-bundle/nifi-azure-processors/src/test/java/org/apache/nifi/processors/azure/cosmos/document/ITAbstractAzureCosmosDBDocument.java
##
@@ -0,0 +1,198 @@
+/*
+ * 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.processors.azure.cosmos.document;
+
+import com.azure.cosmos.ConnectionPolicy;
+import com.azure.cosmos.CosmosClient;
+import com.azure.cosmos.CosmosClientBuilder;
+import com.azure.cosmos.CosmosClientException;
+import com.azure.cosmos.CosmosContainer;
+import com.azure.cosmos.CosmosDatabase;
+
+import com.azure.cosmos.models.CosmosContainerProperties;
+import com.azure.cosmos.models.CosmosItemRequestOptions;
+import com.azure.cosmos.models.FeedOptions;
+import com.azure.cosmos.models.PartitionKey;
+import com.azure.cosmos.util.CosmosPagedIterable;
+import com.fasterxml.jackson.databind.JsonNode;
+
+import org.apache.nifi.processor.Processor;
+import
org.apache.nifi.services.azure.cosmos.document.AzureCosmosDBConnectionControllerService;
+import org.apache.nifi.util.TestRunner;
+import org.apache.nifi.util.TestRunners;
+import org.apache.nifi.util.file.FileUtils;
+import org.junit.AfterClass;
+import org.junit.Before;
+import org.junit.BeforeClass;
+
+import java.io.FileInputStream;
+import java.io.FileNotFoundException;
+import java.io.IOException;
+import java.util.Properties;
+import java.util.logging.Logger;
+
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.fail;
+
+public abstract class ITAbstractAzureCosmosDBDocument {
+static Logger logger =
Logger.getLogger(ITAbstractAzureCosmosDBDocument.class.getName());
+
+private static final Properties CONFIG;
+
+private static final String CREDENTIALS_FILE =
System.getProperty("user.home") + "/azure-credentials.PROPERTIES";
+protected static final String TEST_COSMOS_DB_NAME = "nifi-test-db";
+protected static final String TEST_COSMOS_CONTAINER_NAME =
"nifi-test-container";
+protected static final String TEST_COSMOS_PARTITION_KEY_FIELD_NAME =
"category";
+protected static CosmosClient client;
+protected static CosmosDatabase cdb;
+protected static CosmosContainer container;
+
+static {
+final FileInputStream fis;
+CONFIG = new Properties();
+try {
+fis = new FileInputStream(CREDENTIALS_FILE);
+try {
+CONFIG.load(fis);
+} catch (IOException e) {
+fail("Could not open credentials file " + CREDENTIALS_FILE +
": " + e.getLocalizedMessage());
+} finally {
+FileUtils.closeQuietly(fis);
+}
+} catch (FileNotFoundException e) {
+fail("Could not open credentials file " + CREDENTIALS_FILE + ": "
+ e.getLocalizedMessage());
+}
+}
+
+protected static String getComosURI() {
+return CONFIG.getProperty("cosmosURI");
+}
+
+protected static String getCosmosKey() {
+return CONFIG.getProperty("cosmosKey");
+}
+
+protected TestRunner runner;
+
+@BeforeClass
+public static void createTestDBContainerIfNeeded() throws
CosmosClientException {
+final String testDBURI = getComosURI();
+final String testDBContainer = getCosmosKey();
+client = new CosmosClientBuilder()
+.endpoint(testDBURI)
+.key(testDBContainer)
+.connectionPolicy(ConnectionPolicy.getDefaultPolicy())
+.buildClient();
+cdb =
client.createDatabaseIfNotExists(TEST_COSMOS_DB_NAME).getDatabase();
+CosmosContainerProperties containerProperties =
+new CosmosContainerProperties(TEST_COSMOS_CONTAINER_NAME,
"/"+TEST_COSMOS_PARTITION_KEY_FIELD_NAME);
+container = cdb.createContainerIfNotExists(containerProperties,
400).getContainer();
+assertNotNull(container);
+}
+
+@AfterClass
+public static void dropTestDBAndContainer() throws CosmosClientException {
+resetTestCosmosConnection();
+if(container != null) {
+
[GitHub] [nifi] sjyang18 commented on a change in pull request #4253: NIFI-7406: PutAzureCosmosRecord Processor
sjyang18 commented on a change in pull request #4253:
URL: https://github.com/apache/nifi/pull/4253#discussion_r421165461
##
File path:
nifi-nar-bundles/nifi-azure-bundle/nifi-azure-processors/src/test/java/org/apache/nifi/processors/azure/cosmos/document/ITAbstractAzureCosmosDBDocument.java
##
@@ -0,0 +1,198 @@
+/*
+ * 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.processors.azure.cosmos.document;
+
+import com.azure.cosmos.ConnectionPolicy;
+import com.azure.cosmos.CosmosClient;
+import com.azure.cosmos.CosmosClientBuilder;
+import com.azure.cosmos.CosmosClientException;
+import com.azure.cosmos.CosmosContainer;
+import com.azure.cosmos.CosmosDatabase;
+
+import com.azure.cosmos.models.CosmosContainerProperties;
+import com.azure.cosmos.models.CosmosItemRequestOptions;
+import com.azure.cosmos.models.FeedOptions;
+import com.azure.cosmos.models.PartitionKey;
+import com.azure.cosmos.util.CosmosPagedIterable;
+import com.fasterxml.jackson.databind.JsonNode;
+
+import org.apache.nifi.processor.Processor;
+import
org.apache.nifi.services.azure.cosmos.document.AzureCosmosDBConnectionControllerService;
+import org.apache.nifi.util.TestRunner;
+import org.apache.nifi.util.TestRunners;
+import org.apache.nifi.util.file.FileUtils;
+import org.junit.AfterClass;
+import org.junit.Before;
+import org.junit.BeforeClass;
+
+import java.io.FileInputStream;
+import java.io.FileNotFoundException;
+import java.io.IOException;
+import java.util.Properties;
+import java.util.logging.Logger;
+
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.fail;
+
+public abstract class ITAbstractAzureCosmosDBDocument {
+static Logger logger =
Logger.getLogger(ITAbstractAzureCosmosDBDocument.class.getName());
+
+private static final Properties CONFIG;
+
+private static final String CREDENTIALS_FILE =
System.getProperty("user.home") + "/azure-credentials.PROPERTIES";
+protected static final String TEST_COSMOS_DB_NAME = "nifi-test-db";
+protected static final String TEST_COSMOS_CONTAINER_NAME =
"nifi-test-container";
+protected static final String TEST_COSMOS_PARTITION_KEY_FIELD_NAME =
"category";
+protected static CosmosClient client;
+protected static CosmosDatabase cdb;
+protected static CosmosContainer container;
+
+static {
+final FileInputStream fis;
+CONFIG = new Properties();
+try {
+fis = new FileInputStream(CREDENTIALS_FILE);
+try {
+CONFIG.load(fis);
+} catch (IOException e) {
+fail("Could not open credentials file " + CREDENTIALS_FILE +
": " + e.getLocalizedMessage());
+} finally {
+FileUtils.closeQuietly(fis);
+}
+} catch (FileNotFoundException e) {
+fail("Could not open credentials file " + CREDENTIALS_FILE + ": "
+ e.getLocalizedMessage());
+}
+}
+
+protected static String getComosURI() {
+return CONFIG.getProperty("cosmosURI");
+}
+
+protected static String getCosmosKey() {
+return CONFIG.getProperty("cosmosKey");
+}
+
+protected TestRunner runner;
+
+@BeforeClass
+public static void createTestDBContainerIfNeeded() throws
CosmosClientException {
+final String testDBURI = getComosURI();
+final String testDBContainer = getCosmosKey();
+client = new CosmosClientBuilder()
+.endpoint(testDBURI)
+.key(testDBContainer)
+.connectionPolicy(ConnectionPolicy.getDefaultPolicy())
+.buildClient();
+cdb =
client.createDatabaseIfNotExists(TEST_COSMOS_DB_NAME).getDatabase();
+CosmosContainerProperties containerProperties =
+new CosmosContainerProperties(TEST_COSMOS_CONTAINER_NAME,
"/"+TEST_COSMOS_PARTITION_KEY_FIELD_NAME);
+container = cdb.createContainerIfNotExists(containerProperties,
400).getContainer();
+assertNotNull(container);
+}
+
+@AfterClass
+public static void dropTestDBAndContainer() throws CosmosClientException {
+resetTestCosmosConnection();
+if(container != null) {
+
[GitHub] [nifi] sjyang18 commented on a change in pull request #4253: NIFI-7406: PutAzureCosmosRecord Processor
sjyang18 commented on a change in pull request #4253:
URL: https://github.com/apache/nifi/pull/4253#discussion_r421165100
##
File path:
nifi-nar-bundles/nifi-azure-bundle/nifi-azure-processors/src/main/java/org/apache/nifi/processors/azure/cosmos/document/PutAzureCosmosDBRecord.java
##
@@ -0,0 +1,186 @@
+/*
+ * 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.processors.azure.cosmos.document;
+
+import org.apache.nifi.annotation.behavior.EventDriven;
+import org.apache.nifi.annotation.behavior.SystemResourceConsideration;
+import org.apache.nifi.annotation.behavior.InputRequirement;
+import org.apache.nifi.annotation.behavior.InputRequirement.Requirement;
+import org.apache.nifi.annotation.behavior.SystemResource;
+import org.apache.nifi.annotation.documentation.CapabilityDescription;
+import org.apache.nifi.annotation.documentation.Tags;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.flowfile.FlowFile;
+import org.apache.nifi.logging.ComponentLog;
+import org.apache.nifi.processor.ProcessContext;
+import org.apache.nifi.processor.ProcessSession;
+import org.apache.nifi.processor.Relationship;
+import org.apache.nifi.processor.exception.ProcessException;
+import org.apache.nifi.processor.util.StandardValidators;
+import org.apache.nifi.schema.access.SchemaNotFoundException;
+import org.apache.nifi.serialization.MalformedRecordException;
+import org.apache.nifi.serialization.RecordReader;
+import org.apache.nifi.serialization.RecordReaderFactory;
+import org.apache.nifi.serialization.record.Record;
+import org.apache.nifi.serialization.record.RecordFieldType;
+import org.apache.nifi.serialization.record.RecordSchema;
+import org.apache.nifi.serialization.record.util.DataTypeUtils;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+import com.azure.cosmos.CosmosClientException;
+import com.azure.cosmos.implementation.ConflictException;
+
+@EventDriven
+@Tags({ "azure", "cosmos", "insert", "record", "put" })
+@InputRequirement(Requirement.INPUT_REQUIRED)
+@CapabilityDescription("This processor is a record-aware processor for
inserting data into Azure CosmosDB with Core SQL API. It uses a configured
record reader and " +
+"schema to read an incoming record set from the body of a flowfile and
then inserts those records into " +
+"a configured Cosmos Container.")
+@SystemResourceConsideration(resource = SystemResource.MEMORY)
+public class PutAzureCosmosDBRecord extends AbstractAzureCosmosDBProcessor {
+
+static final PropertyDescriptor RECORD_READER_FACTORY = new
PropertyDescriptor.Builder()
+.name("record-reader")
+.displayName("Record Reader")
+.description("Specifies the Controller Service to use for parsing
incoming data and determining the data's schema")
+.identifiesControllerService(RecordReaderFactory.class)
+.required(true)
+.build();
+
+static final PropertyDescriptor INSERT_BATCH_SIZE = new
PropertyDescriptor.Builder()
+.name("insert-batch-size")
+.displayName("Insert Batch Size")
+.description("The number of records to group together for one single
insert operation against Azure CosmosDB.")
+.defaultValue("20")
+.required(false)
+.addValidator(StandardValidators.POSITIVE_INTEGER_VALIDATOR)
+.build();
+
+private final static Set relationships;
+private final static List propertyDescriptors;
+
+static {
+List _propertyDescriptors = new ArrayList<>();
+_propertyDescriptors.addAll(descriptors);
+_propertyDescriptors.add(RECORD_READER_FACTORY);
+_propertyDescriptors.add(INSERT_BATCH_SIZE);
+propertyDescriptors =
Collections.unmodifiableList(_propertyDescriptors);
+
+final Set _relationships = new HashSet<>();
+_relationships.add(REL_SUCCESS);
+_relationships.add(REL_FAILURE);
+relationships = Collections.unmodifiableSet(_relationships);
+}
+
+@Override
+public Set getRelationships() {
+
[GitHub] [nifi] sjyang18 commented on a change in pull request #4253: NIFI-7406: PutAzureCosmosRecord Processor
sjyang18 commented on a change in pull request #4253:
URL: https://github.com/apache/nifi/pull/4253#discussion_r421165270
##
File path:
nifi-nar-bundles/nifi-azure-bundle/nifi-azure-processors/src/main/java/org/apache/nifi/services/azure/cosmos/document/AzureCosmosDBConnectionControllerService.java
##
@@ -0,0 +1,188 @@
+/*
+ * 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.services.azure.cosmos.document;
+
+import org.apache.nifi.annotation.documentation.CapabilityDescription;
+import org.apache.nifi.annotation.documentation.Tags;
+
+import org.apache.nifi.annotation.lifecycle.OnEnabled;
+import org.apache.nifi.annotation.lifecycle.OnStopped;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.components.ValidationContext;
+import org.apache.nifi.components.ValidationResult;
+import org.apache.nifi.controller.AbstractControllerService;
+import org.apache.nifi.controller.ConfigurationContext;
+import org.apache.nifi.processor.util.StandardValidators;
+import org.apache.nifi.services.azure.cosmos.AzureCosmosDBConnectionService;
+import org.apache.nifi.util.StringUtils;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.List;
+
+import com.azure.cosmos.ConnectionPolicy;
+import com.azure.cosmos.ConsistencyLevel;
+import com.azure.cosmos.CosmosClient;
+import com.azure.cosmos.CosmosClientBuilder;
+
+@Tags({"azure", "cosmos", "document", "service"})
+@CapabilityDescription(
+"Provides a controller service that configures a connection to Azure
Cosmos DB (Document API or Core SQL API recently renamed) " +
+" and provides access to that connection to other
AzureCosmosDB-related components."
+)
+public class AzureCosmosDBConnectionControllerService extends
AbstractControllerService implements AzureCosmosDBConnectionService {
+static final String CONSISTENCY_STRONG = "STRONG";
+static final String CONSISTENCY_BOUNDED_STALENESS= "BOUNDED_STALENESS";
+static final String CONSISTENCY_SESSION = "SESSION";
+static final String CONSISTENCY_CONSISTENT_PREFIX = "CONSISTENT_PREFIX";
+static final String CONSISTENCY_EVENTUAL = "EVENTUAL";
Review comment:
moved this definition to AzureCosmosDBUtils class and reused.
##
File path:
nifi-nar-bundles/nifi-azure-bundle/nifi-azure-processors/src/main/java/org/apache/nifi/services/azure/cosmos/document/AzureCosmosDBConnectionControllerService.java
##
@@ -0,0 +1,188 @@
+/*
+ * 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.services.azure.cosmos.document;
+
+import org.apache.nifi.annotation.documentation.CapabilityDescription;
+import org.apache.nifi.annotation.documentation.Tags;
+
+import org.apache.nifi.annotation.lifecycle.OnEnabled;
+import org.apache.nifi.annotation.lifecycle.OnStopped;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.components.ValidationContext;
+import org.apache.nifi.components.ValidationResult;
+import org.apache.nifi.controller.AbstractControllerService;
+import org.apache.nifi.controller.ConfigurationContext;
+import org.apache.nifi.processor.util.StandardValidators;
+import org.apache.nifi.services.azure.cosmos.AzureCosmosDBConnectionService;
+import org.apache.nifi.util.StringUtils;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.List;
+
+import com.azure.cosmos.ConnectionPolicy;
+import
[GitHub] [nifi] sjyang18 commented on a change in pull request #4253: NIFI-7406: PutAzureCosmosRecord Processor
sjyang18 commented on a change in pull request #4253:
URL: https://github.com/apache/nifi/pull/4253#discussion_r421164958
##
File path: nifi-nar-bundles/nifi-azure-bundle/nifi-azure-processors/pom.xml
##
@@ -56,6 +56,28 @@
1.12.0-SNAPSHOT
provided
+
+com.azure
+azure-core
+1.3.0
+
+
+com.fasterxml.jackson.dataformat
+jackson-dataformat-xml
+
+
+
+
+com.azure
+azure-identity
Review comment:
dropped. It came from managed identity PR and this dependency is not
needed for this PR.
##
File path:
nifi-nar-bundles/nifi-azure-bundle/nifi-azure-processors/src/main/java/org/apache/nifi/processors/azure/cosmos/document/PutAzureCosmosDBRecord.java
##
@@ -0,0 +1,186 @@
+/*
+ * 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.processors.azure.cosmos.document;
+
+import org.apache.nifi.annotation.behavior.EventDriven;
+import org.apache.nifi.annotation.behavior.SystemResourceConsideration;
+import org.apache.nifi.annotation.behavior.InputRequirement;
+import org.apache.nifi.annotation.behavior.InputRequirement.Requirement;
+import org.apache.nifi.annotation.behavior.SystemResource;
+import org.apache.nifi.annotation.documentation.CapabilityDescription;
+import org.apache.nifi.annotation.documentation.Tags;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.flowfile.FlowFile;
+import org.apache.nifi.logging.ComponentLog;
+import org.apache.nifi.processor.ProcessContext;
+import org.apache.nifi.processor.ProcessSession;
+import org.apache.nifi.processor.Relationship;
+import org.apache.nifi.processor.exception.ProcessException;
+import org.apache.nifi.processor.util.StandardValidators;
+import org.apache.nifi.schema.access.SchemaNotFoundException;
+import org.apache.nifi.serialization.MalformedRecordException;
+import org.apache.nifi.serialization.RecordReader;
+import org.apache.nifi.serialization.RecordReaderFactory;
+import org.apache.nifi.serialization.record.Record;
+import org.apache.nifi.serialization.record.RecordFieldType;
+import org.apache.nifi.serialization.record.RecordSchema;
+import org.apache.nifi.serialization.record.util.DataTypeUtils;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+import com.azure.cosmos.CosmosClientException;
+import com.azure.cosmos.implementation.ConflictException;
+
+@EventDriven
+@Tags({ "azure", "cosmos", "insert", "record", "put" })
+@InputRequirement(Requirement.INPUT_REQUIRED)
+@CapabilityDescription("This processor is a record-aware processor for
inserting data into Azure CosmosDB with Core SQL API. It uses a configured
record reader and " +
+"schema to read an incoming record set from the body of a flowfile and
then inserts those records into " +
+"a configured Cosmos Container.")
+@SystemResourceConsideration(resource = SystemResource.MEMORY)
+public class PutAzureCosmosDBRecord extends AbstractAzureCosmosDBProcessor {
+
+static final PropertyDescriptor RECORD_READER_FACTORY = new
PropertyDescriptor.Builder()
+.name("record-reader")
+.displayName("Record Reader")
+.description("Specifies the Controller Service to use for parsing
incoming data and determining the data's schema")
+.identifiesControllerService(RecordReaderFactory.class)
+.required(true)
+.build();
+
+static final PropertyDescriptor INSERT_BATCH_SIZE = new
PropertyDescriptor.Builder()
+.name("insert-batch-size")
+.displayName("Insert Batch Size")
+.description("The number of records to group together for one single
insert operation against Azure CosmosDB.")
+.defaultValue("20")
+.required(false)
+.addValidator(StandardValidators.POSITIVE_INTEGER_VALIDATOR)
+.build();
+
+private final static Set relationships;
+private final static List
[GitHub] [nifi] sjyang18 commented on a change in pull request #4253: NIFI-7406: PutAzureCosmosRecord Processor
sjyang18 commented on a change in pull request #4253:
URL: https://github.com/apache/nifi/pull/4253#discussion_r421164676
##
File path:
nifi-nar-bundles/nifi-azure-bundle/nifi-azure-processors/src/main/java/org/apache/nifi/processors/azure/cosmos/document/AbstractAzureCosmosDBProcessor.java
##
@@ -0,0 +1,337 @@
+/*
+ * 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.processors.azure.cosmos.document;
+
+import java.io.ByteArrayInputStream;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+
+import com.azure.cosmos.ConnectionPolicy;
+import com.azure.cosmos.ConsistencyLevel;
+import com.azure.cosmos.CosmosClient;
+import com.azure.cosmos.CosmosClientBuilder;
+import com.azure.cosmos.CosmosClientException;
+import com.azure.cosmos.CosmosContainer;
+import com.azure.cosmos.CosmosDatabase;
+import com.azure.cosmos.models.CosmosContainerProperties;
+
+import org.apache.nifi.annotation.lifecycle.OnScheduled;
+import org.apache.nifi.annotation.lifecycle.OnStopped;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.components.ValidationContext;
+import org.apache.nifi.components.ValidationResult;
+import org.apache.nifi.logging.ComponentLog;
+
+import org.apache.nifi.flowfile.FlowFile;
+import org.apache.nifi.processor.AbstractProcessor;
+import org.apache.nifi.processor.ProcessContext;
+import org.apache.nifi.processor.ProcessSession;
+import org.apache.nifi.processor.Relationship;
+import org.apache.nifi.processor.util.StandardValidators;
+import org.apache.nifi.services.azure.cosmos.AzureCosmosDBConnectionService;
+
+public abstract class AbstractAzureCosmosDBProcessor extends AbstractProcessor
{
+static final String CONSISTENCY_STRONG = "STRONG";
+static final String CONSISTENCY_BOUNDED_STALENESS= "BOUNDED_STALENESS";
+static final String CONSISTENCY_SESSION = "SESSION";
+static final String CONSISTENCY_CONSISTENT_PREFIX = "CONSISTENT_PREFIX";
+static final String CONSISTENCY_EVENTUAL = "EVENTUAL";
+
+static final Relationship REL_SUCCESS = new
Relationship.Builder().name("success")
+.description("All FlowFiles that are written to Azure Cosmos DB are routed
to this relationship").build();
+
+static final Relationship REL_FAILURE = new
Relationship.Builder().name("failure")
+.description("All FlowFiles that cannot be written to Azure Cosmos DB are
routed to this relationship").build();
+
+static final Relationship REL_ORIGINAL = new Relationship.Builder()
+.name("original")
+.description("All input FlowFiles that are part of a successful
query execution go here.")
+.build();
+
+static final PropertyDescriptor CONNECTION_SERVICE = new
PropertyDescriptor.Builder()
+.name("azure-cosmos-db-connection-service")
+.displayName("Azure Cosmos DB Connection Service")
+.description("If configured, this property will use the assigned for
retrieving connection string info.")
+.required(false)
+.identifiesControllerService(AzureCosmosDBConnectionService.class)
+.build();
+
+
+static final PropertyDescriptor URI = new PropertyDescriptor.Builder()
+.name("azure-cosmos-db-uri")
+.displayName("Azure Cosmos DB URI")
+.description("Azure Cosmos DB URI, typically of the form:
https://{databaseaccount}.documents.azure.com:443/. Note this host URL is for
Azure Cosmos DB with CORE SQL API")
+.required(false)
+.addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+.build();
+
+static final PropertyDescriptor DB_ACCESS_KEY = new
PropertyDescriptor.Builder()
+.name("azure-cosmos-db-key")
+.displayName("Azure Cosmos DB Access Key")
+.description("Azure Cosmos DB Access Key from Azure Portal
(Settings->Keys)")
+.required(false)
+.addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+.sensitive(true)
+.build();
+
+static final PropertyDescriptor DATABASE_NAME = new
PropertyDescriptor.Builder()
+.name("azure-cosmos-db-name")
+.displayName("Azure Cosmos DB
[GitHub] [nifi] sjyang18 commented on a change in pull request #4253: NIFI-7406: PutAzureCosmosRecord Processor
sjyang18 commented on a change in pull request #4253:
URL: https://github.com/apache/nifi/pull/4253#discussion_r421164175
##
File path:
nifi-nar-bundles/nifi-azure-bundle/nifi-azure-processors/src/main/java/org/apache/nifi/processors/azure/cosmos/document/AbstractAzureCosmosDBProcessor.java
##
@@ -0,0 +1,337 @@
+/*
+ * 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.processors.azure.cosmos.document;
+
+import java.io.ByteArrayInputStream;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+
+import com.azure.cosmos.ConnectionPolicy;
+import com.azure.cosmos.ConsistencyLevel;
+import com.azure.cosmos.CosmosClient;
+import com.azure.cosmos.CosmosClientBuilder;
+import com.azure.cosmos.CosmosClientException;
+import com.azure.cosmos.CosmosContainer;
+import com.azure.cosmos.CosmosDatabase;
+import com.azure.cosmos.models.CosmosContainerProperties;
+
+import org.apache.nifi.annotation.lifecycle.OnScheduled;
+import org.apache.nifi.annotation.lifecycle.OnStopped;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.components.ValidationContext;
+import org.apache.nifi.components.ValidationResult;
+import org.apache.nifi.logging.ComponentLog;
+
+import org.apache.nifi.flowfile.FlowFile;
+import org.apache.nifi.processor.AbstractProcessor;
+import org.apache.nifi.processor.ProcessContext;
+import org.apache.nifi.processor.ProcessSession;
+import org.apache.nifi.processor.Relationship;
+import org.apache.nifi.processor.util.StandardValidators;
+import org.apache.nifi.services.azure.cosmos.AzureCosmosDBConnectionService;
+
+public abstract class AbstractAzureCosmosDBProcessor extends AbstractProcessor
{
+static final String CONSISTENCY_STRONG = "STRONG";
+static final String CONSISTENCY_BOUNDED_STALENESS= "BOUNDED_STALENESS";
+static final String CONSISTENCY_SESSION = "SESSION";
+static final String CONSISTENCY_CONSISTENT_PREFIX = "CONSISTENT_PREFIX";
+static final String CONSISTENCY_EVENTUAL = "EVENTUAL";
+
+static final Relationship REL_SUCCESS = new
Relationship.Builder().name("success")
+.description("All FlowFiles that are written to Azure Cosmos DB are routed
to this relationship").build();
+
+static final Relationship REL_FAILURE = new
Relationship.Builder().name("failure")
+.description("All FlowFiles that cannot be written to Azure Cosmos DB are
routed to this relationship").build();
+
+static final Relationship REL_ORIGINAL = new Relationship.Builder()
+.name("original")
+.description("All input FlowFiles that are part of a successful
query execution go here.")
+.build();
+
+static final PropertyDescriptor CONNECTION_SERVICE = new
PropertyDescriptor.Builder()
+.name("azure-cosmos-db-connection-service")
+.displayName("Azure Cosmos DB Connection Service")
+.description("If configured, this property will use the assigned for
retrieving connection string info.")
+.required(false)
+.identifiesControllerService(AzureCosmosDBConnectionService.class)
+.build();
+
+
+static final PropertyDescriptor URI = new PropertyDescriptor.Builder()
+.name("azure-cosmos-db-uri")
+.displayName("Azure Cosmos DB URI")
+.description("Azure Cosmos DB URI, typically of the form:
https://{databaseaccount}.documents.azure.com:443/. Note this host URL is for
Azure Cosmos DB with CORE SQL API")
+.required(false)
+.addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+.build();
+
+static final PropertyDescriptor DB_ACCESS_KEY = new
PropertyDescriptor.Builder()
+.name("azure-cosmos-db-key")
+.displayName("Azure Cosmos DB Access Key")
+.description("Azure Cosmos DB Access Key from Azure Portal
(Settings->Keys)")
+.required(false)
+.addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+.sensitive(true)
+.build();
+
+static final PropertyDescriptor DATABASE_NAME = new
PropertyDescriptor.Builder()
+.name("azure-cosmos-db-name")
+.displayName("Azure Cosmos DB
[GitHub] [nifi] sjyang18 commented on a change in pull request #4253: NIFI-7406: PutAzureCosmosRecord Processor
sjyang18 commented on a change in pull request #4253:
URL: https://github.com/apache/nifi/pull/4253#discussion_r421164141
##
File path:
nifi-nar-bundles/nifi-azure-bundle/nifi-azure-processors/src/main/java/org/apache/nifi/processors/azure/cosmos/document/AbstractAzureCosmosDBProcessor.java
##
@@ -0,0 +1,337 @@
+/*
+ * 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.processors.azure.cosmos.document;
+
+import java.io.ByteArrayInputStream;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+
+import com.azure.cosmos.ConnectionPolicy;
+import com.azure.cosmos.ConsistencyLevel;
+import com.azure.cosmos.CosmosClient;
+import com.azure.cosmos.CosmosClientBuilder;
+import com.azure.cosmos.CosmosClientException;
+import com.azure.cosmos.CosmosContainer;
+import com.azure.cosmos.CosmosDatabase;
+import com.azure.cosmos.models.CosmosContainerProperties;
+
+import org.apache.nifi.annotation.lifecycle.OnScheduled;
+import org.apache.nifi.annotation.lifecycle.OnStopped;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.components.ValidationContext;
+import org.apache.nifi.components.ValidationResult;
+import org.apache.nifi.logging.ComponentLog;
+
+import org.apache.nifi.flowfile.FlowFile;
+import org.apache.nifi.processor.AbstractProcessor;
+import org.apache.nifi.processor.ProcessContext;
+import org.apache.nifi.processor.ProcessSession;
+import org.apache.nifi.processor.Relationship;
+import org.apache.nifi.processor.util.StandardValidators;
+import org.apache.nifi.services.azure.cosmos.AzureCosmosDBConnectionService;
+
+public abstract class AbstractAzureCosmosDBProcessor extends AbstractProcessor
{
+static final String CONSISTENCY_STRONG = "STRONG";
+static final String CONSISTENCY_BOUNDED_STALENESS= "BOUNDED_STALENESS";
+static final String CONSISTENCY_SESSION = "SESSION";
+static final String CONSISTENCY_CONSISTENT_PREFIX = "CONSISTENT_PREFIX";
+static final String CONSISTENCY_EVENTUAL = "EVENTUAL";
+
+static final Relationship REL_SUCCESS = new
Relationship.Builder().name("success")
+.description("All FlowFiles that are written to Azure Cosmos DB are routed
to this relationship").build();
+
+static final Relationship REL_FAILURE = new
Relationship.Builder().name("failure")
+.description("All FlowFiles that cannot be written to Azure Cosmos DB are
routed to this relationship").build();
+
+static final Relationship REL_ORIGINAL = new Relationship.Builder()
+.name("original")
+.description("All input FlowFiles that are part of a successful
query execution go here.")
+.build();
+
+static final PropertyDescriptor CONNECTION_SERVICE = new
PropertyDescriptor.Builder()
+.name("azure-cosmos-db-connection-service")
+.displayName("Azure Cosmos DB Connection Service")
+.description("If configured, this property will use the assigned for
retrieving connection string info.")
+.required(false)
+.identifiesControllerService(AzureCosmosDBConnectionService.class)
+.build();
+
+
+static final PropertyDescriptor URI = new PropertyDescriptor.Builder()
+.name("azure-cosmos-db-uri")
+.displayName("Azure Cosmos DB URI")
+.description("Azure Cosmos DB URI, typically of the form:
https://{databaseaccount}.documents.azure.com:443/. Note this host URL is for
Azure Cosmos DB with CORE SQL API")
+.required(false)
+.addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+.build();
+
+static final PropertyDescriptor DB_ACCESS_KEY = new
PropertyDescriptor.Builder()
+.name("azure-cosmos-db-key")
+.displayName("Azure Cosmos DB Access Key")
+.description("Azure Cosmos DB Access Key from Azure Portal
(Settings->Keys)")
+.required(false)
+.addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+.sensitive(true)
+.build();
+
+static final PropertyDescriptor DATABASE_NAME = new
PropertyDescriptor.Builder()
+.name("azure-cosmos-db-name")
+.displayName("Azure Cosmos DB
[GitHub] [nifi] sjyang18 commented on a change in pull request #4253: NIFI-7406: PutAzureCosmosRecord Processor
sjyang18 commented on a change in pull request #4253:
URL: https://github.com/apache/nifi/pull/4253#discussion_r421163785
##
File path:
nifi-nar-bundles/nifi-azure-bundle/nifi-azure-processors/src/main/java/org/apache/nifi/processors/azure/cosmos/document/AbstractAzureCosmosDBProcessor.java
##
@@ -0,0 +1,337 @@
+/*
+ * 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.processors.azure.cosmos.document;
+
+import java.io.ByteArrayInputStream;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+
+import com.azure.cosmos.ConnectionPolicy;
+import com.azure.cosmos.ConsistencyLevel;
+import com.azure.cosmos.CosmosClient;
+import com.azure.cosmos.CosmosClientBuilder;
+import com.azure.cosmos.CosmosClientException;
+import com.azure.cosmos.CosmosContainer;
+import com.azure.cosmos.CosmosDatabase;
+import com.azure.cosmos.models.CosmosContainerProperties;
+
+import org.apache.nifi.annotation.lifecycle.OnScheduled;
+import org.apache.nifi.annotation.lifecycle.OnStopped;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.components.ValidationContext;
+import org.apache.nifi.components.ValidationResult;
+import org.apache.nifi.logging.ComponentLog;
+
+import org.apache.nifi.flowfile.FlowFile;
+import org.apache.nifi.processor.AbstractProcessor;
+import org.apache.nifi.processor.ProcessContext;
+import org.apache.nifi.processor.ProcessSession;
+import org.apache.nifi.processor.Relationship;
+import org.apache.nifi.processor.util.StandardValidators;
+import org.apache.nifi.services.azure.cosmos.AzureCosmosDBConnectionService;
+
+public abstract class AbstractAzureCosmosDBProcessor extends AbstractProcessor
{
+static final String CONSISTENCY_STRONG = "STRONG";
+static final String CONSISTENCY_BOUNDED_STALENESS= "BOUNDED_STALENESS";
+static final String CONSISTENCY_SESSION = "SESSION";
+static final String CONSISTENCY_CONSISTENT_PREFIX = "CONSISTENT_PREFIX";
+static final String CONSISTENCY_EVENTUAL = "EVENTUAL";
+
+static final Relationship REL_SUCCESS = new
Relationship.Builder().name("success")
+.description("All FlowFiles that are written to Azure Cosmos DB are routed
to this relationship").build();
+
+static final Relationship REL_FAILURE = new
Relationship.Builder().name("failure")
+.description("All FlowFiles that cannot be written to Azure Cosmos DB are
routed to this relationship").build();
+
+static final Relationship REL_ORIGINAL = new Relationship.Builder()
+.name("original")
+.description("All input FlowFiles that are part of a successful
query execution go here.")
+.build();
+
+static final PropertyDescriptor CONNECTION_SERVICE = new
PropertyDescriptor.Builder()
+.name("azure-cosmos-db-connection-service")
+.displayName("Azure Cosmos DB Connection Service")
+.description("If configured, this property will use the assigned for
retrieving connection string info.")
+.required(false)
+.identifiesControllerService(AzureCosmosDBConnectionService.class)
+.build();
+
+
+static final PropertyDescriptor URI = new PropertyDescriptor.Builder()
+.name("azure-cosmos-db-uri")
+.displayName("Azure Cosmos DB URI")
+.description("Azure Cosmos DB URI, typically of the form:
https://{databaseaccount}.documents.azure.com:443/. Note this host URL is for
Azure Cosmos DB with CORE SQL API")
+.required(false)
+.addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
Review comment:
fixed and moved to AzureCosmosDBUtils class.
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.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
[GitHub] [nifi] sjyang18 commented on a change in pull request #4253: NIFI-7406: PutAzureCosmosRecord Processor
sjyang18 commented on a change in pull request #4253:
URL: https://github.com/apache/nifi/pull/4253#discussion_r421162417
##
File path: nifi-nar-bundles/nifi-azure-bundle/nifi-azure-processors/pom.xml
##
@@ -65,21 +87,74 @@
com.microsoft.azure
azure-eventhubs-eph
${azure-eventhubs-eph.version}
+
+
+
+com.fasterxml.jackson.core
+jackson-core
+
+
+com.fasterxml.jackson.core
+jackson-databind
+
+
com.microsoft.azure
azure-storage
-
+
+
+com.fasterxml.jackson.core
+jackson-core
+
+
+
+
+com.azure
+azure-cosmos
+4.0.1-beta.2
+
+
+com.azure
+azure-core
+
+
+com.google.guava
+guava
+
+
+
Review comment:
fixed
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.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
[GitHub] [nifi] sjyang18 commented on a change in pull request #4253: NIFI-7406: PutAzureCosmosRecord Processor
sjyang18 commented on a change in pull request #4253:
URL: https://github.com/apache/nifi/pull/4253#discussion_r421162515
##
File path:
nifi-nar-bundles/nifi-azure-bundle/nifi-azure-processors/src/main/java/org/apache/nifi/processors/azure/cosmos/document/AbstractAzureCosmosDBProcessor.java
##
@@ -0,0 +1,337 @@
+/*
+ * 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.processors.azure.cosmos.document;
+
+import java.io.ByteArrayInputStream;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+
+import com.azure.cosmos.ConnectionPolicy;
+import com.azure.cosmos.ConsistencyLevel;
+import com.azure.cosmos.CosmosClient;
+import com.azure.cosmos.CosmosClientBuilder;
+import com.azure.cosmos.CosmosClientException;
+import com.azure.cosmos.CosmosContainer;
+import com.azure.cosmos.CosmosDatabase;
+import com.azure.cosmos.models.CosmosContainerProperties;
+
+import org.apache.nifi.annotation.lifecycle.OnScheduled;
+import org.apache.nifi.annotation.lifecycle.OnStopped;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.components.ValidationContext;
+import org.apache.nifi.components.ValidationResult;
+import org.apache.nifi.logging.ComponentLog;
+
+import org.apache.nifi.flowfile.FlowFile;
+import org.apache.nifi.processor.AbstractProcessor;
+import org.apache.nifi.processor.ProcessContext;
+import org.apache.nifi.processor.ProcessSession;
+import org.apache.nifi.processor.Relationship;
+import org.apache.nifi.processor.util.StandardValidators;
+import org.apache.nifi.services.azure.cosmos.AzureCosmosDBConnectionService;
+
+public abstract class AbstractAzureCosmosDBProcessor extends AbstractProcessor
{
+static final String CONSISTENCY_STRONG = "STRONG";
+static final String CONSISTENCY_BOUNDED_STALENESS= "BOUNDED_STALENESS";
+static final String CONSISTENCY_SESSION = "SESSION";
+static final String CONSISTENCY_CONSISTENT_PREFIX = "CONSISTENT_PREFIX";
+static final String CONSISTENCY_EVENTUAL = "EVENTUAL";
+
+static final Relationship REL_SUCCESS = new
Relationship.Builder().name("success")
+.description("All FlowFiles that are written to Azure Cosmos DB are routed
to this relationship").build();
+
+static final Relationship REL_FAILURE = new
Relationship.Builder().name("failure")
+.description("All FlowFiles that cannot be written to Azure Cosmos DB are
routed to this relationship").build();
Review comment:
fixed
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.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
[GitHub] [nifi] sjyang18 commented on a change in pull request #4253: NIFI-7406: PutAzureCosmosRecord Processor
sjyang18 commented on a change in pull request #4253:
URL: https://github.com/apache/nifi/pull/4253#discussion_r421148531
##
File path:
nifi-nar-bundles/nifi-azure-bundle/nifi-azure-processors/src/test/java/org/apache/nifi/processors/azure/cosmos/document/ITAbstractAzureCosmosDBDocument.java
##
@@ -0,0 +1,198 @@
+/*
+ * 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.processors.azure.cosmos.document;
+
+import com.azure.cosmos.ConnectionPolicy;
+import com.azure.cosmos.CosmosClient;
+import com.azure.cosmos.CosmosClientBuilder;
+import com.azure.cosmos.CosmosClientException;
+import com.azure.cosmos.CosmosContainer;
+import com.azure.cosmos.CosmosDatabase;
+
+import com.azure.cosmos.models.CosmosContainerProperties;
+import com.azure.cosmos.models.CosmosItemRequestOptions;
+import com.azure.cosmos.models.FeedOptions;
+import com.azure.cosmos.models.PartitionKey;
+import com.azure.cosmos.util.CosmosPagedIterable;
+import com.fasterxml.jackson.databind.JsonNode;
+
+import org.apache.nifi.processor.Processor;
+import
org.apache.nifi.services.azure.cosmos.document.AzureCosmosDBConnectionControllerService;
+import org.apache.nifi.util.TestRunner;
+import org.apache.nifi.util.TestRunners;
+import org.apache.nifi.util.file.FileUtils;
+import org.junit.AfterClass;
+import org.junit.Before;
+import org.junit.BeforeClass;
+
+import java.io.FileInputStream;
+import java.io.FileNotFoundException;
+import java.io.IOException;
+import java.util.Properties;
+import java.util.logging.Logger;
+
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.fail;
+
+public abstract class ITAbstractAzureCosmosDBDocument {
+static Logger logger =
Logger.getLogger(ITAbstractAzureCosmosDBDocument.class.getName());
+
+private static final Properties CONFIG;
+
+private static final String CREDENTIALS_FILE =
System.getProperty("user.home") + "/azure-credentials.PROPERTIES";
Review comment:
I think this is an IT assumption in nifi-azure-bundle. I followed the
same assumption from the storage IT (AbstractAzureStorageIT.java). Did I
mis-understand your suggestion?
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.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
[GitHub] [nifi] sjyang18 commented on a change in pull request #4253: NIFI-7406: PutAzureCosmosRecord Processor
sjyang18 commented on a change in pull request #4253:
URL: https://github.com/apache/nifi/pull/4253#discussion_r421142466
##
File path: nifi-nar-bundles/nifi-azure-bundle/nifi-azure-processors/pom.xml
##
@@ -65,21 +87,74 @@
com.microsoft.azure
azure-eventhubs-eph
${azure-eventhubs-eph.version}
+
+
+
+com.fasterxml.jackson.core
+jackson-core
+
+
+com.fasterxml.jackson.core
+jackson-databind
+
+
com.microsoft.azure
azure-storage
-
+
+
+com.fasterxml.jackson.core
+jackson-core
+
+
+
+
+com.azure
+azure-cosmos
+4.0.1-beta.2
+
+
+com.azure
+azure-core
+
+
+com.google.guava
+guava
+
+
+
com.azure
azure-storage-file-datalake
12.0.1
+
+
+com.azure
+azure-core
+
+
com.fasterxml.jackson.core
jackson-core
-2.10.3
+${jackson.version}
Review comment:
I override this jackson.version variable value to 2.10.3 in
nifi-auzure-bundle/pom.xml. So, it is the same value.
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.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
[GitHub] [nifi] sjyang18 commented on a change in pull request #4253: NIFI-7406: PutAzureCosmosRecord Processor
sjyang18 commented on a change in pull request #4253:
URL: https://github.com/apache/nifi/pull/4253#discussion_r421138897
##
File path:
nifi-nar-bundles/nifi-azure-bundle/nifi-azure-processors/src/main/java/org/apache/nifi/processors/azure/cosmos/document/PutAzureCosmosDBRecord.java
##
@@ -0,0 +1,186 @@
+/*
+ * 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.processors.azure.cosmos.document;
+
+import org.apache.nifi.annotation.behavior.EventDriven;
+import org.apache.nifi.annotation.behavior.SystemResourceConsideration;
+import org.apache.nifi.annotation.behavior.InputRequirement;
+import org.apache.nifi.annotation.behavior.InputRequirement.Requirement;
+import org.apache.nifi.annotation.behavior.SystemResource;
+import org.apache.nifi.annotation.documentation.CapabilityDescription;
+import org.apache.nifi.annotation.documentation.Tags;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.flowfile.FlowFile;
+import org.apache.nifi.logging.ComponentLog;
+import org.apache.nifi.processor.ProcessContext;
+import org.apache.nifi.processor.ProcessSession;
+import org.apache.nifi.processor.Relationship;
+import org.apache.nifi.processor.exception.ProcessException;
+import org.apache.nifi.processor.util.StandardValidators;
+import org.apache.nifi.schema.access.SchemaNotFoundException;
+import org.apache.nifi.serialization.MalformedRecordException;
+import org.apache.nifi.serialization.RecordReader;
+import org.apache.nifi.serialization.RecordReaderFactory;
+import org.apache.nifi.serialization.record.Record;
+import org.apache.nifi.serialization.record.RecordFieldType;
+import org.apache.nifi.serialization.record.RecordSchema;
+import org.apache.nifi.serialization.record.util.DataTypeUtils;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+import com.azure.cosmos.CosmosClientException;
+import com.azure.cosmos.implementation.ConflictException;
+
+@EventDriven
+@Tags({ "azure", "cosmos", "insert", "record", "put" })
+@InputRequirement(Requirement.INPUT_REQUIRED)
+@CapabilityDescription("This processor is a record-aware processor for
inserting data into Azure CosmosDB with Core SQL API. It uses a configured
record reader and " +
+"schema to read an incoming record set from the body of a flowfile and
then inserts those records into " +
+"a configured Cosmos Container.")
+@SystemResourceConsideration(resource = SystemResource.MEMORY)
+public class PutAzureCosmosDBRecord extends AbstractAzureCosmosDBProcessor {
+
+static final PropertyDescriptor RECORD_READER_FACTORY = new
PropertyDescriptor.Builder()
+.name("record-reader")
+.displayName("Record Reader")
+.description("Specifies the Controller Service to use for parsing
incoming data and determining the data's schema")
+.identifiesControllerService(RecordReaderFactory.class)
+.required(true)
+.build();
+
+static final PropertyDescriptor INSERT_BATCH_SIZE = new
PropertyDescriptor.Builder()
+.name("insert-batch-size")
+.displayName("Insert Batch Size")
+.description("The number of records to group together for one single
insert operation against Azure CosmosDB.")
+.defaultValue("20")
+.required(false)
+.addValidator(StandardValidators.POSITIVE_INTEGER_VALIDATOR)
+.build();
+
+private final static Set relationships;
+private final static List propertyDescriptors;
+
+static {
+List _propertyDescriptors = new ArrayList<>();
+_propertyDescriptors.addAll(descriptors);
+_propertyDescriptors.add(RECORD_READER_FACTORY);
+_propertyDescriptors.add(INSERT_BATCH_SIZE);
+propertyDescriptors =
Collections.unmodifiableList(_propertyDescriptors);
+
+final Set _relationships = new HashSet<>();
+_relationships.add(REL_SUCCESS);
+_relationships.add(REL_FAILURE);
+relationships = Collections.unmodifiableSet(_relationships);
+}
+
+@Override
+public Set getRelationships() {
+
[GitHub] [nifi] sjyang18 commented on a change in pull request #4253: NIFI-7406: PutAzureCosmosRecord Processor
sjyang18 commented on a change in pull request #4253:
URL: https://github.com/apache/nifi/pull/4253#discussion_r420226726
##
File path:
nifi-nar-bundles/nifi-azure-bundle/nifi-azure-processors/src/main/java/org/apache/nifi/processors/azure/cosmos/document/AbstractAzureCosmosDBProcessor.java
##
@@ -0,0 +1,336 @@
+/*
+ * 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.processors.azure.cosmos.document;
+
+import java.io.ByteArrayInputStream;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+
+import com.azure.cosmos.ConnectionPolicy;
+import com.azure.cosmos.ConsistencyLevel;
+import com.azure.cosmos.CosmosClient;
+import com.azure.cosmos.CosmosClientBuilder;
+import com.azure.cosmos.CosmosClientException;
+import com.azure.cosmos.CosmosContainer;
+import com.azure.cosmos.CosmosDatabase;
+import com.azure.cosmos.models.CosmosContainerProperties;
+
+import org.apache.nifi.annotation.lifecycle.OnScheduled;
+import org.apache.nifi.annotation.lifecycle.OnStopped;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.components.ValidationContext;
+import org.apache.nifi.components.ValidationResult;
+import org.apache.nifi.logging.ComponentLog;
+
+import org.apache.nifi.flowfile.FlowFile;
+import org.apache.nifi.processor.AbstractProcessor;
+import org.apache.nifi.processor.ProcessContext;
+import org.apache.nifi.processor.ProcessSession;
+import org.apache.nifi.processor.Relationship;
+import org.apache.nifi.processor.util.StandardValidators;
+import org.apache.nifi.services.azure.cosmos.AzureCosmosDBConnectionService;
+
+public abstract class AbstractAzureCosmosDBProcessor extends AbstractProcessor
{
+static final String CONSISTENCY_STRONG = "STRONG";
+static final String CONSISTENCY_BOUNDED_STALENESS= "BOUNDED_STALENESS";
+static final String CONSISTENCY_SESSION = "SESSION";
+static final String CONSISTENCY_CONSISTENT_PREFIX = "CONSISTENT_PREFIX";
+static final String CONSISTENCY_EVENTUAL = "EVENTUAL";
+
+static final Relationship REL_SUCCESS = new
Relationship.Builder().name("success")
+.description("All FlowFiles that are written to Azure Cosmos DB are routed
to this relationship").build();
+
+static final Relationship REL_FAILURE = new
Relationship.Builder().name("failure")
+.description("All FlowFiles that cannot be written to Azure Cosmos DB are
routed to this relationship").build();
+
+static final Relationship REL_ORIGINAL = new Relationship.Builder()
+.name("original")
+.description("All input FlowFiles that are part of a successful
query execution go here.")
+.build();
+
+static final PropertyDescriptor CONNECTION_SERVICE = new
PropertyDescriptor.Builder()
+.name("azure-cosmos-db-connection-service")
Review comment:
Thanks for reviewing & fixing my mistake. I also added displayName to
"charactor-set" property.
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.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
