You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@nifi.apache.org by GitBox <gi...@apache.org> on 2022/11/07 01:27:10 UTC
[GitHub] [nifi] Lehel44 commented on a diff in pull request #6584: NIFI-10370 Create record oriented PutSnowflake processor
Lehel44 commented on code in PR #6584:
URL: https://github.com/apache/nifi/pull/6584#discussion_r1014298984
##########
nifi-nar-bundles/nifi-extension-utils/nifi-dbcp-base/src/main/java/org/apache/nifi/dbcp/AbstractDBCPConnectionPool.java:
##########
@@ -496,6 +499,15 @@ protected Driver getDriver(final String driverName, final String url) {
}
}
+ /**
+ * Override in subclasses to provide connection properties to the data source
+ *
+ * @return Key-value pairs that will be added as connection properties
+ */
+ protected Map<String, String> getConnectionProperties(final ConfigurationContext context) {
Review Comment:
Can this method be abstract to force overriding?
##########
nifi-nar-bundles/nifi-snowflake-bundle/nifi-snowflake-processors/src/main/java/org/apache/nifi/processors/snowflake/GetSnowflakeIngestStatus.java:
##########
@@ -0,0 +1,208 @@
+/*
+ * 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.snowflake;
+
+import static org.apache.nifi.processors.snowflake.common.Attributes.ATTRIBUTE_STAGED_FILE_PATH;
+
+import java.io.IOException;
+import java.net.URISyntaxException;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+import net.snowflake.ingest.SimpleIngestManager;
+import net.snowflake.ingest.connection.HistoryResponse;
+import net.snowflake.ingest.connection.HistoryResponse.FileEntry;
+import net.snowflake.ingest.connection.IngestResponseException;
+import org.apache.nifi.annotation.behavior.InputRequirement;
+import org.apache.nifi.annotation.behavior.InputRequirement.Requirement;
+import org.apache.nifi.annotation.behavior.ReadsAttribute;
+import org.apache.nifi.annotation.behavior.ReadsAttributes;
+import org.apache.nifi.annotation.behavior.Stateful;
+import org.apache.nifi.annotation.configuration.DefaultSettings;
+import org.apache.nifi.annotation.documentation.CapabilityDescription;
+import org.apache.nifi.annotation.documentation.SeeAlso;
+import org.apache.nifi.annotation.documentation.Tags;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.components.state.Scope;
+import org.apache.nifi.components.state.StateMap;
+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.exception.ProcessException;
+
+@InputRequirement(Requirement.INPUT_REQUIRED)
+@Stateful(scopes = Scope.CLUSTER,
+ description = "The 'begin mark' from the response of a history request is stored to keep track of already requested history time range.")
+@DefaultSettings(penaltyDuration = "5 sec")
+@ReadsAttributes({
+ @ReadsAttribute(attribute = ATTRIBUTE_STAGED_FILE_PATH, description = "The path to the file in the stage")
Review Comment:
Is "in the stage" anything different from being staged? Can it just be e.g. staged file path?
##########
nifi-nar-bundles/nifi-snowflake-bundle/nifi-snowflake-processors/src/main/java/org/apache/nifi/processors/snowflake/PutSnowflakeInternalStage.java:
##########
@@ -0,0 +1,147 @@
+/*
+ * 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.snowflake;
+
+import static org.apache.nifi.processors.snowflake.common.Attributes.ATTRIBUTE_STAGED_FILE_PATH;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.sql.SQLException;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+import org.apache.nifi.annotation.behavior.InputRequirement;
+import org.apache.nifi.annotation.behavior.InputRequirement.Requirement;
+import org.apache.nifi.annotation.behavior.ReadsAttribute;
+import org.apache.nifi.annotation.behavior.ReadsAttributes;
+import org.apache.nifi.annotation.behavior.WritesAttribute;
+import org.apache.nifi.annotation.behavior.WritesAttributes;
+import org.apache.nifi.annotation.documentation.CapabilityDescription;
+import org.apache.nifi.annotation.documentation.SeeAlso;
+import org.apache.nifi.annotation.documentation.Tags;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.expression.ExpressionLanguageScope;
+import org.apache.nifi.flowfile.FlowFile;
+import org.apache.nifi.flowfile.attributes.CoreAttributes;
+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.exception.ProcessException;
+import org.apache.nifi.processor.util.StandardValidators;
+
+@InputRequirement(Requirement.INPUT_REQUIRED)
+@ReadsAttributes({
+ @ReadsAttribute(attribute = "filename", description = "The name of the staged file in the internal stage"),
+ @ReadsAttribute(attribute = "path", description = "The relative path to the staged file in the internal stage")
+})
+@WritesAttributes({
+ @WritesAttribute(attribute = ATTRIBUTE_STAGED_FILE_PATH,
+ description = "The path to the file in the internal stage")
+})
+@Tags({"snowflake", "jdbc", "database", "connection"})
+@CapabilityDescription("Put files into a Snowflake internal stage. The internal stage must be created in the Snowflake account beforehand."
+ + " This processor can be connected to an StartSnowflakeIngest processor to ingest the file in the internal stage")
+@SeeAlso({StartSnowflakeIngest.class, GetSnowflakeIngestStatus.class})
+public class PutSnowflakeInternalStage extends AbstractProcessor {
+
+ static final PropertyDescriptor SNOWFLAKE_CONNECTION_PROVIDER = new PropertyDescriptor.Builder()
+ .name("snowflake-connection-provider")
+ .displayName("Snowflake Connection Provider")
+ .description("Specifies the Controller Service to use for creating SQL connections to Snowflake.")
+ .identifiesControllerService(SnowflakeConnectionProviderService.class)
+ .required(true)
+ .build();
+
+ static final PropertyDescriptor INTERNAL_STAGE_NAME = new PropertyDescriptor.Builder()
+ .name("internal-stage-name")
+ .displayName("Internal Stage Name")
+ .description("The name of the internal stage in the Snowflake account to put files into.")
+ .addValidator(StandardValidators.NON_BLANK_VALIDATOR)
+ .expressionLanguageSupported(ExpressionLanguageScope.VARIABLE_REGISTRY)
Review Comment:
Why not FlowFile scope?
##########
nifi-nar-bundles/nifi-snowflake-bundle/nifi-snowflake-processors/src/main/java/org/apache/nifi/processors/snowflake/PutSnowflakeInternalStage.java:
##########
@@ -0,0 +1,147 @@
+/*
+ * 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.snowflake;
+
+import static org.apache.nifi.processors.snowflake.common.Attributes.ATTRIBUTE_STAGED_FILE_PATH;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.sql.SQLException;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+import org.apache.nifi.annotation.behavior.InputRequirement;
+import org.apache.nifi.annotation.behavior.InputRequirement.Requirement;
+import org.apache.nifi.annotation.behavior.ReadsAttribute;
+import org.apache.nifi.annotation.behavior.ReadsAttributes;
+import org.apache.nifi.annotation.behavior.WritesAttribute;
+import org.apache.nifi.annotation.behavior.WritesAttributes;
+import org.apache.nifi.annotation.documentation.CapabilityDescription;
+import org.apache.nifi.annotation.documentation.SeeAlso;
+import org.apache.nifi.annotation.documentation.Tags;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.expression.ExpressionLanguageScope;
+import org.apache.nifi.flowfile.FlowFile;
+import org.apache.nifi.flowfile.attributes.CoreAttributes;
+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.exception.ProcessException;
+import org.apache.nifi.processor.util.StandardValidators;
+
+@InputRequirement(Requirement.INPUT_REQUIRED)
+@ReadsAttributes({
+ @ReadsAttribute(attribute = "filename", description = "The name of the staged file in the internal stage"),
+ @ReadsAttribute(attribute = "path", description = "The relative path to the staged file in the internal stage")
+})
+@WritesAttributes({
+ @WritesAttribute(attribute = ATTRIBUTE_STAGED_FILE_PATH,
+ description = "The path to the file in the internal stage")
+})
+@Tags({"snowflake", "jdbc", "database", "connection"})
+@CapabilityDescription("Put files into a Snowflake internal stage. The internal stage must be created in the Snowflake account beforehand."
+ + " This processor can be connected to an StartSnowflakeIngest processor to ingest the file in the internal stage")
+@SeeAlso({StartSnowflakeIngest.class, GetSnowflakeIngestStatus.class})
+public class PutSnowflakeInternalStage extends AbstractProcessor {
+
+ static final PropertyDescriptor SNOWFLAKE_CONNECTION_PROVIDER = new PropertyDescriptor.Builder()
+ .name("snowflake-connection-provider")
+ .displayName("Snowflake Connection Provider")
+ .description("Specifies the Controller Service to use for creating SQL connections to Snowflake.")
+ .identifiesControllerService(SnowflakeConnectionProviderService.class)
+ .required(true)
+ .build();
+
+ static final PropertyDescriptor INTERNAL_STAGE_NAME = new PropertyDescriptor.Builder()
+ .name("internal-stage-name")
+ .displayName("Internal Stage Name")
+ .description("The name of the internal stage in the Snowflake account to put files into.")
+ .addValidator(StandardValidators.NON_BLANK_VALIDATOR)
+ .expressionLanguageSupported(ExpressionLanguageScope.VARIABLE_REGISTRY)
+ .required(true)
+ .build();
+
+ static final Relationship REL_SUCCESS = new Relationship.Builder()
+ .name("success")
+ .description("For FlowFiles of successful PUT operation")
+ .build();
+
+ static final Relationship REL_FAILURE = new Relationship.Builder()
+ .name("failure")
+ .description("For FlowFiles of failed PUT operation")
+ .build();
+
+ static final List<PropertyDescriptor> PROPERTIES = Collections.unmodifiableList(Arrays.asList(
+ SNOWFLAKE_CONNECTION_PROVIDER,
+ INTERNAL_STAGE_NAME
+ ));
+
+ static final Set<Relationship> RELATIONSHIPS;
+
+ static {
+ final Set<Relationship> relationships = new HashSet<>();
+ relationships.add(REL_SUCCESS);
+ relationships.add(REL_FAILURE);
+ RELATIONSHIPS = Collections.unmodifiableSet(relationships);
+ }
Review Comment:
```suggestion
static final Set<Relationship> RELATIONSHIPS = Stream.of(REL_SUCCESS, REL_FAILURE)
.collect(collectingAndThen(toSet(), Collections::unmodifiableSet));
```
##########
nifi-nar-bundles/nifi-snowflake-bundle/nifi-snowflake-processors/src/main/java/org/apache/nifi/processors/snowflake/PutSnowflakeInternalStage.java:
##########
@@ -0,0 +1,147 @@
+/*
+ * 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.snowflake;
+
+import static org.apache.nifi.processors.snowflake.common.Attributes.ATTRIBUTE_STAGED_FILE_PATH;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.sql.SQLException;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+import org.apache.nifi.annotation.behavior.InputRequirement;
+import org.apache.nifi.annotation.behavior.InputRequirement.Requirement;
+import org.apache.nifi.annotation.behavior.ReadsAttribute;
+import org.apache.nifi.annotation.behavior.ReadsAttributes;
+import org.apache.nifi.annotation.behavior.WritesAttribute;
+import org.apache.nifi.annotation.behavior.WritesAttributes;
+import org.apache.nifi.annotation.documentation.CapabilityDescription;
+import org.apache.nifi.annotation.documentation.SeeAlso;
+import org.apache.nifi.annotation.documentation.Tags;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.expression.ExpressionLanguageScope;
+import org.apache.nifi.flowfile.FlowFile;
+import org.apache.nifi.flowfile.attributes.CoreAttributes;
+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.exception.ProcessException;
+import org.apache.nifi.processor.util.StandardValidators;
+
+@InputRequirement(Requirement.INPUT_REQUIRED)
+@ReadsAttributes({
+ @ReadsAttribute(attribute = "filename", description = "The name of the staged file in the internal stage"),
+ @ReadsAttribute(attribute = "path", description = "The relative path to the staged file in the internal stage")
+})
+@WritesAttributes({
+ @WritesAttribute(attribute = ATTRIBUTE_STAGED_FILE_PATH,
+ description = "The path to the file in the internal stage")
+})
+@Tags({"snowflake", "jdbc", "database", "connection"})
+@CapabilityDescription("Put files into a Snowflake internal stage. The internal stage must be created in the Snowflake account beforehand."
Review Comment:
```suggestion
@CapabilityDescription("Puts files into a Snowflake internal stage. The internal stage must be created in the Snowflake account beforehand."
```
##########
nifi-nar-bundles/nifi-snowflake-bundle/nifi-snowflake-processors/src/main/resources/docs/org.apache.nifi.processors.snowflake.GetSnowflakeIngestStatus/additionalDetails.html:
##########
@@ -0,0 +1,48 @@
+<!--
+ ~ 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.
+ -->
+
+<!DOCTYPE html>
+<html lang="en">
+<head>
+ <meta charset="utf-8"/>
+ <title>GetSnowflakeIngestStatus</title>
+ <link rel="stylesheet" href="../../../../../css/component-usage.css" type="text/css"/>
+ <style>
+ h2 {margin-top: 4em}
+ h3 {margin-top: 3em}
+ td {text-align: left}
+ </style>
+</head>
+<body>
+
+<h1>GetSnowflakeIngestStatus</h1>
+
+<h3>Description</h3>
+<p>
+ The GetSnowflakeIngestStatus processor can be used to get the status of a staged file ingested by a Snowflake pipe.
+ To wait until a staged file is fully ingested (copied into the table) you should connect this processor's "retry" relationship to itself.
+ The processor requires an upstream connection that provides the path of the staged file to be checked through the "snowflake.staged.file.path" attribute.
+ See StartSnowflakeIngest processor for details about how to properly set up a flow to ingest staged files.
+ <b>
+ NOTE: Snowflake pipes caches the ingested files' path and never ingests the same file multiple times.
Review Comment:
```suggestion
NOTE: Snowflake pipes cache the paths of ingested files and never ingest the same file again/multiple times.
```
##########
nifi-nar-bundles/nifi-snowflake-bundle/nifi-snowflake-processors/src/main/resources/docs/org.apache.nifi.processors.snowflake.GetSnowflakeIngestStatus/additionalDetails.html:
##########
@@ -0,0 +1,48 @@
+<!--
+ ~ 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.
+ -->
+
+<!DOCTYPE html>
+<html lang="en">
+<head>
+ <meta charset="utf-8"/>
+ <title>GetSnowflakeIngestStatus</title>
+ <link rel="stylesheet" href="../../../../../css/component-usage.css" type="text/css"/>
+ <style>
+ h2 {margin-top: 4em}
+ h3 {margin-top: 3em}
+ td {text-align: left}
+ </style>
+</head>
+<body>
+
+<h1>GetSnowflakeIngestStatus</h1>
+
+<h3>Description</h3>
+<p>
+ The GetSnowflakeIngestStatus processor can be used to get the status of a staged file ingested by a Snowflake pipe.
+ To wait until a staged file is fully ingested (copied into the table) you should connect this processor's "retry" relationship to itself.
+ The processor requires an upstream connection that provides the path of the staged file to be checked through the "snowflake.staged.file.path" attribute.
+ See StartSnowflakeIngest processor for details about how to properly set up a flow to ingest staged files.
+ <b>
+ NOTE: Snowflake pipes caches the ingested files' path and never ingests the same file multiple times.
+ This can lead to the processor being stuck in an "infinite loop" with a FlowFile that has the same "snowflake.staged.file.path" as a stage file that has been
+ already ingested by the pipe. It's recommended to set up the retry mechanism in a way that avoids these situations.
Review Comment:
```suggestion
This can cause the processor to enter an "infinite loop" with a FlowFile that has the same "snowflake.staged.file.path" attribute as a stage file that has previously been ingested by the pipe. It is recommended that the retry mechanism be configured to avoid these scenarios.
```
##########
nifi-nar-bundles/nifi-snowflake-bundle/nifi-snowflake-processors/src/main/resources/docs/org.apache.nifi.processors.snowflake.StartSnowflakeIngest/additionalDetails.html:
##########
@@ -0,0 +1,52 @@
+<!--
+ ~ 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.
+ -->
+
+<!DOCTYPE html>
+<html lang="en">
+<head>
+ <meta charset="utf-8"/>
+ <title>StartSnowflakeIngest</title>
+ <link rel="stylesheet" href="../../../../../css/component-usage.css" type="text/css"/>
+ <style>
+ h2 {margin-top: 4em}
+ h3 {margin-top: 3em}
+ td {text-align: left}
+ </style>
+</head>
+<body>
+
+<h1>StartSnowflakeIngest</h1>
+
+<h3>Description</h3>
+<p>
+ The StartSnowflakeIngest processor triggers a Snowflake pipe ingestion for a staged file. The pipe has to be set up in your Snowflake account.
+ The processor requires an upstream connection that provides the path of the file to be ingested in the stage through the "snowflake.staged.file.path" attribute.
+ This attribute is automatically filled in by the PutSnowflakeInternalStage when you are using an internal stage.
Review Comment:
```suggestion
This attribute is automatically filled in by the PutSnowflakeInternalStage when using an internal stage.
```
##########
nifi-nar-bundles/nifi-snowflake-bundle/nifi-snowflake-processors/src/main/java/org/apache/nifi/processors/snowflake/common/Attributes.java:
##########
@@ -0,0 +1,23 @@
+/*
+ * 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.snowflake.common;
+
+public class Attributes {
Review Comment:
Could you please make this class final and add a private constructor to prevent it from being extended and instantiated?
##########
nifi-nar-bundles/nifi-snowflake-bundle/nifi-snowflake-processors/src/main/resources/docs/org.apache.nifi.processors.snowflake.StartSnowflakeIngest/additionalDetails.html:
##########
@@ -0,0 +1,52 @@
+<!--
+ ~ 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.
+ -->
+
+<!DOCTYPE html>
+<html lang="en">
+<head>
+ <meta charset="utf-8"/>
+ <title>StartSnowflakeIngest</title>
+ <link rel="stylesheet" href="../../../../../css/component-usage.css" type="text/css"/>
+ <style>
+ h2 {margin-top: 4em}
+ h3 {margin-top: 3em}
+ td {text-align: left}
+ </style>
+</head>
+<body>
+
+<h1>StartSnowflakeIngest</h1>
+
+<h3>Description</h3>
+<p>
+ The StartSnowflakeIngest processor triggers a Snowflake pipe ingestion for a staged file. The pipe has to be set up in your Snowflake account.
+ The processor requires an upstream connection that provides the path of the file to be ingested in the stage through the "snowflake.staged.file.path" attribute.
+ This attribute is automatically filled in by the PutSnowflakeInternalStage when you are using an internal stage.
+ In case your pipe copies data from an external stage, you need to provide this attribute yourself (e.g. with an UpdateAttribute processor).
Review Comment:
```suggestion
In case a pipe copies data from an external stage, the attribute shall be manually provided (e.g. with an UpdateAttribute processor).
```
##########
nifi-nar-bundles/nifi-snowflake-bundle/nifi-snowflake-processors/src/main/resources/docs/org.apache.nifi.processors.snowflake.StartSnowflakeIngest/additionalDetails.html:
##########
@@ -0,0 +1,52 @@
+<!--
+ ~ 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.
+ -->
+
+<!DOCTYPE html>
+<html lang="en">
+<head>
+ <meta charset="utf-8"/>
+ <title>StartSnowflakeIngest</title>
+ <link rel="stylesheet" href="../../../../../css/component-usage.css" type="text/css"/>
+ <style>
+ h2 {margin-top: 4em}
+ h3 {margin-top: 3em}
+ td {text-align: left}
+ </style>
+</head>
+<body>
+
+<h1>StartSnowflakeIngest</h1>
+
+<h3>Description</h3>
+<p>
+ The StartSnowflakeIngest processor triggers a Snowflake pipe ingestion for a staged file. The pipe has to be set up in your Snowflake account.
+ The processor requires an upstream connection that provides the path of the file to be ingested in the stage through the "snowflake.staged.file.path" attribute.
+ This attribute is automatically filled in by the PutSnowflakeInternalStage when you are using an internal stage.
+ In case your pipe copies data from an external stage, you need to provide this attribute yourself (e.g. with an UpdateAttribute processor).
+ <b>
+ NOTE: A Snowflake pipe doesn't ingest a file synchronously, so this processor transfers a FlowFile to the "success" relationship when the file is marked for ingestion.
+ You need to wait for the actual result of the ingestion through other measures. E.g. you can connect this processor to a downstream GetSnowflakeIngestStatus processor to wait for the results.
Review Comment:
```suggestion
NOTE: Since Snowflake pipes ingest files asynchronously, this processor transfers FlowFiles to the "success" relationship when they're marked for ingestion.
In order to wait for the actual result of the ingestion, the processor may be connected to a downstream GetSnowflakeIngestStatus processor.
```
##########
nifi-nar-bundles/nifi-snowflake-bundle/nifi-snowflake-processors/src/main/resources/docs/org.apache.nifi.processors.snowflake.PutSnowflakeInternalStage/additionalDetails.html:
##########
@@ -0,0 +1,42 @@
+<!--
+ ~ 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.
+ -->
+
+<!DOCTYPE html>
+<html lang="en">
+<head>
+ <meta charset="utf-8"/>
+ <title>PutSnowflakeInternalStage</title>
+ <link rel="stylesheet" href="../../../../../css/component-usage.css" type="text/css"/>
+ <style>
+ h2 {margin-top: 4em}
+ h3 {margin-top: 3em}
+ td {text-align: left}
+ </style>
+</head>
+<body>
+
+<h1>PutSnowflakeInternalStage</h1>
+
+<h3>Description</h3>
+<p>
+ The PutSnowflakeInternalStage processor can upload a file to a Snowflake internal stage. This stage needs to be set up in your Snowflake account.
Review Comment:
We may provide a link from Snowflake docs for setting up a stage. Can it be any type of internal stage?
https://docs.snowflake.com/en/user-guide/data-load-local-file-system-create-stage.html
##########
nifi-nar-bundles/nifi-snowflake-bundle/nifi-snowflake-processors/src/main/java/org/apache/nifi/processors/snowflake/PutSnowflakeInternalStage.java:
##########
@@ -0,0 +1,147 @@
+/*
+ * 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.snowflake;
+
+import static org.apache.nifi.processors.snowflake.common.Attributes.ATTRIBUTE_STAGED_FILE_PATH;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.sql.SQLException;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+import org.apache.nifi.annotation.behavior.InputRequirement;
+import org.apache.nifi.annotation.behavior.InputRequirement.Requirement;
+import org.apache.nifi.annotation.behavior.ReadsAttribute;
+import org.apache.nifi.annotation.behavior.ReadsAttributes;
+import org.apache.nifi.annotation.behavior.WritesAttribute;
+import org.apache.nifi.annotation.behavior.WritesAttributes;
+import org.apache.nifi.annotation.documentation.CapabilityDescription;
+import org.apache.nifi.annotation.documentation.SeeAlso;
+import org.apache.nifi.annotation.documentation.Tags;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.expression.ExpressionLanguageScope;
+import org.apache.nifi.flowfile.FlowFile;
+import org.apache.nifi.flowfile.attributes.CoreAttributes;
+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.exception.ProcessException;
+import org.apache.nifi.processor.util.StandardValidators;
+
+@InputRequirement(Requirement.INPUT_REQUIRED)
+@ReadsAttributes({
+ @ReadsAttribute(attribute = "filename", description = "The name of the staged file in the internal stage"),
+ @ReadsAttribute(attribute = "path", description = "The relative path to the staged file in the internal stage")
+})
+@WritesAttributes({
+ @WritesAttribute(attribute = ATTRIBUTE_STAGED_FILE_PATH,
+ description = "The path to the file in the internal stage")
+})
+@Tags({"snowflake", "jdbc", "database", "connection"})
+@CapabilityDescription("Put files into a Snowflake internal stage. The internal stage must be created in the Snowflake account beforehand."
+ + " This processor can be connected to an StartSnowflakeIngest processor to ingest the file in the internal stage")
+@SeeAlso({StartSnowflakeIngest.class, GetSnowflakeIngestStatus.class})
+public class PutSnowflakeInternalStage extends AbstractProcessor {
+
+ static final PropertyDescriptor SNOWFLAKE_CONNECTION_PROVIDER = new PropertyDescriptor.Builder()
+ .name("snowflake-connection-provider")
+ .displayName("Snowflake Connection Provider")
+ .description("Specifies the Controller Service to use for creating SQL connections to Snowflake.")
+ .identifiesControllerService(SnowflakeConnectionProviderService.class)
+ .required(true)
+ .build();
+
+ static final PropertyDescriptor INTERNAL_STAGE_NAME = new PropertyDescriptor.Builder()
+ .name("internal-stage-name")
+ .displayName("Internal Stage Name")
+ .description("The name of the internal stage in the Snowflake account to put files into.")
+ .addValidator(StandardValidators.NON_BLANK_VALIDATOR)
+ .expressionLanguageSupported(ExpressionLanguageScope.VARIABLE_REGISTRY)
+ .required(true)
+ .build();
+
+ static final Relationship REL_SUCCESS = new Relationship.Builder()
+ .name("success")
+ .description("For FlowFiles of successful PUT operation")
+ .build();
+
+ static final Relationship REL_FAILURE = new Relationship.Builder()
+ .name("failure")
+ .description("For FlowFiles of failed PUT operation")
+ .build();
+
+ static final List<PropertyDescriptor> PROPERTIES = Collections.unmodifiableList(Arrays.asList(
+ SNOWFLAKE_CONNECTION_PROVIDER,
+ INTERNAL_STAGE_NAME
+ ));
+
+ static final Set<Relationship> RELATIONSHIPS;
+
+ static {
+ final Set<Relationship> relationships = new HashSet<>();
+ relationships.add(REL_SUCCESS);
+ relationships.add(REL_FAILURE);
+ RELATIONSHIPS = Collections.unmodifiableSet(relationships);
+ }
+
+ @Override
+ protected List<PropertyDescriptor> getSupportedPropertyDescriptors() {
+ return PROPERTIES;
+ }
+
+ @Override
+ public Set<Relationship> getRelationships() {
+ return RELATIONSHIPS;
+ }
+
+ @Override
+ public void onTrigger(ProcessContext context, ProcessSession session) throws ProcessException {
+ final String internalStageName = context.getProperty(INTERNAL_STAGE_NAME)
+ .evaluateAttributeExpressions()
+ .getValue();
+ final SnowflakeConnectionProviderService connectionProviderService =
+ context.getProperty(SNOWFLAKE_CONNECTION_PROVIDER)
+ .asControllerService(SnowflakeConnectionProviderService.class);
+
+ FlowFile flowFile = session.get();
+ final String fileName = flowFile.getAttribute(CoreAttributes.FILENAME.key());
+ final String relativePath = flowFile.getAttribute(CoreAttributes.PATH.key());
+ final String stageRelativePath = "./".equals(relativePath)
+ ? ""
+ : relativePath;
+ try (final InputStream inputStream = session.read(flowFile);
+ final SnowflakeConnectionWrapper snowflakeConnection = connectionProviderService.getSnowflakeConnection()) {
+ snowflakeConnection.unwrap()
+ .uploadStream(internalStageName, stageRelativePath, inputStream, fileName, false);
+ } catch (SQLException e) {
+ getLogger().error("Failed to upload flow content to internal Snowflake stage", e);
Review Comment:
```suggestion
getLogger().error("Failed to upload flowfile content to internal Snowflake stage", e);
```
##########
nifi-nar-bundles/nifi-snowflake-bundle/nifi-snowflake-processors/src/main/resources/docs/org.apache.nifi.processors.snowflake.PutSnowflakeInternalStage/additionalDetails.html:
##########
@@ -0,0 +1,42 @@
+<!--
+ ~ 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.
+ -->
+
+<!DOCTYPE html>
+<html lang="en">
+<head>
+ <meta charset="utf-8"/>
+ <title>PutSnowflakeInternalStage</title>
+ <link rel="stylesheet" href="../../../../../css/component-usage.css" type="text/css"/>
+ <style>
+ h2 {margin-top: 4em}
+ h3 {margin-top: 3em}
+ td {text-align: left}
+ </style>
+</head>
+<body>
+
+<h1>PutSnowflakeInternalStage</h1>
+
+<h3>Description</h3>
+<p>
+ The PutSnowflakeInternalStage processor can upload a file to a Snowflake internal stage. This stage needs to be set up in your Snowflake account.
+ The processor requires an upstream connection and the incoming FlowFiles' content will be uploaded to the stage. The attributes "filename" and "path" are used to provide a prefix and file name for your file in the stage.
+ While the processor can be used by itself, it's usually recommended to connect it to a StartSnowflakeIngest processor to put the uploaded file into your Snowflake table via a pipe.
Review Comment:
```suggestion
While the processor may be used separately, it's recommended to connect it to a StartSnowflakeIngest processor so that the uploaded file can be piped into your Snowflake table.
```
##########
nifi-nar-bundles/nifi-snowflake-bundle/nifi-snowflake-processors/src/main/java/org/apache/nifi/processors/snowflake/StartSnowflakeIngest.java:
##########
@@ -0,0 +1,124 @@
+/*
+ * 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.snowflake;
+
+import static org.apache.nifi.processors.snowflake.common.Attributes.ATTRIBUTE_STAGED_FILE_PATH;
+
+import java.io.IOException;
+import java.net.URISyntaxException;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+import net.snowflake.ingest.SimpleIngestManager;
+import net.snowflake.ingest.connection.IngestResponseException;
+import net.snowflake.ingest.utils.StagedFileWrapper;
+import org.apache.nifi.annotation.behavior.InputRequirement;
+import org.apache.nifi.annotation.behavior.InputRequirement.Requirement;
+import org.apache.nifi.annotation.behavior.ReadsAttribute;
+import org.apache.nifi.annotation.behavior.ReadsAttributes;
+import org.apache.nifi.annotation.documentation.CapabilityDescription;
+import org.apache.nifi.annotation.documentation.SeeAlso;
+import org.apache.nifi.annotation.documentation.Tags;
+import org.apache.nifi.components.PropertyDescriptor;
+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.exception.ProcessException;
+
+@InputRequirement(Requirement.INPUT_REQUIRED)
+@ReadsAttributes({
+ @ReadsAttribute(attribute = ATTRIBUTE_STAGED_FILE_PATH, description = "The path to the file in the stage")
+})
+@Tags({"snowflake", "snowpipe", "ingest"})
+@CapabilityDescription("Ingest files in a Snowflake stage. The stage must be created in the Snowflake account beforehand."
Review Comment:
```suggestion
@CapabilityDescription("Ingests files in a Snowflake stage. The stage must be created in the Snowflake account beforehand."
```
##########
nifi-nar-bundles/nifi-snowflake-bundle/nifi-snowflake-processors/src/main/java/org/apache/nifi/processors/snowflake/PutSnowflakeInternalStage.java:
##########
@@ -0,0 +1,147 @@
+/*
+ * 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.snowflake;
+
+import static org.apache.nifi.processors.snowflake.common.Attributes.ATTRIBUTE_STAGED_FILE_PATH;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.sql.SQLException;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+import org.apache.nifi.annotation.behavior.InputRequirement;
+import org.apache.nifi.annotation.behavior.InputRequirement.Requirement;
+import org.apache.nifi.annotation.behavior.ReadsAttribute;
+import org.apache.nifi.annotation.behavior.ReadsAttributes;
+import org.apache.nifi.annotation.behavior.WritesAttribute;
+import org.apache.nifi.annotation.behavior.WritesAttributes;
+import org.apache.nifi.annotation.documentation.CapabilityDescription;
+import org.apache.nifi.annotation.documentation.SeeAlso;
+import org.apache.nifi.annotation.documentation.Tags;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.expression.ExpressionLanguageScope;
+import org.apache.nifi.flowfile.FlowFile;
+import org.apache.nifi.flowfile.attributes.CoreAttributes;
+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.exception.ProcessException;
+import org.apache.nifi.processor.util.StandardValidators;
+
+@InputRequirement(Requirement.INPUT_REQUIRED)
+@ReadsAttributes({
+ @ReadsAttribute(attribute = "filename", description = "The name of the staged file in the internal stage"),
+ @ReadsAttribute(attribute = "path", description = "The relative path to the staged file in the internal stage")
+})
+@WritesAttributes({
+ @WritesAttribute(attribute = ATTRIBUTE_STAGED_FILE_PATH,
+ description = "The path to the file in the internal stage")
+})
+@Tags({"snowflake", "jdbc", "database", "connection"})
+@CapabilityDescription("Put files into a Snowflake internal stage. The internal stage must be created in the Snowflake account beforehand."
+ + " This processor can be connected to an StartSnowflakeIngest processor to ingest the file in the internal stage")
+@SeeAlso({StartSnowflakeIngest.class, GetSnowflakeIngestStatus.class})
+public class PutSnowflakeInternalStage extends AbstractProcessor {
+
+ static final PropertyDescriptor SNOWFLAKE_CONNECTION_PROVIDER = new PropertyDescriptor.Builder()
+ .name("snowflake-connection-provider")
+ .displayName("Snowflake Connection Provider")
+ .description("Specifies the Controller Service to use for creating SQL connections to Snowflake.")
+ .identifiesControllerService(SnowflakeConnectionProviderService.class)
+ .required(true)
+ .build();
+
+ static final PropertyDescriptor INTERNAL_STAGE_NAME = new PropertyDescriptor.Builder()
+ .name("internal-stage-name")
+ .displayName("Internal Stage Name")
+ .description("The name of the internal stage in the Snowflake account to put files into.")
+ .addValidator(StandardValidators.NON_BLANK_VALIDATOR)
+ .expressionLanguageSupported(ExpressionLanguageScope.VARIABLE_REGISTRY)
+ .required(true)
+ .build();
+
+ static final Relationship REL_SUCCESS = new Relationship.Builder()
+ .name("success")
+ .description("For FlowFiles of successful PUT operation")
+ .build();
+
+ static final Relationship REL_FAILURE = new Relationship.Builder()
+ .name("failure")
+ .description("For FlowFiles of failed PUT operation")
+ .build();
+
+ static final List<PropertyDescriptor> PROPERTIES = Collections.unmodifiableList(Arrays.asList(
+ SNOWFLAKE_CONNECTION_PROVIDER,
+ INTERNAL_STAGE_NAME
+ ));
+
+ static final Set<Relationship> RELATIONSHIPS;
+
+ static {
+ final Set<Relationship> relationships = new HashSet<>();
+ relationships.add(REL_SUCCESS);
+ relationships.add(REL_FAILURE);
+ RELATIONSHIPS = Collections.unmodifiableSet(relationships);
+ }
+
+ @Override
+ protected List<PropertyDescriptor> getSupportedPropertyDescriptors() {
+ return PROPERTIES;
+ }
+
+ @Override
+ public Set<Relationship> getRelationships() {
+ return RELATIONSHIPS;
+ }
+
+ @Override
+ public void onTrigger(ProcessContext context, ProcessSession session) throws ProcessException {
+ final String internalStageName = context.getProperty(INTERNAL_STAGE_NAME)
+ .evaluateAttributeExpressions()
+ .getValue();
+ final SnowflakeConnectionProviderService connectionProviderService =
+ context.getProperty(SNOWFLAKE_CONNECTION_PROVIDER)
+ .asControllerService(SnowflakeConnectionProviderService.class);
+
+ FlowFile flowFile = session.get();
Review Comment:
I see this pattern for checking the incoming flowfile all around NiFi processors. It may worth considering.
`if (flowFile == null) {
return;
}`
##########
nifi-nar-bundles/nifi-snowflake-bundle/nifi-snowflake-processors/src/main/java/org/apache/nifi/processors/snowflake/GetSnowflakeIngestStatus.java:
##########
@@ -0,0 +1,208 @@
+/*
+ * 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.snowflake;
+
+import static org.apache.nifi.processors.snowflake.common.Attributes.ATTRIBUTE_STAGED_FILE_PATH;
+
+import java.io.IOException;
+import java.net.URISyntaxException;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+import net.snowflake.ingest.SimpleIngestManager;
+import net.snowflake.ingest.connection.HistoryResponse;
+import net.snowflake.ingest.connection.HistoryResponse.FileEntry;
+import net.snowflake.ingest.connection.IngestResponseException;
+import org.apache.nifi.annotation.behavior.InputRequirement;
+import org.apache.nifi.annotation.behavior.InputRequirement.Requirement;
+import org.apache.nifi.annotation.behavior.ReadsAttribute;
+import org.apache.nifi.annotation.behavior.ReadsAttributes;
+import org.apache.nifi.annotation.behavior.Stateful;
+import org.apache.nifi.annotation.configuration.DefaultSettings;
+import org.apache.nifi.annotation.documentation.CapabilityDescription;
+import org.apache.nifi.annotation.documentation.SeeAlso;
+import org.apache.nifi.annotation.documentation.Tags;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.components.state.Scope;
+import org.apache.nifi.components.state.StateMap;
+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.exception.ProcessException;
+
+@InputRequirement(Requirement.INPUT_REQUIRED)
+@Stateful(scopes = Scope.CLUSTER,
+ description = "The 'begin mark' from the response of a history request is stored to keep track of already requested history time range.")
+@DefaultSettings(penaltyDuration = "5 sec")
+@ReadsAttributes({
+ @ReadsAttribute(attribute = ATTRIBUTE_STAGED_FILE_PATH, description = "The path to the file in the stage")
+})
+@Tags({"snowflake", "snowpipe", "ingest", "history"})
+@CapabilityDescription("Waits until a file in a Snowflake stage is ingested. The stage must be created in the Snowflake account beforehand."
+ + " This processor is usually connected to an upstream StartSnowflakeIngest processor to make sure that the file is ingested")
+@SeeAlso({StartSnowflakeIngest.class, PutSnowflakeInternalStage.class})
+public class GetSnowflakeIngestStatus extends AbstractProcessor {
+
+ static final PropertyDescriptor INGEST_MANAGER_PROVIDER = new PropertyDescriptor.Builder()
+ .name("ingest-manager-provider")
+ .displayName("Ingest Manager Provider")
+ .description("Specifies the Controller Service to use for ingesting Snowflake staged files.")
+ .identifiesControllerService(SnowflakeIngestManagerProviderService.class)
+ .required(true)
+ .build();
+
+ static final Relationship REL_SUCCESS = new Relationship.Builder()
+ .name("success")
+ .description("For FlowFiles of successful ingestion")
+ .build();
+
+ static final Relationship REL_FAILURE = new Relationship.Builder()
+ .name("failure")
+ .description("For FlowFiles of failed ingestion")
+ .build();
+
+ static final Relationship REL_RETRY = new Relationship.Builder()
+ .name("retry")
+ .description("For FlowFiles whose file is still not ingested. These FlowFiles should be routed back to this processor to try again later")
+ .build();
+
+ static final List<PropertyDescriptor> PROPERTIES = Collections.singletonList(
+ INGEST_MANAGER_PROVIDER
+ );
+
+ static final Set<Relationship> RELATIONSHIPS;
+
+ static {
+ final Set<Relationship> relationships = new HashSet<>();
+ relationships.add(REL_SUCCESS);
+ relationships.add(REL_RETRY);
+ relationships.add(REL_FAILURE);
+ RELATIONSHIPS = Collections.unmodifiableSet(relationships);
+ }
Review Comment:
```suggestion
private static final Set<Relationship> RELATIONSHIPS = Collections.unmodifiableSet(new HashSet<>(Arrays.asList(
REL_SUCCESS,
REL_RETRY,
REL_FAILURE
)));
```
##########
nifi-nar-bundles/nifi-snowflake-bundle/nifi-snowflake-processors/src/main/java/org/apache/nifi/processors/snowflake/PutSnowflakeInternalStage.java:
##########
@@ -0,0 +1,147 @@
+/*
+ * 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.snowflake;
+
+import static org.apache.nifi.processors.snowflake.common.Attributes.ATTRIBUTE_STAGED_FILE_PATH;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.sql.SQLException;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+import org.apache.nifi.annotation.behavior.InputRequirement;
+import org.apache.nifi.annotation.behavior.InputRequirement.Requirement;
+import org.apache.nifi.annotation.behavior.ReadsAttribute;
+import org.apache.nifi.annotation.behavior.ReadsAttributes;
+import org.apache.nifi.annotation.behavior.WritesAttribute;
+import org.apache.nifi.annotation.behavior.WritesAttributes;
+import org.apache.nifi.annotation.documentation.CapabilityDescription;
+import org.apache.nifi.annotation.documentation.SeeAlso;
+import org.apache.nifi.annotation.documentation.Tags;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.expression.ExpressionLanguageScope;
+import org.apache.nifi.flowfile.FlowFile;
+import org.apache.nifi.flowfile.attributes.CoreAttributes;
+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.exception.ProcessException;
+import org.apache.nifi.processor.util.StandardValidators;
+
+@InputRequirement(Requirement.INPUT_REQUIRED)
+@ReadsAttributes({
+ @ReadsAttribute(attribute = "filename", description = "The name of the staged file in the internal stage"),
+ @ReadsAttribute(attribute = "path", description = "The relative path to the staged file in the internal stage")
+})
+@WritesAttributes({
+ @WritesAttribute(attribute = ATTRIBUTE_STAGED_FILE_PATH,
+ description = "The path to the file in the internal stage")
+})
+@Tags({"snowflake", "jdbc", "database", "connection"})
+@CapabilityDescription("Put files into a Snowflake internal stage. The internal stage must be created in the Snowflake account beforehand."
+ + " This processor can be connected to an StartSnowflakeIngest processor to ingest the file in the internal stage")
+@SeeAlso({StartSnowflakeIngest.class, GetSnowflakeIngestStatus.class})
+public class PutSnowflakeInternalStage extends AbstractProcessor {
+
+ static final PropertyDescriptor SNOWFLAKE_CONNECTION_PROVIDER = new PropertyDescriptor.Builder()
+ .name("snowflake-connection-provider")
+ .displayName("Snowflake Connection Provider")
+ .description("Specifies the Controller Service to use for creating SQL connections to Snowflake.")
+ .identifiesControllerService(SnowflakeConnectionProviderService.class)
+ .required(true)
+ .build();
+
+ static final PropertyDescriptor INTERNAL_STAGE_NAME = new PropertyDescriptor.Builder()
+ .name("internal-stage-name")
+ .displayName("Internal Stage Name")
+ .description("The name of the internal stage in the Snowflake account to put files into.")
+ .addValidator(StandardValidators.NON_BLANK_VALIDATOR)
+ .expressionLanguageSupported(ExpressionLanguageScope.VARIABLE_REGISTRY)
+ .required(true)
+ .build();
+
+ static final Relationship REL_SUCCESS = new Relationship.Builder()
+ .name("success")
+ .description("For FlowFiles of successful PUT operation")
+ .build();
+
+ static final Relationship REL_FAILURE = new Relationship.Builder()
+ .name("failure")
+ .description("For FlowFiles of failed PUT operation")
+ .build();
+
+ static final List<PropertyDescriptor> PROPERTIES = Collections.unmodifiableList(Arrays.asList(
+ SNOWFLAKE_CONNECTION_PROVIDER,
+ INTERNAL_STAGE_NAME
+ ));
+
+ static final Set<Relationship> RELATIONSHIPS;
+
+ static {
+ final Set<Relationship> relationships = new HashSet<>();
+ relationships.add(REL_SUCCESS);
+ relationships.add(REL_FAILURE);
+ RELATIONSHIPS = Collections.unmodifiableSet(relationships);
+ }
+
+ @Override
+ protected List<PropertyDescriptor> getSupportedPropertyDescriptors() {
+ return PROPERTIES;
+ }
+
+ @Override
+ public Set<Relationship> getRelationships() {
+ return RELATIONSHIPS;
+ }
+
+ @Override
+ public void onTrigger(ProcessContext context, ProcessSession session) throws ProcessException {
+ final String internalStageName = context.getProperty(INTERNAL_STAGE_NAME)
+ .evaluateAttributeExpressions()
+ .getValue();
+ final SnowflakeConnectionProviderService connectionProviderService =
+ context.getProperty(SNOWFLAKE_CONNECTION_PROVIDER)
+ .asControllerService(SnowflakeConnectionProviderService.class);
+
+ FlowFile flowFile = session.get();
+ final String fileName = flowFile.getAttribute(CoreAttributes.FILENAME.key());
+ final String relativePath = flowFile.getAttribute(CoreAttributes.PATH.key());
+ final String stageRelativePath = "./".equals(relativePath)
+ ? ""
+ : relativePath;
+ try (final InputStream inputStream = session.read(flowFile);
+ final SnowflakeConnectionWrapper snowflakeConnection = connectionProviderService.getSnowflakeConnection()) {
+ snowflakeConnection.unwrap()
+ .uploadStream(internalStageName, stageRelativePath, inputStream, fileName, false);
+ } catch (SQLException e) {
+ getLogger().error("Failed to upload flow content to internal Snowflake stage", e);
Review Comment:
Can we include the internal stage name and filename to the error message?
##########
nifi-nar-bundles/nifi-snowflake-bundle/nifi-snowflake-services/src/main/java/org/apache/nifi/snowflake/service/SnowflakeComputingConnectionPool.java:
##########
@@ -55,34 +64,107 @@
description = "Snowflake JDBC driver property name prefixed with 'SENSITIVE.' handled as a sensitive property.")
})
@RequiresInstanceClassLoading
-public class SnowflakeComputingConnectionPool extends AbstractDBCPConnectionPool implements DBCPService {
+public class SnowflakeComputingConnectionPool extends AbstractDBCPConnectionPool implements SnowflakeConnectionProviderService, DBCPService {
Review Comment:
AbstractDBCPConnectionPool is abstract and already implements DBCPService
##########
nifi-nar-bundles/nifi-snowflake-bundle/nifi-snowflake-services/src/main/java/org/apache/nifi/snowflake/service/util/ConnectionUrlFormat.java:
##########
@@ -0,0 +1,112 @@
+/*
+ * 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.snowflake.service.util;
+
+import static org.apache.nifi.snowflake.service.SnowflakeComputingConnectionPool.SNOWFLAKE_ACCOUNT_LOCATOR;
+import static org.apache.nifi.snowflake.service.SnowflakeComputingConnectionPool.SNOWFLAKE_ACCOUNT_NAME;
+import static org.apache.nifi.snowflake.service.SnowflakeComputingConnectionPool.SNOWFLAKE_CLOUD_REGION;
+import static org.apache.nifi.snowflake.service.SnowflakeComputingConnectionPool.SNOWFLAKE_CLOUD_TYPE;
+import static org.apache.nifi.snowflake.service.SnowflakeComputingConnectionPool.SNOWFLAKE_ORGANIZATION_NAME;
+import static org.apache.nifi.snowflake.service.SnowflakeComputingConnectionPool.SNOWFLAKE_URL;
+
+import java.util.stream.Stream;
+import org.apache.nifi.components.DescribedValue;
+import org.apache.nifi.controller.ConfigurationContext;
+
+public enum ConnectionUrlFormat implements DescribedValue {
+ FULL_URL("full-url", "Full URL", "Provide connection URL in a single property") {
+ @Override
+ public String buildConnectionUrl(final ConfigurationContext context) {
+ String snowflakeUrl = context.getProperty(SNOWFLAKE_URL).evaluateAttributeExpressions().getValue();
+ if (!snowflakeUrl.startsWith("jdbc:snowflake")) {
Review Comment:
Can "jdbc:snowflake" be extracted to Attributes as SNOWFLAKE_JDBC_PREFIX to avoid duplication?
##########
nifi-nar-bundles/nifi-snowflake-bundle/nifi-snowflake-services/src/main/java/org/apache/nifi/snowflake/service/util/CommonProperties.java:
##########
@@ -0,0 +1,68 @@
+/*
+ * 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.snowflake.service.util;
+
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.expression.ExpressionLanguageScope;
+import org.apache.nifi.processor.util.StandardValidators;
+
+public class CommonProperties {
Review Comment:
Would you please make the class final and add a private constructor to it?
##########
nifi-nar-bundles/nifi-snowflake-bundle/nifi-snowflake-services/src/main/java/org/apache/nifi/snowflake/service/SnowflakeComputingConnectionPool.java:
##########
@@ -55,34 +64,107 @@
description = "Snowflake JDBC driver property name prefixed with 'SENSITIVE.' handled as a sensitive property.")
})
@RequiresInstanceClassLoading
-public class SnowflakeComputingConnectionPool extends AbstractDBCPConnectionPool implements DBCPService {
+public class SnowflakeComputingConnectionPool extends AbstractDBCPConnectionPool implements SnowflakeConnectionProviderService, DBCPService {
+
+ public static final PropertyDescriptor CONNECTION_URL_FORMAT = new PropertyDescriptor.Builder()
+ .name("connection-url-format")
+ .displayName("Connection URL Format")
+ .description("The format of the connection URL.")
+ .allowableValues(ConnectionUrlFormat.class)
+ .required(true)
+ .defaultValue(ConnectionUrlFormat.FULL_URL.getValue())
+ .build();
public static final PropertyDescriptor SNOWFLAKE_URL = new PropertyDescriptor.Builder()
- .fromPropertyDescriptor(AbstractDBCPConnectionPool.DATABASE_URL)
- .displayName("Snowflake URL")
- .description("Example connection string: jdbc:snowflake://[account].[region].snowflakecomputing.com/?[connection_params]" +
- " The connection parameters can include db=DATABASE_NAME to avoid using qualified table names such as DATABASE_NAME.PUBLIC.TABLE_NAME")
- .build();
+ .fromPropertyDescriptor(AbstractDBCPConnectionPool.DATABASE_URL)
+ .displayName("Snowflake URL")
+ .description("Example connection string: jdbc:snowflake://[account].[region].snowflakecomputing.com/?[connection_params]" +
+ " The connection parameters can include db=DATABASE_NAME to avoid using qualified table names such as DATABASE_NAME.PUBLIC.TABLE_NAME")
+ .required(true)
+ .dependsOn(CONNECTION_URL_FORMAT, ConnectionUrlFormat.FULL_URL)
+ .build();
+
+ public static final PropertyDescriptor SNOWFLAKE_ACCOUNT_LOCATOR = new PropertyDescriptor.Builder()
+ .fromPropertyDescriptor(CommonProperties.ACCOUNT_LOCATOR)
+ .dependsOn(CONNECTION_URL_FORMAT, ConnectionUrlFormat.ACCOUNT_LOCATOR)
+ .build();
+
+ public static final PropertyDescriptor SNOWFLAKE_CLOUD_REGION = new PropertyDescriptor.Builder()
+ .fromPropertyDescriptor(CommonProperties.CLOUD_REGION)
+ .dependsOn(CONNECTION_URL_FORMAT, ConnectionUrlFormat.ACCOUNT_LOCATOR)
+ .build();
+
+ public static final PropertyDescriptor SNOWFLAKE_CLOUD_TYPE = new PropertyDescriptor.Builder()
+ .fromPropertyDescriptor(CommonProperties.CLOUD_TYPE)
+ .dependsOn(CONNECTION_URL_FORMAT, ConnectionUrlFormat.ACCOUNT_LOCATOR)
+ .build();
+
+ public static final PropertyDescriptor SNOWFLAKE_ORGANIZATION_NAME = new PropertyDescriptor.Builder()
+ .fromPropertyDescriptor(CommonProperties.ORGANIZATION_NAME)
+ .dependsOn(CONNECTION_URL_FORMAT, ConnectionUrlFormat.ACCOUNT_NAME)
+ .build();
+
+ public static final PropertyDescriptor SNOWFLAKE_ACCOUNT_NAME = new PropertyDescriptor.Builder()
+ .fromPropertyDescriptor(CommonProperties.ACCOUNT_NAME)
+ .dependsOn(CONNECTION_URL_FORMAT, ConnectionUrlFormat.ACCOUNT_NAME)
+ .build();
public static final PropertyDescriptor SNOWFLAKE_USER = new PropertyDescriptor.Builder()
- .fromPropertyDescriptor(AbstractDBCPConnectionPool.DB_USER)
- .displayName("Snowflake User")
- .description("The Snowflake user name")
- .build();
+ .fromPropertyDescriptor(AbstractDBCPConnectionPool.DB_USER)
+ .displayName("Snowflake User")
+ .description("The Snowflake user name.")
+ .required(true)
+ .build();
public static final PropertyDescriptor SNOWFLAKE_PASSWORD = new PropertyDescriptor.Builder()
- .fromPropertyDescriptor(AbstractDBCPConnectionPool.DB_PASSWORD)
- .displayName("Snowflake Password")
- .description("The password for the Snowflake user")
- .build();
+ .fromPropertyDescriptor(AbstractDBCPConnectionPool.DB_PASSWORD)
+ .displayName("Snowflake Password")
+ .description("The password for the Snowflake user.")
+ .required(true)
+ .build();
+
+ public static final PropertyDescriptor SNOWFLAKE_DATABASE = new PropertyDescriptor.Builder()
+ .name("database")
+ .displayName("Database")
+ .description("The database to use by default. The same as passing 'db=DATABASE_NAME' to the connection string.")
+ .addValidator(StandardValidators.NON_BLANK_VALIDATOR)
+ .expressionLanguageSupported(ExpressionLanguageScope.VARIABLE_REGISTRY)
+ .build();
+
+ public static final PropertyDescriptor SNOWFLAKE_SCHEMA = new PropertyDescriptor.Builder()
+ .name("schema")
+ .displayName("Schema")
+ .description("The schema to use by default. The same as passing 'schema=SCHEMA' to the connection string.")
+ .addValidator(StandardValidators.NON_BLANK_VALIDATOR)
+ .expressionLanguageSupported(ExpressionLanguageScope.VARIABLE_REGISTRY)
+ .defaultValue("PUBLIC")
+ .build();
+
+ public static final PropertyDescriptor SNOWFLAKE_WAREHOUSE = new PropertyDescriptor.Builder()
+ .name("warehouse")
+ .displayName("Warehouse")
+ .description("The warehouse to use by default. The same as passing 'warehouse=WAREHOUSE' to the connection string.")
+ .addValidator(StandardValidators.NON_BLANK_VALIDATOR)
+ .expressionLanguageSupported(ExpressionLanguageScope.VARIABLE_REGISTRY)
+ .build();
private static final List<PropertyDescriptor> PROPERTIES;
static {
final List<PropertyDescriptor> props = new ArrayList<>();
+ props.add(CONNECTION_URL_FORMAT);
props.add(SNOWFLAKE_URL);
+ props.add(SNOWFLAKE_ACCOUNT_LOCATOR);
+ props.add(SNOWFLAKE_CLOUD_REGION);
+ props.add(SNOWFLAKE_CLOUD_TYPE);
+ props.add(SNOWFLAKE_ORGANIZATION_NAME);
+ props.add(SNOWFLAKE_ACCOUNT_NAME);
props.add(SNOWFLAKE_USER);
props.add(SNOWFLAKE_PASSWORD);
+ props.add(SNOWFLAKE_DATABASE);
+ props.add(SNOWFLAKE_SCHEMA);
+ props.add(SNOWFLAKE_WAREHOUSE);
+ props.add(ProxyConfigurationService.PROXY_CONFIGURATION_SERVICE);
props.add(VALIDATION_QUERY);
props.add(MAX_WAIT_TIME);
props.add(MAX_TOTAL_CONNECTIONS);
Review Comment:
```suggestion
private static final List<PropertyDescriptor> PROPERTIES = Collections.unmodifiableList(Arrays.asList(
CONNECTION_URL_FORMAT,
SNOWFLAKE_URL,
SNOWFLAKE_ACCOUNT_LOCATOR,
SNOWFLAKE_CLOUD_REGION,
SNOWFLAKE_CLOUD_TYPE,
SNOWFLAKE_ORGANIZATION_NAME,
SNOWFLAKE_ACCOUNT_NAME,
SNOWFLAKE_USER,
SNOWFLAKE_PASSWORD,
SNOWFLAKE_DATABASE,
SNOWFLAKE_SCHEMA,
SNOWFLAKE_WAREHOUSE,
ProxyConfigurationService.PROXY_CONFIGURATION_SERVICE,
VALIDATION_QUERY,
MAX_WAIT_TIME,
MAX_TOTAL_CONNECTIONS,
MIN_IDLE,
MAX_IDLE,
MAX_CONN_LIFETIME,
EVICTION_RUN_PERIOD,
MIN_EVICTABLE_IDLE_TIME,
SOFT_MIN_EVICTABLE_IDLE_TIME
));
```
##########
nifi-nar-bundles/nifi-snowflake-bundle/nifi-snowflake-processors-nar/src/main/resources/META-INF/NOTICE:
##########
@@ -0,0 +1,5 @@
+nifi-snowflake-processors-nar
+Copyright 2015-2020 The Apache Software Foundation
+
+This product includes software developed at
+The Apache Software Foundation (http://www.apache.org/).
Review Comment:
Is this one empty on purpose?
--
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: issues-unsubscribe@nifi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org