You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@nifi.apache.org by tu...@apache.org on 2023/03/03 17:59:26 UTC
[nifi] branch support/nifi-1.x updated (88d2c03b37 -> 1273152d78)
This is an automated email from the ASF dual-hosted git repository.
turcsanyi pushed a change to branch support/nifi-1.x
in repository https://gitbox.apache.org/repos/asf/nifi.git
from 88d2c03b37 NIFI-11245 Corrected reducing Maximum Thread Count while running (#7005)
new 85b36cac60 NIFI-11124: Add hadoop.file.url attribute to HDFS processors
new 20914c1af9 NIFI-11146 Proxy usage in BoxFile processors
new b9a0216ee6 NIFI-11173 Add default values to DeleteAzureBlobStorage_v12
new 9438eb8b86 NIFI-11190: Added provenance events to GetHubSpot and GetShopify
new 7475232bb0 NIFI-11045: Sensitive dynamic property support for parameterized queries in ExecuteSQL and ExecuteSQLRecord
new 1273152d78 NIFI-11158 PutSalesforceObject processor improvements
The 6 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../nifi/util/StandardProcessorTestRunner.java | 13 +++-
.../main/java/org/apache/nifi/util/TestRunner.java | 7 ++
.../azure/storage/DeleteAzureBlobStorage_v12.java | 15 ++++-
.../azure/storage/FetchAzureBlobStorage_v12.java | 4 +-
.../azure/storage/ITListAzureBlobStorage_v12.java | 3 +-
.../nifi-box-bundle/nifi-box-services-api/pom.xml | 2 +-
.../nifi-box-bundle/nifi-box-services/pom.xml | 4 ++
.../JsonConfigBasedBoxClientService.java | 44 +++++++-----
.../java/org/apache/nifi/util/db/JdbcCommon.java | 78 ++++++++++++++--------
.../apache/nifi/util/db/SensitiveValueWrapper.java | 71 ++++++++++----------
.../org/apache/nifi/util/db/TestJdbcCommon.java | 34 +++++++++-
.../processors/hadoop/AbstractHadoopProcessor.java | 1 +
.../processors/hadoop/AbstractFetchHDFSRecord.java | 31 ++++-----
.../processors/hadoop/AbstractPutHDFSRecord.java | 1 +
.../apache/nifi/processors/hadoop/DeleteHDFS.java | 2 +
.../apache/nifi/processors/hadoop/FetchHDFS.java | 9 ++-
.../apache/nifi/processors/hadoop/MoveHDFS.java | 6 +-
.../org/apache/nifi/processors/hadoop/PutHDFS.java | 2 +
.../apache/nifi/processors/hadoop/PutHDFSTest.java | 6 ++
.../nifi/processors/hadoop/TestDeleteHDFS.java | 5 ++
.../nifi/processors/hadoop/TestFetchHDFS.java | 4 ++
.../org/apache/nifi/processors/orc/PutORC.java | 1 +
.../org/apache/nifi/processors/orc/PutORCTest.java | 3 +
.../apache/nifi/processors/hubspot/GetHubSpot.java | 1 +
.../nifi/processors/hubspot/GetHubSpotTest.java | 9 +++
.../nifi/processors/parquet/FetchParquet.java | 3 +-
.../apache/nifi/processors/parquet/PutParquet.java | 1 +
.../nifi/processors/parquet/FetchParquetTest.java | 2 +
.../nifi/processors/parquet/PutParquetTest.java | 36 +++++-----
.../processors/salesforce/PutSalesforceObject.java | 34 +++++++---
.../salesforce/QuerySalesforceObject.java | 8 ++-
.../processors/salesforce/util/RecordExtender.java | 4 +-
.../salesforce/util/SalesforceRestService.java | 15 ++++-
.../salesforce/PutSalesforceObjectIT.java | 67 ++++++++++++++++---
.../salesforce/QuerySalesforceObjectIT.java | 2 +
.../apache/nifi/processors/shopify/GetShopify.java | 3 +
.../shopify/rest/ShopifyRestService.java | 4 ++
.../{GetShopifyIT.java => GetShopifyTest.java} | 12 ++--
.../nifi-standard-processors/pom.xml | 9 +--
.../processors/standard/AbstractExecuteSQL.java | 32 ++++++---
.../nifi/processors/standard/ExecuteSQL.java | 61 ++++++++++++++---
.../nifi/processors/standard/ExecuteSQLRecord.java | 47 ++++++++++++-
42 files changed, 523 insertions(+), 173 deletions(-)
copy nifi-registry/nifi-registry-core/nifi-registry-framework/src/main/java/org/apache/nifi/registry/security/ldap/ReferralStrategy.java => nifi-nar-bundles/nifi-extension-utils/nifi-database-utils/src/main/java/org/apache/nifi/util/db/SensitiveValueWrapper.java (74%)
rename nifi-nar-bundles/nifi-shopify-bundle/nifi-shopify-processors/src/test/java/org/apache/nifi/processors/shopify/{GetShopifyIT.java => GetShopifyTest.java} (92%)
[nifi] 03/06: NIFI-11173 Add default values to DeleteAzureBlobStorage_v12
Posted by tu...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
turcsanyi pushed a commit to branch support/nifi-1.x
in repository https://gitbox.apache.org/repos/asf/nifi.git
commit b9a0216ee6bac1531e9cc19e15003bcc4ece6faf
Author: Nandor Soma Abonyi <ab...@gmail.com>
AuthorDate: Mon Feb 13 17:29:52 2023 +0100
NIFI-11173 Add default values to DeleteAzureBlobStorage_v12
This closes #6945.
Signed-off-by: Peter Turcsanyi <tu...@apache.org>
---
.../azure/storage/DeleteAzureBlobStorage_v12.java | 15 ++++++++++++++-
.../azure/storage/FetchAzureBlobStorage_v12.java | 4 ++--
.../azure/storage/ITListAzureBlobStorage_v12.java | 3 ++-
3 files changed, 18 insertions(+), 4 deletions(-)
diff --git a/nifi-nar-bundles/nifi-azure-bundle/nifi-azure-processors/src/main/java/org/apache/nifi/processors/azure/storage/DeleteAzureBlobStorage_v12.java b/nifi-nar-bundles/nifi-azure-bundle/nifi-azure-processors/src/main/java/org/apache/nifi/processors/azure/storage/DeleteAzureBlobStorage_v12.java
index 1ba5f7c7d7..b49cbd0247 100644
--- a/nifi-nar-bundles/nifi-azure-bundle/nifi-azure-processors/src/main/java/org/apache/nifi/processors/azure/storage/DeleteAzureBlobStorage_v12.java
+++ b/nifi-nar-bundles/nifi-azure-bundle/nifi-azure-processors/src/main/java/org/apache/nifi/processors/azure/storage/DeleteAzureBlobStorage_v12.java
@@ -40,6 +40,9 @@ import java.util.Collections;
import java.util.List;
import java.util.concurrent.TimeUnit;
+import static org.apache.nifi.processors.azure.storage.utils.BlobAttributes.ATTR_NAME_BLOBNAME;
+import static org.apache.nifi.processors.azure.storage.utils.BlobAttributes.ATTR_NAME_CONTAINER;
+
@Tags({"azure", "microsoft", "cloud", "storage", "blob"})
@SeeAlso({ListAzureBlobStorage_v12.class, FetchAzureBlobStorage_v12.class, PutAzureBlobStorage_v12.class})
@CapabilityDescription("Deletes the specified blob from Azure Blob Storage. The processor uses Azure Blob Storage client library v12.")
@@ -52,6 +55,16 @@ public class DeleteAzureBlobStorage_v12 extends AbstractAzureBlobProcessor_v12 {
public static final AllowableValue DELETE_SNAPSHOTS_ONLY = new AllowableValue(DeleteSnapshotsOptionType.ONLY.name(), "Delete Snapshots Only", "Delete only the blob's snapshots.");
+ public static final PropertyDescriptor CONTAINER = new PropertyDescriptor.Builder()
+ .fromPropertyDescriptor(AzureStorageUtils.CONTAINER)
+ .defaultValue(String.format("${%s}", ATTR_NAME_CONTAINER))
+ .build();
+
+ public static final PropertyDescriptor BLOB_NAME = new PropertyDescriptor.Builder()
+ .fromPropertyDescriptor(AbstractAzureBlobProcessor_v12.BLOB_NAME)
+ .defaultValue(String.format("${%s}", ATTR_NAME_BLOBNAME))
+ .build();
+
public static final PropertyDescriptor DELETE_SNAPSHOTS_OPTION = new PropertyDescriptor.Builder()
.name("delete-snapshots-option")
.displayName("Delete Snapshots Option")
@@ -64,7 +77,7 @@ public class DeleteAzureBlobStorage_v12 extends AbstractAzureBlobProcessor_v12 {
private static final List<PropertyDescriptor> PROPERTIES = Collections.unmodifiableList(Arrays.asList(
STORAGE_CREDENTIALS_SERVICE,
- AzureStorageUtils.CONTAINER,
+ CONTAINER,
BLOB_NAME,
DELETE_SNAPSHOTS_OPTION,
AzureStorageUtils.PROXY_CONFIGURATION_SERVICE
diff --git a/nifi-nar-bundles/nifi-azure-bundle/nifi-azure-processors/src/main/java/org/apache/nifi/processors/azure/storage/FetchAzureBlobStorage_v12.java b/nifi-nar-bundles/nifi-azure-bundle/nifi-azure-processors/src/main/java/org/apache/nifi/processors/azure/storage/FetchAzureBlobStorage_v12.java
index ac6dcae8b3..0a12de3051 100644
--- a/nifi-nar-bundles/nifi-azure-bundle/nifi-azure-processors/src/main/java/org/apache/nifi/processors/azure/storage/FetchAzureBlobStorage_v12.java
+++ b/nifi-nar-bundles/nifi-azure-bundle/nifi-azure-processors/src/main/java/org/apache/nifi/processors/azure/storage/FetchAzureBlobStorage_v12.java
@@ -80,12 +80,12 @@ public class FetchAzureBlobStorage_v12 extends AbstractAzureBlobProcessor_v12 {
public static final PropertyDescriptor CONTAINER = new PropertyDescriptor.Builder()
.fromPropertyDescriptor(AzureStorageUtils.CONTAINER)
- .defaultValue("${azure.container}")
+ .defaultValue(String.format("${%s}", ATTR_NAME_CONTAINER))
.build();
public static final PropertyDescriptor BLOB_NAME = new PropertyDescriptor.Builder()
.fromPropertyDescriptor(AbstractAzureBlobProcessor_v12.BLOB_NAME)
- .defaultValue("${azure.blobname}")
+ .defaultValue(String.format("${%s}", ATTR_NAME_BLOBNAME))
.build();
public static final PropertyDescriptor RANGE_START = new PropertyDescriptor.Builder()
diff --git a/nifi-nar-bundles/nifi-azure-bundle/nifi-azure-processors/src/test/java/org/apache/nifi/processors/azure/storage/ITListAzureBlobStorage_v12.java b/nifi-nar-bundles/nifi-azure-bundle/nifi-azure-processors/src/test/java/org/apache/nifi/processors/azure/storage/ITListAzureBlobStorage_v12.java
index 7421b5f74d..8064ff77e9 100644
--- a/nifi-nar-bundles/nifi-azure-bundle/nifi-azure-processors/src/test/java/org/apache/nifi/processors/azure/storage/ITListAzureBlobStorage_v12.java
+++ b/nifi-nar-bundles/nifi-azure-bundle/nifi-azure-processors/src/test/java/org/apache/nifi/processors/azure/storage/ITListAzureBlobStorage_v12.java
@@ -19,6 +19,7 @@ package org.apache.nifi.processors.azure.storage;
import org.apache.nifi.flowfile.attributes.CoreAttributes;
import org.apache.nifi.processor.Processor;
import org.apache.nifi.processors.azure.storage.utils.AzureStorageUtils;
+import org.apache.nifi.processors.azure.storage.utils.BlobAttributes;
import org.apache.nifi.serialization.record.MockRecordWriter;
import org.apache.nifi.util.MockFlowFile;
import org.junit.jupiter.api.Test;
@@ -198,7 +199,7 @@ public class ITListAzureBlobStorage_v12 extends AbstractAzureBlobStorage_v12IT {
Set<String> expectedBlobNames = new HashSet<>(Arrays.asList(blobNames));
for (MockFlowFile flowFile : flowFiles) {
- String blobName = flowFile.getAttribute("azure.blobname");
+ String blobName = flowFile.getAttribute(BlobAttributes.ATTR_NAME_BLOBNAME);
assertTrue(expectedBlobNames.remove(blobName), "Blob should not be listed: " + blobName);
assertFlowFile(flowFile, blobName);
[nifi] 05/06: NIFI-11045: Sensitive dynamic property support for parameterized queries in ExecuteSQL and ExecuteSQLRecord
Posted by tu...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
turcsanyi pushed a commit to branch support/nifi-1.x
in repository https://gitbox.apache.org/repos/asf/nifi.git
commit 7475232bb09c15bad67bf1fe863d9afd7f503569
Author: Lehel <Le...@hotmail.com>
AuthorDate: Tue Jan 17 16:15:15 2023 +0100
NIFI-11045: Sensitive dynamic property support for parameterized queries in ExecuteSQL and ExecuteSQLRecord
This closes #6853.
Signed-off-by: Peter Turcsanyi <tu...@apache.org>
---
.../java/org/apache/nifi/util/db/JdbcCommon.java | 78 ++++++++++++++--------
.../apache/nifi/util/db/SensitiveValueWrapper.java | 36 ++++++++++
.../org/apache/nifi/util/db/TestJdbcCommon.java | 34 +++++++++-
.../nifi-standard-processors/pom.xml | 9 +--
.../processors/standard/AbstractExecuteSQL.java | 32 ++++++---
.../nifi/processors/standard/ExecuteSQL.java | 61 ++++++++++++++---
.../nifi/processors/standard/ExecuteSQLRecord.java | 47 ++++++++++++-
7 files changed, 243 insertions(+), 54 deletions(-)
diff --git a/nifi-nar-bundles/nifi-extension-utils/nifi-database-utils/src/main/java/org/apache/nifi/util/db/JdbcCommon.java b/nifi-nar-bundles/nifi-extension-utils/nifi-database-utils/src/main/java/org/apache/nifi/util/db/JdbcCommon.java
index ad7471a54c..dbd18e931d 100644
--- a/nifi-nar-bundles/nifi-extension-utils/nifi-database-utils/src/main/java/org/apache/nifi/util/db/JdbcCommon.java
+++ b/nifi-nar-bundles/nifi-extension-utils/nifi-database-utils/src/main/java/org/apache/nifi/util/db/JdbcCommon.java
@@ -68,9 +68,11 @@ import java.time.ZoneId;
import java.time.format.DateTimeFormatter;
import java.util.Date;
import java.util.Map;
+import java.util.Optional;
import java.util.function.Function;
import java.util.regex.Matcher;
import java.util.regex.Pattern;
+import java.util.stream.Collectors;
import static java.sql.Types.ARRAY;
import static java.sql.Types.BIGINT;
@@ -120,6 +122,7 @@ public class JdbcCommon {
public static final Pattern NUMBER_PATTERN = Pattern.compile("-?\\d+");
public static final String MIME_TYPE_AVRO_BINARY = "application/avro-binary";
+ public static final String MASKED_LOG_VALUE = "MASKED VALUE";
public static long convertToAvroStream(final ResultSet rs, final OutputStream outStream, boolean convertNames) throws SQLException, IOException {
return convertToAvroStream(rs, outStream, null, null, convertNames);
@@ -681,32 +684,55 @@ public class JdbcCommon {
* @throws SQLException if the PreparedStatement throws a SQLException when the appropriate setter is called
*/
public static void setParameters(final PreparedStatement stmt, final Map<String, String> attributes) throws SQLException {
- for (final Map.Entry<String, String> entry : attributes.entrySet()) {
- final String key = entry.getKey();
- final Matcher matcher = SQL_TYPE_ATTRIBUTE_PATTERN.matcher(key);
- if (matcher.matches()) {
- final int parameterIndex = Integer.parseInt(matcher.group(1));
-
- final boolean isNumeric = NUMBER_PATTERN.matcher(entry.getValue()).matches();
- if (!isNumeric) {
- throw new SQLDataException("Value of the " + key + " attribute is '" + entry.getValue() + "', which is not a valid JDBC numeral type");
- }
+ final Map<String, SensitiveValueWrapper> sensitiveValueWrapperMap = attributes.entrySet()
+ .stream()
+ .collect(Collectors.toMap(Map.Entry::getKey, e -> new SensitiveValueWrapper(e.getValue(), false)));
+ setSensitiveParameters(stmt, sensitiveValueWrapperMap);
+ }
+
+ /**
+ * Sets all of the appropriate parameters on the given PreparedStatement, based on the given FlowFile attributes
+ * and masks sensitive values.
+ *
+ * @param stmt the statement to set the parameters on
+ * @param attributes the attributes from which to derive parameter indices, values, and types
+ * @throws SQLException if the PreparedStatement throws a SQLException when the appropriate setter is called
+ */
+ public static void setSensitiveParameters(final PreparedStatement stmt, final Map<String, SensitiveValueWrapper> attributes) throws SQLException {
+ for (final Map.Entry<String, SensitiveValueWrapper> entry : attributes.entrySet()) {
+ final String flowFileAttributeKey = entry.getKey();
+ setParameterAtIndex(stmt, attributes, flowFileAttributeKey);
+ }
+ }
- final int jdbcType = Integer.parseInt(entry.getValue());
- final String valueAttrName = "sql.args." + parameterIndex + ".value";
- final String parameterValue = attributes.get(valueAttrName);
- final String formatAttrName = "sql.args." + parameterIndex + ".format";
- final String parameterFormat = attributes.containsKey(formatAttrName)? attributes.get(formatAttrName):"";
+ private static void setParameterAtIndex(final PreparedStatement stmt, final Map<String, SensitiveValueWrapper> attributes, final String flowFileAttributeKey) throws SQLException {
+ final Matcher sqlArgumentTypeMatcher = SQL_TYPE_ATTRIBUTE_PATTERN.matcher(flowFileAttributeKey);
+ if (sqlArgumentTypeMatcher.matches()) {
+ final int sqlArgumentIndex = Integer.parseInt(sqlArgumentTypeMatcher.group(1));
- try {
- JdbcCommon.setParameter(stmt, valueAttrName, parameterIndex, parameterValue, jdbcType, parameterFormat);
- } catch (final NumberFormatException nfe) {
- throw new SQLDataException("The value of the " + valueAttrName + " is '" + parameterValue + "', which cannot be converted into the necessary data type", nfe);
- } catch (ParseException pe) {
- throw new SQLDataException("The value of the " + valueAttrName + " is '" + parameterValue + "', which cannot be converted to a timestamp", pe);
- } catch (UnsupportedEncodingException uee) {
- throw new SQLDataException("The value of the " + valueAttrName + " is '" + parameterValue + "', which cannot be converted to UTF-8", uee);
- }
+ final String sqlArgumentTypeAttributeName = flowFileAttributeKey;
+ final String sqlArgumentType = attributes.get(sqlArgumentTypeAttributeName).getValue();
+ final boolean isNumeric = NUMBER_PATTERN.matcher(sqlArgumentType).matches();
+ if (!isNumeric) {
+ throw new SQLDataException("Value of the " + sqlArgumentTypeAttributeName + " attribute is '" + sqlArgumentType + "', which is not a valid numeral SQL data type");
+ }
+
+ final int sqlType = Integer.parseInt(sqlArgumentType);
+ final String sqlArgumentValueAttributeName = "sql.args." + sqlArgumentIndex + ".value";
+ final Optional<SensitiveValueWrapper> sqlArgumentValueWrapper = Optional.ofNullable(attributes.get(sqlArgumentValueAttributeName));
+ final String sqlArgumentValue = sqlArgumentValueWrapper.map(SensitiveValueWrapper::getValue).orElse(null);
+ final String sqlArgumentLogValue = sqlArgumentValueWrapper.map(a -> a.isSensitive() ? MASKED_LOG_VALUE : a.getValue()).orElse(null);
+ final String sqlArgumentFormatAttributeName = "sql.args." + sqlArgumentIndex + ".format";
+ final String sqlArgumentFormat = attributes.containsKey(sqlArgumentFormatAttributeName) ? attributes.get(sqlArgumentFormatAttributeName).getValue() : "";
+
+ try {
+ JdbcCommon.setParameter(stmt, sqlArgumentIndex, sqlArgumentValue, sqlType, sqlArgumentFormat);
+ } catch (final NumberFormatException nfe) {
+ throw new SQLDataException("The value of the " + sqlArgumentValueAttributeName + " is '" + sqlArgumentLogValue + "', which cannot be converted into the necessary data type", nfe);
+ } catch (ParseException pe) {
+ throw new SQLDataException("The value of the " + sqlArgumentValueAttributeName + " is '" + sqlArgumentLogValue + "', which cannot be converted to a timestamp", pe);
+ } catch (UnsupportedEncodingException uee) {
+ throw new SQLDataException("The value of the " + sqlArgumentValueAttributeName + " is '" + sqlArgumentLogValue + "', which cannot be converted to UTF-8", uee);
}
}
}
@@ -716,13 +742,12 @@ public class JdbcCommon {
* provided PreparedStatement
*
* @param stmt the PreparedStatement to set the parameter on
- * @param attrName the name of the attribute that the parameter is coming from - for logging purposes
* @param parameterIndex the index of the SQL parameter to set
* @param parameterValue the value of the SQL parameter to set
* @param jdbcType the JDBC Type of the SQL parameter to set
* @throws SQLException if the PreparedStatement throws a SQLException when calling the appropriate setter
*/
- public static void setParameter(final PreparedStatement stmt, final String attrName, final int parameterIndex, final String parameterValue, final int jdbcType,
+ public static void setParameter(final PreparedStatement stmt, final int parameterIndex, final String parameterValue, final int jdbcType,
final String valueFormat)
throws SQLException, ParseException, UnsupportedEncodingException {
if (parameterValue == null) {
@@ -902,5 +927,4 @@ public class JdbcCommon {
void processRow(ResultSet resultSet) throws IOException;
void applyStateChanges();
}
-
}
diff --git a/nifi-nar-bundles/nifi-extension-utils/nifi-database-utils/src/main/java/org/apache/nifi/util/db/SensitiveValueWrapper.java b/nifi-nar-bundles/nifi-extension-utils/nifi-database-utils/src/main/java/org/apache/nifi/util/db/SensitiveValueWrapper.java
new file mode 100644
index 0000000000..686e43f9f9
--- /dev/null
+++ b/nifi-nar-bundles/nifi-extension-utils/nifi-database-utils/src/main/java/org/apache/nifi/util/db/SensitiveValueWrapper.java
@@ -0,0 +1,36 @@
+/*
+ * 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.util.db;
+
+public class SensitiveValueWrapper {
+
+ private final String value;
+ private final boolean sensitive;
+
+ public SensitiveValueWrapper(final String value, final boolean sensitive) {
+ this.value = value;
+ this.sensitive = sensitive;
+ }
+
+ public String getValue() {
+ return value;
+ }
+
+ public boolean isSensitive() {
+ return sensitive;
+ }
+}
diff --git a/nifi-nar-bundles/nifi-extension-utils/nifi-database-utils/src/test/java/org/apache/nifi/util/db/TestJdbcCommon.java b/nifi-nar-bundles/nifi-extension-utils/nifi-database-utils/src/test/java/org/apache/nifi/util/db/TestJdbcCommon.java
index aa858316a6..20e6cb87df 100644
--- a/nifi-nar-bundles/nifi-extension-utils/nifi-database-utils/src/test/java/org/apache/nifi/util/db/TestJdbcCommon.java
+++ b/nifi-nar-bundles/nifi-extension-utils/nifi-database-utils/src/test/java/org/apache/nifi/util/db/TestJdbcCommon.java
@@ -67,14 +67,18 @@ import java.sql.Types;
import java.time.Instant;
import java.time.LocalTime;
import java.time.ZoneId;
+import java.util.HashMap;
import java.util.HashSet;
+import java.util.Map;
import java.util.Set;
import java.util.UUID;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.BiConsumer;
import java.util.stream.IntStream;
+import static org.apache.nifi.util.db.JdbcCommon.MASKED_LOG_VALUE;
import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertNotNull;
import static org.junit.jupiter.api.Assertions.assertNull;
import static org.junit.jupiter.api.Assertions.assertThrows;
@@ -196,7 +200,7 @@ public class TestJdbcCommon {
}
@Test
- public void testCreateSchemaOnlyColumnLabel() throws ClassNotFoundException, SQLException {
+ public void testCreateSchemaOnlyColumnLabel() throws SQLException {
final ResultSet resultSet = mock(ResultSet.class);
final ResultSetMetaData resultSetMetaData = mock(ResultSetMetaData.class);
@@ -845,4 +849,32 @@ public class TestJdbcCommon {
assertNotNull(clazz);
}
+ @Test
+ public void testSetSensitiveParametersDoesNotLogSensitiveValues() throws SQLException {
+ final Map<String, SensitiveValueWrapper> attributes = new HashMap<>();
+ attributes.put("sql.args.1.type", new SensitiveValueWrapper("4", false));
+ attributes.put("sql.args.1.value", new SensitiveValueWrapper("123.4", true));
+ try (final Statement stmt = con.createStatement()) {
+ stmt.executeUpdate("CREATE TABLE inttest (id INT)");
+ PreparedStatement ps = con.prepareStatement("INSERT INTO inttest VALUES (?)");
+ final SQLException exception = assertThrows(SQLException.class, () -> JdbcCommon.setSensitiveParameters(ps, attributes));
+ assertTrue(exception.getMessage().contains(MASKED_LOG_VALUE));
+ assertFalse(exception.getMessage().contains("123.4"));
+ }
+ }
+
+ @Test
+ public void testSetParametersDoesNotHaveSensitiveValues() throws SQLException {
+ final Map<String, String> attributes = new HashMap<>();
+ attributes.put("sql.args.1.type", "4");
+ attributes.put("sql.args.1.value","123.4");
+ try (final Statement stmt = con.createStatement()) {
+ stmt.executeUpdate("CREATE TABLE inttest (id INT)");
+ PreparedStatement ps = con.prepareStatement("INSERT INTO inttest VALUES (?)");
+ final SQLException exception = assertThrows(SQLException.class, () -> JdbcCommon.setParameters(ps, attributes));
+ assertFalse(exception.getMessage().contains(MASKED_LOG_VALUE));
+ assertTrue(exception.getMessage().contains("123.4"));
+ }
+ }
+
}
diff --git a/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/pom.xml b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/pom.xml
index 7d5be10a20..6ecc22e9a6 100644
--- a/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/pom.xml
+++ b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/pom.xml
@@ -520,8 +520,8 @@
<exclude>src/test/resources/hello.txt</exclude>
<exclude>src/test/resources/CharacterSetConversionSamples/Converted.txt</exclude>
<exclude>src/test/resources/CharacterSetConversionSamples/Original.txt</exclude>
- <exclude>src/test/resources/CompressedData/SampleFile.txt</exclude>
- <exclude>src/test/resources/CompressedData/SampleFileConcat.txt</exclude>
+ <exclude>src/test/resources/CompressedData/SampleFile.txt*</exclude>
+ <exclude>src/test/resources/CompressedData/SampleFileConcat.txt*</exclude>
<exclude>src/test/resources/ExecuteCommand/1000bytes.txt</exclude>
<exclude>src/test/resources/ExecuteCommand/1mb.txt</exclude>
<exclude>src/test/resources/ExecuteCommand/test.txt</exclude>
@@ -637,11 +637,6 @@
<exclude>src/test/resources/TestXml/xml-bundle-1</exclude>
<exclude>src/test/resources/TestXml/namespaceSplit1.xml</exclude>
<exclude>src/test/resources/TestXml/namespaceSplit2.xml</exclude>
- <exclude>src/test/resources/CompressedData/SampleFile.txt.bz2</exclude>
- <exclude>src/test/resources/CompressedData/SampleFile.txt.gz</exclude>
- <exclude>src/test/resources/CompressedData/SampleFile1.txt.bz2</exclude>
- <exclude>src/test/resources/CompressedData/SampleFile1.txt.gz</exclude>
- <exclude>src/test/resources/CompressedData/SampleFileConcat.txt.bz2</exclude>
<exclude>src/test/resources/ExecuteCommand/TestIngestAndUpdate.jar</exclude>
<exclude>src/test/resources/ExecuteCommand/TestSuccess.jar</exclude>
<exclude>src/test/resources/ExecuteCommand/TestDynamicEnvironment.jar</exclude>
diff --git a/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/AbstractExecuteSQL.java b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/AbstractExecuteSQL.java
index 55a4326908..cb6bc761f8 100644
--- a/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/AbstractExecuteSQL.java
+++ b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/AbstractExecuteSQL.java
@@ -35,6 +35,7 @@ import org.apache.nifi.processor.util.StandardValidators;
import org.apache.nifi.processors.standard.sql.SqlWriter;
import org.apache.nifi.util.StopWatch;
import org.apache.nifi.util.db.JdbcCommon;
+import org.apache.nifi.util.db.SensitiveValueWrapper;
import java.nio.charset.Charset;
import java.sql.Connection;
@@ -52,6 +53,7 @@ import java.util.Set;
import java.util.UUID;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;
+import java.util.stream.Collectors;
public abstract class AbstractExecuteSQL extends AbstractProcessor {
@@ -273,10 +275,23 @@ public abstract class AbstractExecuteSQL extends AbstractProcessor {
throw failure.getRight();
}
+ final Map<String, SensitiveValueWrapper> sqlParameters = context.getProperties()
+ .entrySet()
+ .stream()
+ .filter(e -> e.getKey().isDynamic())
+ .collect(Collectors.toMap(e -> e.getKey().getName(), e -> new SensitiveValueWrapper(e.getValue(), e.getKey().isSensitive())));
+
if (fileToProcess != null) {
- JdbcCommon.setParameters(st, fileToProcess.getAttributes());
+ for (Map.Entry<String, String> entry : fileToProcess.getAttributes().entrySet()) {
+ sqlParameters.put(entry.getKey(), new SensitiveValueWrapper(entry.getValue(), false));
+ }
}
- logger.debug("Executing query {}", new Object[]{selectQuery});
+
+ if (!sqlParameters.isEmpty()) {
+ JdbcCommon.setSensitiveParameters(st, sqlParameters);
+ }
+
+ logger.debug("Executing query {}", selectQuery);
int fragmentIndex = 0;
final String fragmentId = UUID.randomUUID().toString();
@@ -350,7 +365,7 @@ public abstract class AbstractExecuteSQL extends AbstractProcessor {
resultSetFF = session.putAttribute(resultSetFF, FRAGMENT_INDEX, String.valueOf(fragmentIndex));
}
- logger.info("{} contains {} records; transferring to 'success'", new Object[]{resultSetFF, nrOfRows.get()});
+ logger.info("{} contains {} records; transferring to 'success'", resultSetFF, nrOfRows.get());
// Report a FETCH event if there was an incoming flow file, or a RECEIVE event otherwise
if (context.hasIncomingConnection()) {
@@ -414,7 +429,7 @@ public abstract class AbstractExecuteSQL extends AbstractProcessor {
failure = executeConfigStatements(con, postQueries);
if (failure != null) {
selectQuery = failure.getLeft();
- resultSetFlowFiles.forEach(ff -> session.remove(ff));
+ resultSetFlowFiles.forEach(session::remove);
throw failure.getRight();
}
@@ -454,17 +469,14 @@ public abstract class AbstractExecuteSQL extends AbstractProcessor {
// pass the original flow file down the line to trigger downstream processors
if (fileToProcess == null) {
// This can happen if any exceptions occur while setting up the connection, statement, etc.
- logger.error("Unable to execute SQL select query {} due to {}. No FlowFile to route to failure",
- new Object[]{selectQuery, e});
+ logger.error("Unable to execute SQL select query [{}]. No FlowFile to route to failure", selectQuery, e);
context.yield();
} else {
if (context.hasIncomingConnection()) {
- logger.error("Unable to execute SQL select query {} for {} due to {}; routing to failure",
- new Object[]{selectQuery, fileToProcess, e});
+ logger.error("Unable to execute SQL select query [{}] for {} routing to failure", selectQuery, fileToProcess, e);
fileToProcess = session.penalize(fileToProcess);
} else {
- logger.error("Unable to execute SQL select query {} due to {}; routing to failure",
- new Object[]{selectQuery, e});
+ logger.error("Unable to execute SQL select query [{}] routing to failure", selectQuery, e);
context.yield();
}
session.putAttribute(fileToProcess,RESULT_ERROR_MESSAGE,e.getMessage());
diff --git a/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/ExecuteSQL.java b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/ExecuteSQL.java
index cc819db405..850657b97e 100644
--- a/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/ExecuteSQL.java
+++ b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/ExecuteSQL.java
@@ -16,36 +16,41 @@
*/
package org.apache.nifi.processors.standard;
-import java.util.ArrayList;
-import java.util.Collections;
-import java.util.HashSet;
-import java.util.List;
-import java.util.Set;
-
+import org.apache.nifi.annotation.behavior.DynamicProperties;
+import org.apache.nifi.annotation.behavior.DynamicProperty;
import org.apache.nifi.annotation.behavior.EventDriven;
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.SupportsSensitiveDynamicProperties;
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.Tags;
import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.expression.AttributeExpression;
import org.apache.nifi.expression.ExpressionLanguageScope;
import org.apache.nifi.flowfile.FlowFile;
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.processors.standard.sql.DefaultAvroSqlWriter;
import org.apache.nifi.processors.standard.sql.SqlWriter;
import org.apache.nifi.util.db.JdbcCommon;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+
+import static org.apache.nifi.util.db.AvroUtil.CodecType;
import static org.apache.nifi.util.db.JdbcProperties.DEFAULT_PRECISION;
import static org.apache.nifi.util.db.JdbcProperties.DEFAULT_SCALE;
import static org.apache.nifi.util.db.JdbcProperties.NORMALIZE_NAMES_FOR_AVRO;
import static org.apache.nifi.util.db.JdbcProperties.USE_AVRO_LOGICAL_TYPES;
-import static org.apache.nifi.util.db.AvroUtil.CodecType;
@EventDriven
@InputRequirement(Requirement.INPUT_ALLOWED)
@@ -60,7 +65,9 @@ import static org.apache.nifi.util.db.AvroUtil.CodecType;
+ "FlowFile attribute 'executesql.row.count' indicates how many rows were selected.")
@ReadsAttributes({
@ReadsAttribute(attribute = "sql.args.N.type", description = "Incoming FlowFiles are expected to be parametrized SQL statements. The type of each Parameter is specified as an integer "
- + "that represents the JDBC Type of the parameter."),
+ + "that represents the JDBC Type of the parameter. The following types are accepted: [LONGNVARCHAR: -16], [BIT: -7], [BOOLEAN: 16], [TINYINT: -6], [BIGINT: -5], "
+ + "[LONGVARBINARY: -4], [VARBINARY: -3], [BINARY: -2], [LONGVARCHAR: -1], [CHAR: 1], [NUMERIC: 2], [DECIMAL: 3], [INTEGER: 4], [SMALLINT: 5] "
+ + "[FLOAT: 6], [REAL: 7], [DOUBLE: 8], [VARCHAR: 12], [DATE: 91], [TIME: 92], [TIMESTAMP: 93], [VARCHAR: 12], [CLOB: 2005], [NCLOB: 2011]"),
@ReadsAttribute(attribute = "sql.args.N.value", description = "Incoming FlowFiles are expected to be parametrized SQL statements. The value of the Parameters are specified as "
+ "sql.args.1.value, sql.args.2.value, sql.args.3.value, and so on. The type of the sql.args.1.value Parameter is specified by the sql.args.1.type attribute."),
@ReadsAttribute(attribute = "sql.args.N.format", description = "This attribute is always optional, but default options may not always work for your data. "
@@ -102,6 +109,33 @@ import static org.apache.nifi.util.db.AvroUtil.CodecType;
@WritesAttribute(attribute = "input.flowfile.uuid", description = "If the processor has an incoming connection, outgoing FlowFiles will have this attribute "
+ "set to the value of the input FlowFile's UUID. If there is no incoming connection, the attribute will not be added.")
})
+@SupportsSensitiveDynamicProperties
+@DynamicProperties({
+ @DynamicProperty(name = "sql.args.N.type",
+ value = "SQL type argument to be supplied",
+ description = "Incoming FlowFiles are expected to be parametrized SQL statements. The type of each Parameter is specified as an integer "
+ + "that represents the JDBC Type of the parameter. The following types are accepted: [LONGNVARCHAR: -16], [BIT: -7], [BOOLEAN: 16], [TINYINT: -6], [BIGINT: -5], "
+ + "[LONGVARBINARY: -4], [VARBINARY: -3], [BINARY: -2], [LONGVARCHAR: -1], [CHAR: 1], [NUMERIC: 2], [DECIMAL: 3], [INTEGER: 4], [SMALLINT: 5] "
+ + "[FLOAT: 6], [REAL: 7], [DOUBLE: 8], [VARCHAR: 12], [DATE: 91], [TIME: 92], [TIMESTAMP: 93], [VARCHAR: 12], [CLOB: 2005], [NCLOB: 2011]"),
+ @DynamicProperty(name = "sql.args.N.value",
+ value = "Argument to be supplied",
+ description = "Incoming FlowFiles are expected to be parametrized SQL statements. The value of the Parameters are specified as "
+ + "sql.args.1.value, sql.args.2.value, sql.args.3.value, and so on. The type of the sql.args.1.value Parameter is specified by the sql.args.1.type attribute."),
+ @DynamicProperty(name = "sql.args.N.format",
+ value = "SQL format argument to be supplied",
+ description = "This attribute is always optional, but default options may not always work for your data. "
+ + "Incoming FlowFiles are expected to be parametrized SQL statements. In some cases "
+ + "a format option needs to be specified, currently this is only applicable for binary data types, dates, times and timestamps. Binary Data Types (defaults to 'ascii') - "
+ + "ascii: each string character in your attribute value represents a single byte. This is the format provided by Avro Processors. "
+ + "base64: the string is a Base64 encoded string that can be decoded to bytes. "
+ + "hex: the string is hex encoded with all letters in upper case and no '0x' at the beginning. "
+ + "Dates/Times/Timestamps - "
+ + "Date, Time and Timestamp formats all support both custom formats or named format ('yyyy-MM-dd','ISO_OFFSET_DATE_TIME') "
+ + "as specified according to java.time.format.DateTimeFormatter. "
+ + "If not specified, a long value input is expected to be an unix epoch (milli seconds from 1970/1/1), or a string value in "
+ + "'yyyy-MM-dd' format for Date, 'HH:mm:ss.SSS' for Time (some database engines e.g. Derby or MySQL do not support milliseconds and will truncate milliseconds), "
+ + "'yyyy-MM-dd HH:mm:ss.SSS' for Timestamp is used.")
+})
public class ExecuteSQL extends AbstractExecuteSQL {
public static final PropertyDescriptor COMPRESSION_FORMAT = new PropertyDescriptor.Builder()
@@ -157,4 +191,15 @@ public class ExecuteSQL extends AbstractExecuteSQL {
.build();
return new DefaultAvroSqlWriter(options);
}
+
+ @Override
+ protected PropertyDescriptor getSupportedDynamicPropertyDescriptor(final String propertyDescriptorName) {
+ return new PropertyDescriptor.Builder()
+ .name(propertyDescriptorName)
+ .required(false)
+ .dynamic(true)
+ .addValidator(StandardValidators.createAttributeExpressionLanguageValidator(AttributeExpression.ResultType.STRING, true))
+ .addValidator(StandardValidators.ATTRIBUTE_KEY_PROPERTY_NAME_VALIDATOR)
+ .build();
+ }
}
diff --git a/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/ExecuteSQLRecord.java b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/ExecuteSQLRecord.java
index 2ebddacef0..cd36e42fef 100644
--- a/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/ExecuteSQLRecord.java
+++ b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/ExecuteSQLRecord.java
@@ -16,20 +16,25 @@
*/
package org.apache.nifi.processors.standard;
+import org.apache.nifi.annotation.behavior.DynamicProperties;
+import org.apache.nifi.annotation.behavior.DynamicProperty;
import org.apache.nifi.annotation.behavior.EventDriven;
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.SupportsSensitiveDynamicProperties;
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.Tags;
import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.expression.AttributeExpression;
import org.apache.nifi.flowfile.FlowFile;
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.processors.standard.sql.RecordSqlWriter;
import org.apache.nifi.processors.standard.sql.SqlWriter;
import org.apache.nifi.serialization.RecordSetWriterFactory;
@@ -59,7 +64,9 @@ import static org.apache.nifi.util.db.JdbcProperties.USE_AVRO_LOGICAL_TYPES;
+ "FlowFile attribute 'executesql.row.count' indicates how many rows were selected.")
@ReadsAttributes({
@ReadsAttribute(attribute = "sql.args.N.type", description = "Incoming FlowFiles are expected to be parametrized SQL statements. The type of each Parameter is specified as an integer "
- + "that represents the JDBC Type of the parameter."),
+ + "that represents the JDBC Type of the parameter. The following types are accepted: [LONGNVARCHAR: -16], [BIT: -7], [BOOLEAN: 16], [TINYINT: -6], [BIGINT: -5], "
+ + "[LONGVARBINARY: -4], [VARBINARY: -3], [BINARY: -2], [LONGVARCHAR: -1], [CHAR: 1], [NUMERIC: 2], [DECIMAL: 3], [INTEGER: 4], [SMALLINT: 5] "
+ + "[FLOAT: 6], [REAL: 7], [DOUBLE: 8], [VARCHAR: 12], [DATE: 91], [TIME: 92], [TIMESTAMP: 93], [VARCHAR: 12], [CLOB: 2005], [NCLOB: 2011]"),
@ReadsAttribute(attribute = "sql.args.N.value", description = "Incoming FlowFiles are expected to be parametrized SQL statements. The value of the Parameters are specified as "
+ "sql.args.1.value, sql.args.2.value, sql.args.3.value, and so on. The type of the sql.args.1.value Parameter is specified by the sql.args.1.type attribute."),
@ReadsAttribute(attribute = "sql.args.N.format", description = "This attribute is always optional, but default options may not always work for your data. "
@@ -99,6 +106,33 @@ import static org.apache.nifi.util.db.JdbcProperties.USE_AVRO_LOGICAL_TYPES;
@WritesAttribute(attribute = "mime.type", description = "Sets the mime.type attribute to the MIME Type specified by the Record Writer."),
@WritesAttribute(attribute = "record.count", description = "The number of records output by the Record Writer.")
})
+@SupportsSensitiveDynamicProperties
+@DynamicProperties({
+ @DynamicProperty(name = "sql.args.N.type",
+ value = "SQL type argument to be supplied",
+ description = "Incoming FlowFiles are expected to be parametrized SQL statements. The type of each Parameter is specified as an integer "
+ + "that represents the JDBC Type of the parameter. The following types are accepted: [LONGNVARCHAR: -16], [BIT: -7], [BOOLEAN: 16], [TINYINT: -6], [BIGINT: -5], "
+ + "[LONGVARBINARY: -4], [VARBINARY: -3], [BINARY: -2], [LONGVARCHAR: -1], [CHAR: 1], [NUMERIC: 2], [DECIMAL: 3], [INTEGER: 4], [SMALLINT: 5] "
+ + "[FLOAT: 6], [REAL: 7], [DOUBLE: 8], [VARCHAR: 12], [DATE: 91], [TIME: 92], [TIMESTAMP: 93], [VARCHAR: 12], [CLOB: 2005], [NCLOB: 2011]"),
+ @DynamicProperty(name = "sql.args.N.value",
+ value = "Argument to be supplied",
+ description = "Incoming FlowFiles are expected to be parametrized SQL statements. The value of the Parameters are specified as "
+ + "sql.args.1.value, sql.args.2.value, sql.args.3.value, and so on. The type of the sql.args.1.value Parameter is specified by the sql.args.1.type attribute."),
+ @DynamicProperty(name = "sql.args.N.format",
+ value = "SQL format argument to be supplied",
+ description = "This attribute is always optional, but default options may not always work for your data. "
+ + "Incoming FlowFiles are expected to be parametrized SQL statements. In some cases "
+ + "a format option needs to be specified, currently this is only applicable for binary data types, dates, times and timestamps. Binary Data Types (defaults to 'ascii') - "
+ + "ascii: each string character in your attribute value represents a single byte. This is the format provided by Avro Processors. "
+ + "base64: the string is a Base64 encoded string that can be decoded to bytes. "
+ + "hex: the string is hex encoded with all letters in upper case and no '0x' at the beginning. "
+ + "Dates/Times/Timestamps - "
+ + "Date, Time and Timestamp formats all support both custom formats or named format ('yyyy-MM-dd','ISO_OFFSET_DATE_TIME') "
+ + "as specified according to java.time.format.DateTimeFormatter. "
+ + "If not specified, a long value input is expected to be an unix epoch (milli seconds from 1970/1/1), or a string value in "
+ + "'yyyy-MM-dd' format for Date, 'HH:mm:ss.SSS' for Time (some database engines e.g. Derby or MySQL do not support milliseconds and will truncate milliseconds), "
+ + "'yyyy-MM-dd HH:mm:ss.SSS' for Timestamp is used.")
+})
public class ExecuteSQLRecord extends AbstractExecuteSQL {
@@ -161,4 +195,15 @@ public class ExecuteSQLRecord extends AbstractExecuteSQL {
return new RecordSqlWriter(recordSetWriterFactory, options, maxRowsPerFlowFile, fileToProcess == null ? Collections.emptyMap() : fileToProcess.getAttributes());
}
+
+ @Override
+ protected PropertyDescriptor getSupportedDynamicPropertyDescriptor(final String propertyDescriptorName) {
+ return new PropertyDescriptor.Builder()
+ .name(propertyDescriptorName)
+ .required(false)
+ .dynamic(true)
+ .addValidator(StandardValidators.createAttributeExpressionLanguageValidator(AttributeExpression.ResultType.STRING, true))
+ .addValidator(StandardValidators.ATTRIBUTE_KEY_PROPERTY_NAME_VALIDATOR)
+ .build();
+ }
}
[nifi] 01/06: NIFI-11124: Add hadoop.file.url attribute to HDFS processors
Posted by tu...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
turcsanyi pushed a commit to branch support/nifi-1.x
in repository https://gitbox.apache.org/repos/asf/nifi.git
commit 85b36cac609f752040bbe8b3a5a676a22988d8d6
Author: Mark Bathori <ba...@gmail.com>
AuthorDate: Wed Feb 1 20:21:51 2023 +0100
NIFI-11124: Add hadoop.file.url attribute to HDFS processors
This closes #6916.
Signed-off-by: Peter Turcsanyi <tu...@apache.org>
---
.../processors/hadoop/AbstractHadoopProcessor.java | 1 +
.../processors/hadoop/AbstractFetchHDFSRecord.java | 31 ++++++++++---------
.../processors/hadoop/AbstractPutHDFSRecord.java | 1 +
.../apache/nifi/processors/hadoop/DeleteHDFS.java | 2 ++
.../apache/nifi/processors/hadoop/FetchHDFS.java | 9 ++++--
.../apache/nifi/processors/hadoop/MoveHDFS.java | 6 +++-
.../org/apache/nifi/processors/hadoop/PutHDFS.java | 2 ++
.../apache/nifi/processors/hadoop/PutHDFSTest.java | 6 ++++
.../nifi/processors/hadoop/TestDeleteHDFS.java | 5 +++
.../nifi/processors/hadoop/TestFetchHDFS.java | 4 +++
.../org/apache/nifi/processors/orc/PutORC.java | 1 +
.../org/apache/nifi/processors/orc/PutORCTest.java | 3 ++
.../nifi/processors/parquet/FetchParquet.java | 3 +-
.../apache/nifi/processors/parquet/PutParquet.java | 1 +
.../nifi/processors/parquet/FetchParquetTest.java | 2 ++
.../nifi/processors/parquet/PutParquetTest.java | 36 ++++++++++++----------
16 files changed, 77 insertions(+), 36 deletions(-)
diff --git a/nifi-nar-bundles/nifi-extension-utils/nifi-hadoop-utils/src/main/java/org/apache/nifi/processors/hadoop/AbstractHadoopProcessor.java b/nifi-nar-bundles/nifi-extension-utils/nifi-hadoop-utils/src/main/java/org/apache/nifi/processors/hadoop/AbstractHadoopProcessor.java
index f7c082bf12..a967b9037a 100644
--- a/nifi-nar-bundles/nifi-extension-utils/nifi-hadoop-utils/src/main/java/org/apache/nifi/processors/hadoop/AbstractHadoopProcessor.java
+++ b/nifi-nar-bundles/nifi-extension-utils/nifi-hadoop-utils/src/main/java/org/apache/nifi/processors/hadoop/AbstractHadoopProcessor.java
@@ -158,6 +158,7 @@ public abstract class AbstractHadoopProcessor extends AbstractProcessor implemen
public static final String ABSOLUTE_HDFS_PATH_ATTRIBUTE = "absolute.hdfs.path";
+ public static final String HADOOP_FILE_URL_ATTRIBUTE = "hadoop.file.url";
protected static final String TARGET_HDFS_DIR_CREATED_ATTRIBUTE = "target.dir.created";
diff --git a/nifi-nar-bundles/nifi-extension-utils/nifi-record-utils/nifi-hadoop-record-utils/src/main/java/org/apache/nifi/processors/hadoop/AbstractFetchHDFSRecord.java b/nifi-nar-bundles/nifi-extension-utils/nifi-record-utils/nifi-hadoop-record-utils/src/main/java/org/apache/nifi/processors/hadoop/AbstractFetchHDFSRecord.java
index 33e762fa31..03770ac56f 100644
--- a/nifi-nar-bundles/nifi-extension-utils/nifi-record-utils/nifi-hadoop-record-utils/src/main/java/org/apache/nifi/processors/hadoop/AbstractFetchHDFSRecord.java
+++ b/nifi-nar-bundles/nifi-extension-utils/nifi-record-utils/nifi-hadoop-record-utils/src/main/java/org/apache/nifi/processors/hadoop/AbstractFetchHDFSRecord.java
@@ -16,21 +16,6 @@
*/
package org.apache.nifi.processors.hadoop;
-import java.io.BufferedOutputStream;
-import java.io.FileNotFoundException;
-import java.io.IOException;
-import java.io.OutputStream;
-import java.security.PrivilegedAction;
-import java.util.ArrayList;
-import java.util.Collections;
-import java.util.HashMap;
-import java.util.HashSet;
-import java.util.List;
-import java.util.Map;
-import java.util.Set;
-import java.util.concurrent.TimeUnit;
-import java.util.concurrent.atomic.AtomicReference;
-
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
@@ -57,6 +42,21 @@ import org.apache.nifi.serialization.record.Record;
import org.apache.nifi.serialization.record.RecordSchema;
import org.apache.nifi.util.StopWatch;
+import java.io.BufferedOutputStream;
+import java.io.FileNotFoundException;
+import java.io.IOException;
+import java.io.OutputStream;
+import java.security.PrivilegedAction;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicReference;
+
/**
* Base processor for reading a data from HDFS that can be fetched into records.
*/
@@ -234,6 +234,7 @@ public abstract class AbstractFetchHDFSRecord extends AbstractHadoopProcessor {
final Path qualifiedPath = path.makeQualified(fileSystem.getUri(), fileSystem.getWorkingDirectory());
+ successFlowFile = session.putAttribute(successFlowFile, HADOOP_FILE_URL_ATTRIBUTE, qualifiedPath.toString());
getLogger().info("Successfully received content from {} for {} in {} milliseconds", new Object[] {qualifiedPath, successFlowFile, stopWatch.getDuration()});
session.getProvenanceReporter().fetch(successFlowFile, qualifiedPath.toString(), stopWatch.getDuration(TimeUnit.MILLISECONDS));
session.transfer(successFlowFile, REL_SUCCESS);
diff --git a/nifi-nar-bundles/nifi-extension-utils/nifi-record-utils/nifi-hadoop-record-utils/src/main/java/org/apache/nifi/processors/hadoop/AbstractPutHDFSRecord.java b/nifi-nar-bundles/nifi-extension-utils/nifi-record-utils/nifi-hadoop-record-utils/src/main/java/org/apache/nifi/processors/hadoop/AbstractPutHDFSRecord.java
index c2fb3bd725..5aef6b4621 100644
--- a/nifi-nar-bundles/nifi-extension-utils/nifi-record-utils/nifi-hadoop-record-utils/src/main/java/org/apache/nifi/processors/hadoop/AbstractPutHDFSRecord.java
+++ b/nifi-nar-bundles/nifi-extension-utils/nifi-record-utils/nifi-hadoop-record-utils/src/main/java/org/apache/nifi/processors/hadoop/AbstractPutHDFSRecord.java
@@ -378,6 +378,7 @@ public abstract class AbstractPutHDFSRecord extends AbstractHadoopProcessor {
// Send a provenance event and transfer to success
final Path qualifiedPath = destFile.makeQualified(fileSystem.getUri(), fileSystem.getWorkingDirectory());
+ putFlowFile = session.putAttribute(putFlowFile, HADOOP_FILE_URL_ATTRIBUTE, qualifiedPath.toString());
session.getProvenanceReporter().send(putFlowFile, qualifiedPath.toString());
session.transfer(putFlowFile, REL_SUCCESS);
diff --git a/nifi-nar-bundles/nifi-hadoop-bundle/nifi-hdfs-processors/src/main/java/org/apache/nifi/processors/hadoop/DeleteHDFS.java b/nifi-nar-bundles/nifi-hadoop-bundle/nifi-hdfs-processors/src/main/java/org/apache/nifi/processors/hadoop/DeleteHDFS.java
index 2c9285fab5..6b676f9493 100644
--- a/nifi-nar-bundles/nifi-hadoop-bundle/nifi-hdfs-processors/src/main/java/org/apache/nifi/processors/hadoop/DeleteHDFS.java
+++ b/nifi-nar-bundles/nifi-hadoop-bundle/nifi-hdfs-processors/src/main/java/org/apache/nifi/processors/hadoop/DeleteHDFS.java
@@ -72,6 +72,7 @@ import java.util.regex.Pattern;
+ "If multiple files are deleted, then only the last filename is set."),
@WritesAttribute(attribute="hdfs.path", description="HDFS Path specified in the delete request. "
+ "If multiple paths are deleted, then only the last path is set."),
+ @WritesAttribute(attribute = "hadoop.file.url", description = "The hadoop url for the file to be deleted."),
@WritesAttribute(attribute="hdfs.error.message", description="HDFS error message related to the hdfs.error.code")
})
@SeeAlso({ListHDFS.class, PutHDFS.class})
@@ -176,6 +177,7 @@ public class DeleteHDFS extends AbstractHadoopProcessor {
fileSystem.delete(path, isRecursive(context, session));
getLogger().debug("For flowfile {} Deleted file at path {} with name {}", new Object[]{originalFlowFile, path.getParent().toString(), path.getName()});
final Path qualifiedPath = path.makeQualified(fileSystem.getUri(), fileSystem.getWorkingDirectory());
+ flowFile = session.putAttribute(flowFile, HADOOP_FILE_URL_ATTRIBUTE, qualifiedPath.toString());
session.getProvenanceReporter().invokeRemoteProcess(flowFile, qualifiedPath.toString());
} catch (IOException ioe) {
// One possible scenario is that the IOException is permissions based, however it would be impractical to check every possible
diff --git a/nifi-nar-bundles/nifi-hadoop-bundle/nifi-hdfs-processors/src/main/java/org/apache/nifi/processors/hadoop/FetchHDFS.java b/nifi-nar-bundles/nifi-hadoop-bundle/nifi-hdfs-processors/src/main/java/org/apache/nifi/processors/hadoop/FetchHDFS.java
index e4acaac109..2a4986cadd 100644
--- a/nifi-nar-bundles/nifi-hadoop-bundle/nifi-hdfs-processors/src/main/java/org/apache/nifi/processors/hadoop/FetchHDFS.java
+++ b/nifi-nar-bundles/nifi-hadoop-bundle/nifi-hdfs-processors/src/main/java/org/apache/nifi/processors/hadoop/FetchHDFS.java
@@ -31,6 +31,7 @@ import org.apache.nifi.annotation.behavior.Restricted;
import org.apache.nifi.annotation.behavior.Restriction;
import org.apache.nifi.annotation.behavior.SupportsBatching;
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;
@@ -61,8 +62,11 @@ import java.util.concurrent.TimeUnit;
@Tags({"hadoop", "hcfs", "hdfs", "get", "ingest", "fetch", "source"})
@CapabilityDescription("Retrieves a file from HDFS. The content of the incoming FlowFile is replaced by the content of the file in HDFS. "
+ "The file in HDFS is left intact without any changes being made to it.")
-@WritesAttribute(attribute="hdfs.failure.reason", description="When a FlowFile is routed to 'failure', this attribute is added indicating why the file could "
- + "not be fetched from HDFS")
+@WritesAttributes({
+ @WritesAttribute(attribute="hdfs.failure.reason", description="When a FlowFile is routed to 'failure', this attribute is added indicating why the file could "
+ + "not be fetched from HDFS"),
+ @WritesAttribute(attribute = "hadoop.file.url", description = "The hadoop url for the file is stored in this attribute.")
+})
@SeeAlso({ListHDFS.class, GetHDFS.class, PutHDFS.class})
@Restricted(restrictions = {
@Restriction(
@@ -173,6 +177,7 @@ public class FetchHDFS extends AbstractHadoopProcessor {
stopWatch.stop();
getLogger().info("Successfully received content from {} for {} in {}", new Object[] {qualifiedPath, flowFile, stopWatch.getDuration()});
+ flowFile = session.putAttribute(flowFile, HADOOP_FILE_URL_ATTRIBUTE, qualifiedPath.toString());
session.getProvenanceReporter().fetch(flowFile, qualifiedPath.toString(), stopWatch.getDuration(TimeUnit.MILLISECONDS));
session.transfer(flowFile, getSuccessRelationship());
} catch (final FileNotFoundException | AccessControlException e) {
diff --git a/nifi-nar-bundles/nifi-hadoop-bundle/nifi-hdfs-processors/src/main/java/org/apache/nifi/processors/hadoop/MoveHDFS.java b/nifi-nar-bundles/nifi-hadoop-bundle/nifi-hdfs-processors/src/main/java/org/apache/nifi/processors/hadoop/MoveHDFS.java
index a01d236f51..d7a3d13045 100644
--- a/nifi-nar-bundles/nifi-hadoop-bundle/nifi-hdfs-processors/src/main/java/org/apache/nifi/processors/hadoop/MoveHDFS.java
+++ b/nifi-nar-bundles/nifi-hadoop-bundle/nifi-hdfs-processors/src/main/java/org/apache/nifi/processors/hadoop/MoveHDFS.java
@@ -75,7 +75,9 @@ import java.util.regex.Pattern;
@ReadsAttribute(attribute = "filename", description = "The name of the file written to HDFS comes from the value of this attribute.")
@WritesAttributes({
@WritesAttribute(attribute = "filename", description = "The name of the file written to HDFS is stored in this attribute."),
- @WritesAttribute(attribute = "absolute.hdfs.path", description = "The absolute path to the file on HDFS is stored in this attribute.")})
+ @WritesAttribute(attribute = "absolute.hdfs.path", description = "The absolute path to the file on HDFS is stored in this attribute."),
+ @WritesAttribute(attribute = "hadoop.file.url", description = "The hadoop url for the file is stored in this attribute.")
+})
@SeeAlso({PutHDFS.class, GetHDFS.class})
@Restricted(restrictions = {
@Restriction(
@@ -426,6 +428,8 @@ public class MoveHDFS extends AbstractHadoopProcessor {
final String hdfsPath = newFile.getParent().toString();
flowFile = session.putAttribute(flowFile, CoreAttributes.FILENAME.key(), newFilename);
flowFile = session.putAttribute(flowFile, ABSOLUTE_HDFS_PATH_ATTRIBUTE, hdfsPath);
+ final Path qualifiedPath = newFile.makeQualified(hdfs.getUri(), hdfs.getWorkingDirectory());
+ flowFile = session.putAttribute(flowFile, HADOOP_FILE_URL_ATTRIBUTE, qualifiedPath.toString());
final String transitUri = hdfs.getUri() + StringUtils.prependIfMissing(outputPath, "/");
session.getProvenanceReporter().send(flowFile, transitUri);
session.transfer(flowFile, REL_SUCCESS);
diff --git a/nifi-nar-bundles/nifi-hadoop-bundle/nifi-hdfs-processors/src/main/java/org/apache/nifi/processors/hadoop/PutHDFS.java b/nifi-nar-bundles/nifi-hadoop-bundle/nifi-hdfs-processors/src/main/java/org/apache/nifi/processors/hadoop/PutHDFS.java
index eb7dfb56bb..91e91ff7b1 100644
--- a/nifi-nar-bundles/nifi-hadoop-bundle/nifi-hdfs-processors/src/main/java/org/apache/nifi/processors/hadoop/PutHDFS.java
+++ b/nifi-nar-bundles/nifi-hadoop-bundle/nifi-hdfs-processors/src/main/java/org/apache/nifi/processors/hadoop/PutHDFS.java
@@ -89,6 +89,7 @@ import java.util.stream.Stream;
@WritesAttributes({
@WritesAttribute(attribute = "filename", description = "The name of the file written to HDFS is stored in this attribute."),
@WritesAttribute(attribute = "absolute.hdfs.path", description = "The absolute path to the file on HDFS is stored in this attribute."),
+ @WritesAttribute(attribute = "hadoop.file.url", description = "The hadoop url for the file is stored in this attribute."),
@WritesAttribute(attribute = "target.dir.created", description = "The result(true/false) indicates if the folder is created by the processor.")
})
@SeeAlso(GetHDFS.class)
@@ -456,6 +457,7 @@ public class PutHDFS extends AbstractHadoopProcessor {
putFlowFile = session.putAttribute(putFlowFile, ABSOLUTE_HDFS_PATH_ATTRIBUTE, hdfsPath);
putFlowFile = session.putAttribute(putFlowFile, TARGET_HDFS_DIR_CREATED_ATTRIBUTE, String.valueOf(targetDirCreated));
final Path qualifiedPath = copyFile.makeQualified(hdfs.getUri(), hdfs.getWorkingDirectory());
+ putFlowFile = session.putAttribute(putFlowFile, HADOOP_FILE_URL_ATTRIBUTE, qualifiedPath.toString());
session.getProvenanceReporter().send(putFlowFile, qualifiedPath.toString());
session.transfer(putFlowFile, getSuccessRelationship());
diff --git a/nifi-nar-bundles/nifi-hadoop-bundle/nifi-hdfs-processors/src/test/java/org/apache/nifi/processors/hadoop/PutHDFSTest.java b/nifi-nar-bundles/nifi-hadoop-bundle/nifi-hdfs-processors/src/test/java/org/apache/nifi/processors/hadoop/PutHDFSTest.java
index ff603352ee..ab33df270e 100644
--- a/nifi-nar-bundles/nifi-hadoop-bundle/nifi-hdfs-processors/src/test/java/org/apache/nifi/processors/hadoop/PutHDFSTest.java
+++ b/nifi-nar-bundles/nifi-hadoop-bundle/nifi-hdfs-processors/src/test/java/org/apache/nifi/processors/hadoop/PutHDFSTest.java
@@ -233,6 +233,7 @@ public class PutHDFSTest {
assertEquals(ProvenanceEventType.SEND, sendEvent.getEventType());
// If it runs with a real HDFS, the protocol will be "hdfs://", but with a local filesystem, just assert the filename.
assertTrue(sendEvent.getTransitUri().endsWith(TARGET_DIRECTORY + "/" + FILE_NAME));
+ assertTrue(flowFile.getAttribute(PutHDFS.HADOOP_FILE_URL_ATTRIBUTE).endsWith(TARGET_DIRECTORY + "/" + FILE_NAME));
verify(spyFileSystem, times(1)).rename(any(Path.class), any(Path.class));
}
@@ -267,6 +268,8 @@ public class PutHDFSTest {
assertEquals(FILE_NAME, flowFile.getAttribute(CoreAttributes.FILENAME.key()));
assertEquals(TARGET_DIRECTORY, flowFile.getAttribute(PutHDFS.ABSOLUTE_HDFS_PATH_ATTRIBUTE));
assertEquals("true", flowFile.getAttribute(PutHDFS.TARGET_HDFS_DIR_CREATED_ATTRIBUTE));
+ // If it runs with a real HDFS, the protocol will be "hdfs://", but with a local filesystem, just assert the filename.
+ assertTrue(flowFile.getAttribute(PutHDFS.HADOOP_FILE_URL_ATTRIBUTE).endsWith(TARGET_DIRECTORY + "/" + FILE_NAME));
verify(spyFileSystem, Mockito.never()).rename(any(Path.class), any(Path.class));
}
@@ -304,6 +307,7 @@ public class PutHDFSTest {
assertEquals(ProvenanceEventType.SEND, sendEvent.getEventType());
// If it runs with a real HDFS, the protocol will be "hdfs://", but with a local filesystem, just assert the filename.
assertTrue(sendEvent.getTransitUri().endsWith("target/test-classes/randombytes-1"));
+ assertTrue(flowFile.getAttribute(PutHDFS.HADOOP_FILE_URL_ATTRIBUTE).endsWith("target/test-classes/randombytes-1"));
}
@Test
@@ -330,6 +334,7 @@ public class PutHDFSTest {
assertTrue(mockFileSystem.exists(new Path("target/test-classes/randombytes-1.gz")));
assertEquals("randombytes-1.gz", flowFile.getAttribute(CoreAttributes.FILENAME.key()));
assertEquals("target/test-classes", flowFile.getAttribute(PutHDFS.ABSOLUTE_HDFS_PATH_ATTRIBUTE));
+ assertTrue(flowFile.getAttribute(PutHDFS.HADOOP_FILE_URL_ATTRIBUTE).endsWith(TARGET_DIRECTORY + "/" + FILE_NAME + ".gz"));
}
@Test
@@ -423,6 +428,7 @@ public class PutHDFSTest {
assertTrue(mockFileSystem.exists(new Path("target/data_test/randombytes-1")));
assertEquals("randombytes-1", flowFile.getAttribute(CoreAttributes.FILENAME.key()));
assertEquals("target/data_test", flowFile.getAttribute(PutHDFS.ABSOLUTE_HDFS_PATH_ATTRIBUTE));
+ assertTrue(flowFile.getAttribute(PutHDFS.HADOOP_FILE_URL_ATTRIBUTE).endsWith("target/data_test/" + FILE_NAME));
}
@Test
diff --git a/nifi-nar-bundles/nifi-hadoop-bundle/nifi-hdfs-processors/src/test/java/org/apache/nifi/processors/hadoop/TestDeleteHDFS.java b/nifi-nar-bundles/nifi-hadoop-bundle/nifi-hdfs-processors/src/test/java/org/apache/nifi/processors/hadoop/TestDeleteHDFS.java
index e3472bf331..6b64cca767 100644
--- a/nifi-nar-bundles/nifi-hadoop-bundle/nifi-hdfs-processors/src/test/java/org/apache/nifi/processors/hadoop/TestDeleteHDFS.java
+++ b/nifi-nar-bundles/nifi-hadoop-bundle/nifi-hdfs-processors/src/test/java/org/apache/nifi/processors/hadoop/TestDeleteHDFS.java
@@ -36,6 +36,7 @@ import java.net.URI;
import java.util.List;
import java.util.Map;
+import static org.apache.nifi.processors.hadoop.AbstractHadoopProcessor.HADOOP_FILE_URL_ATTRIBUTE;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertTrue;
@@ -77,6 +78,10 @@ public class TestDeleteHDFS {
assertEquals(1, provenanceEvents.size());
assertEquals(ProvenanceEventType.REMOTE_INVOCATION, provenanceEvents.get(0).getEventType());
assertEquals("hdfs://0.example.com:8020/some/path/to/file.txt", provenanceEvents.get(0).getTransitUri());
+
+ MockFlowFile flowFile = runner.getFlowFilesForRelationship(DeleteHDFS.REL_SUCCESS).get(0);
+ assertEquals("hdfs://0.example.com:8020/some/path/to/file.txt", flowFile.getAttribute(HADOOP_FILE_URL_ATTRIBUTE));
+
}
@Test
diff --git a/nifi-nar-bundles/nifi-hadoop-bundle/nifi-hdfs-processors/src/test/java/org/apache/nifi/processors/hadoop/TestFetchHDFS.java b/nifi-nar-bundles/nifi-hadoop-bundle/nifi-hdfs-processors/src/test/java/org/apache/nifi/processors/hadoop/TestFetchHDFS.java
index 4d79c309a7..43f0c3ce6d 100644
--- a/nifi-nar-bundles/nifi-hadoop-bundle/nifi-hdfs-processors/src/test/java/org/apache/nifi/processors/hadoop/TestFetchHDFS.java
+++ b/nifi-nar-bundles/nifi-hadoop-bundle/nifi-hdfs-processors/src/test/java/org/apache/nifi/processors/hadoop/TestFetchHDFS.java
@@ -34,6 +34,7 @@ import java.util.HashMap;
import java.util.List;
import java.util.Map;
+import static org.apache.nifi.processors.hadoop.AbstractHadoopProcessor.HADOOP_FILE_URL_ATTRIBUTE;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertTrue;
@@ -71,6 +72,9 @@ public class TestFetchHDFS {
assertEquals(ProvenanceEventType.FETCH, fetchEvent.getEventType());
// If it runs with a real HDFS, the protocol will be "hdfs://", but with a local filesystem, just assert the filename.
assertTrue(fetchEvent.getTransitUri().endsWith(file));
+
+ MockFlowFile flowFile = runner.getFlowFilesForRelationship(DeleteHDFS.REL_SUCCESS).get(0);
+ assertTrue(flowFile.getAttribute(HADOOP_FILE_URL_ATTRIBUTE).endsWith(file));
}
@Test
diff --git a/nifi-nar-bundles/nifi-hive-bundle/nifi-hive3-processors/src/main/java/org/apache/nifi/processors/orc/PutORC.java b/nifi-nar-bundles/nifi-hive-bundle/nifi-hive3-processors/src/main/java/org/apache/nifi/processors/orc/PutORC.java
index f691038c58..4e5511ac7d 100644
--- a/nifi-nar-bundles/nifi-hive-bundle/nifi-hive3-processors/src/main/java/org/apache/nifi/processors/orc/PutORC.java
+++ b/nifi-nar-bundles/nifi-hive-bundle/nifi-hive3-processors/src/main/java/org/apache/nifi/processors/orc/PutORC.java
@@ -60,6 +60,7 @@ import java.util.List;
@WritesAttributes({
@WritesAttribute(attribute = "filename", description = "The name of the file is stored in this attribute."),
@WritesAttribute(attribute = "absolute.hdfs.path", description = "The absolute path to the file is stored in this attribute."),
+ @WritesAttribute(attribute = "hadoop.file.url", description = "The hadoop url for the file is stored in this attribute."),
@WritesAttribute(attribute = "record.count", description = "The number of records written to the ORC file"),
@WritesAttribute(attribute = "hive.ddl", description = "Creates a partial Hive DDL statement for creating an external table in Hive from the destination folder. "
+ "This can be used in ReplaceText for setting the content to the DDL. To make it valid DDL, add \"LOCATION '<path_to_orc_file_in_hdfs>'\", where "
diff --git a/nifi-nar-bundles/nifi-hive-bundle/nifi-hive3-processors/src/test/java/org/apache/nifi/processors/orc/PutORCTest.java b/nifi-nar-bundles/nifi-hive-bundle/nifi-hive3-processors/src/test/java/org/apache/nifi/processors/orc/PutORCTest.java
index 38338d7c9c..b6390f9fb5 100644
--- a/nifi-nar-bundles/nifi-hive-bundle/nifi-hive3-processors/src/test/java/org/apache/nifi/processors/orc/PutORCTest.java
+++ b/nifi-nar-bundles/nifi-hive-bundle/nifi-hive3-processors/src/test/java/org/apache/nifi/processors/orc/PutORCTest.java
@@ -80,6 +80,7 @@ import java.util.Map;
import java.util.TimeZone;
import java.util.function.BiFunction;
+import static org.apache.nifi.processors.hadoop.AbstractHadoopProcessor.HADOOP_FILE_URL_ATTRIBUTE;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertNotNull;
@@ -191,6 +192,7 @@ public class PutORCTest {
mockFlowFile.assertAttributeEquals(PutORC.RECORD_COUNT_ATTR, "100");
mockFlowFile.assertAttributeEquals(PutORC.HIVE_DDL_ATTRIBUTE,
"CREATE EXTERNAL TABLE IF NOT EXISTS `myTable` (`name` STRING, `favorite_number` INT, `favorite_color` STRING, `scale` DOUBLE) STORED AS ORC");
+ assertTrue(mockFlowFile.getAttribute(HADOOP_FILE_URL_ATTRIBUTE).endsWith(DIRECTORY + "/" + filename));
// verify we generated a provenance event
final List<ProvenanceEventRecord> provEvents = testRunner.getProvenanceEvents();
@@ -260,6 +262,7 @@ public class PutORCTest {
// DDL will be created with field names normalized (lowercased, e.g.) for Hive by default
mockFlowFile.assertAttributeEquals(PutORC.HIVE_DDL_ATTRIBUTE,
"CREATE EXTERNAL TABLE IF NOT EXISTS `myTable` (`id` INT, `timemillis` INT, `timestampmillis` TIMESTAMP, `dt` DATE, `dec` DECIMAL) STORED AS ORC");
+ assertTrue(mockFlowFile.getAttribute(HADOOP_FILE_URL_ATTRIBUTE).endsWith(DIRECTORY + "/" + filename));
// verify we generated a provenance event
final List<ProvenanceEventRecord> provEvents = testRunner.getProvenanceEvents();
diff --git a/nifi-nar-bundles/nifi-parquet-bundle/nifi-parquet-processors/src/main/java/org/apache/nifi/processors/parquet/FetchParquet.java b/nifi-nar-bundles/nifi-parquet-bundle/nifi-parquet-processors/src/main/java/org/apache/nifi/processors/parquet/FetchParquet.java
index 17e99c6132..6d289da8ba 100644
--- a/nifi-nar-bundles/nifi-parquet-bundle/nifi-parquet-processors/src/main/java/org/apache/nifi/processors/parquet/FetchParquet.java
+++ b/nifi-nar-bundles/nifi-parquet-bundle/nifi-parquet-processors/src/main/java/org/apache/nifi/processors/parquet/FetchParquet.java
@@ -48,7 +48,8 @@ import java.io.IOException;
@WritesAttributes({
@WritesAttribute(attribute="fetch.failure.reason", description="When a FlowFile is routed to 'failure', this attribute is added " +
"indicating why the file could not be fetched from the given filesystem."),
- @WritesAttribute(attribute = "record.count", description = "The number of records in the resulting flow file")
+ @WritesAttribute(attribute = "record.count", description = "The number of records in the resulting flow file"),
+ @WritesAttribute(attribute = "hadoop.file.url", description = "The hadoop url for the file is stored in this attribute.")
})
@SeeAlso({PutParquet.class})
@Restricted(restrictions = {
diff --git a/nifi-nar-bundles/nifi-parquet-bundle/nifi-parquet-processors/src/main/java/org/apache/nifi/processors/parquet/PutParquet.java b/nifi-nar-bundles/nifi-parquet-bundle/nifi-parquet-processors/src/main/java/org/apache/nifi/processors/parquet/PutParquet.java
index 2d656f2b10..eab970bd0d 100644
--- a/nifi-nar-bundles/nifi-parquet-bundle/nifi-parquet-processors/src/main/java/org/apache/nifi/processors/parquet/PutParquet.java
+++ b/nifi-nar-bundles/nifi-parquet-bundle/nifi-parquet-processors/src/main/java/org/apache/nifi/processors/parquet/PutParquet.java
@@ -66,6 +66,7 @@ import static org.apache.nifi.parquet.utils.ParquetUtils.applyCommonConfig;
@WritesAttributes({
@WritesAttribute(attribute = "filename", description = "The name of the file is stored in this attribute."),
@WritesAttribute(attribute = "absolute.hdfs.path", description = "The absolute path to the file is stored in this attribute."),
+ @WritesAttribute(attribute = "hadoop.file.url", description = "The hadoop url for the file is stored in this attribute."),
@WritesAttribute(attribute = "record.count", description = "The number of records written to the Parquet file")
})
@Restricted(restrictions = {
diff --git a/nifi-nar-bundles/nifi-parquet-bundle/nifi-parquet-processors/src/test/java/org/apache/nifi/processors/parquet/FetchParquetTest.java b/nifi-nar-bundles/nifi-parquet-bundle/nifi-parquet-processors/src/test/java/org/apache/nifi/processors/parquet/FetchParquetTest.java
index 3bb5475bd9..df412c8fe2 100644
--- a/nifi-nar-bundles/nifi-parquet-bundle/nifi-parquet-processors/src/test/java/org/apache/nifi/processors/parquet/FetchParquetTest.java
+++ b/nifi-nar-bundles/nifi-parquet-bundle/nifi-parquet-processors/src/test/java/org/apache/nifi/processors/parquet/FetchParquetTest.java
@@ -58,6 +58,7 @@ import java.util.HashMap;
import java.util.List;
import java.util.Map;
+import static org.apache.nifi.processors.hadoop.AbstractHadoopProcessor.HADOOP_FILE_URL_ATTRIBUTE;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertNotNull;
import static org.junit.jupiter.api.Assertions.assertTrue;
@@ -131,6 +132,7 @@ public class FetchParquetTest {
final MockFlowFile flowFile = testRunner.getFlowFilesForRelationship(FetchParquet.REL_SUCCESS).get(0);
flowFile.assertAttributeEquals(FetchParquet.RECORD_COUNT_ATTR, String.valueOf(USERS));
flowFile.assertAttributeEquals(CoreAttributes.MIME_TYPE.key(), "text/plain");
+ assertTrue(flowFile.getAttribute(HADOOP_FILE_URL_ATTRIBUTE).endsWith(DIRECTORY + "/" + parquetFile.getName()));
// the mock record writer will write the header for each record so replace those to get down to just the records
String flowFileContent = new String(flowFile.toByteArray(), StandardCharsets.UTF_8);
diff --git a/nifi-nar-bundles/nifi-parquet-bundle/nifi-parquet-processors/src/test/java/org/apache/nifi/processors/parquet/PutParquetTest.java b/nifi-nar-bundles/nifi-parquet-bundle/nifi-parquet-processors/src/test/java/org/apache/nifi/processors/parquet/PutParquetTest.java
index 302e36a5d9..befc898d4d 100644
--- a/nifi-nar-bundles/nifi-parquet-bundle/nifi-parquet-processors/src/test/java/org/apache/nifi/processors/parquet/PutParquetTest.java
+++ b/nifi-nar-bundles/nifi-parquet-bundle/nifi-parquet-processors/src/test/java/org/apache/nifi/processors/parquet/PutParquetTest.java
@@ -16,22 +16,6 @@
*/
package org.apache.nifi.processors.parquet;
-import static org.junit.jupiter.api.Assertions.assertEquals;
-import static org.junit.jupiter.api.Assertions.assertFalse;
-import static org.junit.jupiter.api.Assertions.assertNotNull;
-import static org.junit.jupiter.api.Assertions.assertTrue;
-import static org.mockito.ArgumentMatchers.any;
-import static org.mockito.Mockito.when;
-
-import java.io.File;
-import java.io.FileInputStream;
-import java.io.IOException;
-import java.io.InputStream;
-import java.nio.charset.StandardCharsets;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
-
import org.apache.avro.Schema;
import org.apache.avro.generic.GenericRecord;
import org.apache.commons.io.IOUtils;
@@ -42,10 +26,10 @@ import org.apache.nifi.avro.AvroTypeUtil;
import org.apache.nifi.flowfile.FlowFile;
import org.apache.nifi.flowfile.attributes.CoreAttributes;
import org.apache.nifi.logging.ComponentLog;
+import org.apache.nifi.parquet.utils.ParquetUtils;
import org.apache.nifi.processor.ProcessContext;
import org.apache.nifi.processors.hadoop.exception.FailureException;
import org.apache.nifi.processors.hadoop.record.HDFSRecordWriter;
-import org.apache.nifi.parquet.utils.ParquetUtils;
import org.apache.nifi.provenance.ProvenanceEventRecord;
import org.apache.nifi.provenance.ProvenanceEventType;
import org.apache.nifi.reporting.InitializationException;
@@ -70,6 +54,23 @@ import org.junit.jupiter.api.condition.DisabledOnOs;
import org.junit.jupiter.api.condition.OS;
import org.mockito.Mockito;
+import java.io.File;
+import java.io.FileInputStream;
+import java.io.IOException;
+import java.io.InputStream;
+import java.nio.charset.StandardCharsets;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import static org.apache.nifi.processors.hadoop.AbstractHadoopProcessor.HADOOP_FILE_URL_ATTRIBUTE;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.when;
+
@DisabledOnOs(OS.WINDOWS)
public class PutParquetTest {
@@ -135,6 +136,7 @@ public class PutParquetTest {
mockFlowFile.assertAttributeEquals(PutParquet.ABSOLUTE_HDFS_PATH_ATTRIBUTE, avroParquetFile.getParent().toString());
mockFlowFile.assertAttributeEquals(CoreAttributes.FILENAME.key(), filename);
mockFlowFile.assertAttributeEquals(PutParquet.RECORD_COUNT_ATTR, "100");
+ assertTrue(mockFlowFile.getAttribute(HADOOP_FILE_URL_ATTRIBUTE).endsWith(DIRECTORY + "/" + filename));
// verify we generated a provenance event
final List<ProvenanceEventRecord> provEvents = testRunner.getProvenanceEvents();
[nifi] 04/06: NIFI-11190: Added provenance events to GetHubSpot and GetShopify
Posted by tu...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
turcsanyi pushed a commit to branch support/nifi-1.x
in repository https://gitbox.apache.org/repos/asf/nifi.git
commit 9438eb8b86bde3506b92468f6bfa33d988531a93
Author: Lehel <Le...@hotmail.com>
AuthorDate: Thu Feb 16 17:28:19 2023 +0100
NIFI-11190: Added provenance events to GetHubSpot and GetShopify
This closes #6965.
Signed-off-by: Peter Turcsanyi <tu...@apache.org>
---
.../java/org/apache/nifi/processors/hubspot/GetHubSpot.java | 1 +
.../org/apache/nifi/processors/hubspot/GetHubSpotTest.java | 9 +++++++++
.../java/org/apache/nifi/processors/shopify/GetShopify.java | 3 +++
.../nifi/processors/shopify/rest/ShopifyRestService.java | 4 ++++
.../shopify/{GetShopifyIT.java => GetShopifyTest.java} | 12 ++++++++----
5 files changed, 25 insertions(+), 4 deletions(-)
diff --git a/nifi-nar-bundles/nifi-hubspot-bundle/nifi-hubspot-processors/src/main/java/org/apache/nifi/processors/hubspot/GetHubSpot.java b/nifi-nar-bundles/nifi-hubspot-bundle/nifi-hubspot-processors/src/main/java/org/apache/nifi/processors/hubspot/GetHubSpot.java
index 53567ff212..d624b58afe 100644
--- a/nifi-nar-bundles/nifi-hubspot-bundle/nifi-hubspot-processors/src/main/java/org/apache/nifi/processors/hubspot/GetHubSpot.java
+++ b/nifi-nar-bundles/nifi-hubspot-bundle/nifi-hubspot-processors/src/main/java/org/apache/nifi/processors/hubspot/GetHubSpot.java
@@ -239,6 +239,7 @@ public class GetHubSpot extends AbstractProcessor {
if (total.get() > 0) {
flowFile = session.putAttribute(flowFile, CoreAttributes.MIME_TYPE.key(), "application/json");
session.transfer(flowFile, REL_SUCCESS);
+ session.getProvenanceReporter().receive(flowFile, uri.toString());
} else {
getLogger().debug("Empty response when requested HubSpot endpoint: [{}]", endpoint);
context.yield();
diff --git a/nifi-nar-bundles/nifi-hubspot-bundle/nifi-hubspot-processors/src/test/java/org/apache/nifi/processors/hubspot/GetHubSpotTest.java b/nifi-nar-bundles/nifi-hubspot-bundle/nifi-hubspot-processors/src/test/java/org/apache/nifi/processors/hubspot/GetHubSpotTest.java
index fd8ed18fed..55f371acf6 100644
--- a/nifi-nar-bundles/nifi-hubspot-bundle/nifi-hubspot-processors/src/test/java/org/apache/nifi/processors/hubspot/GetHubSpotTest.java
+++ b/nifi-nar-bundles/nifi-hubspot-bundle/nifi-hubspot-processors/src/test/java/org/apache/nifi/processors/hubspot/GetHubSpotTest.java
@@ -45,6 +45,7 @@ import org.apache.nifi.components.PropertyDescriptor;
import org.apache.nifi.components.state.Scope;
import org.apache.nifi.flowfile.attributes.CoreAttributes;
import org.apache.nifi.processor.ProcessContext;
+import org.apache.nifi.provenance.ProvenanceEventRecord;
import org.apache.nifi.reporting.InitializationException;
import org.apache.nifi.util.MockFlowFile;
import org.apache.nifi.util.TestRunner;
@@ -109,6 +110,8 @@ class GetHubSpotTest {
flowFile.assertAttributeEquals(CoreAttributes.MIME_TYPE.key(), "application/json");
assertEquals(expectedJsonNode, actualJsonNode);
+ List<ProvenanceEventRecord> provenanceEvents = runner.getProvenanceEvents();
+ assertEquals(baseUrl.toString(), provenanceEvents.get(0).getTransitUri());
}
@Test
@@ -122,6 +125,7 @@ class GetHubSpotTest {
final List<MockFlowFile> flowFiles = runner.getFlowFilesForRelationship(GetHubSpot.REL_SUCCESS);
assertTrue(flowFiles.isEmpty());
+ assertTrue(runner.getProvenanceEvents().isEmpty());
}
@Test
@@ -131,6 +135,7 @@ class GetHubSpotTest {
server.enqueue(new MockResponse().setBody(response).setResponseCode(429));
assertThrows(AssertionError.class, () -> runner.run(1));
+ assertTrue(runner.getProvenanceEvents().isEmpty());
}
@Test
@@ -173,6 +178,8 @@ class GetHubSpotTest {
final String expectedJsonString = root.toString();
assertEquals(OBJECT_MAPPER.readTree(expectedJsonString), OBJECT_MAPPER.readTree(requestBodyString));
+ List<ProvenanceEventRecord> provenanceEvents = runner.getProvenanceEvents();
+ assertEquals(baseUrl.toString(), provenanceEvents.get(0).getTransitUri());
}
@Test
@@ -220,6 +227,8 @@ class GetHubSpotTest {
final String expectedJsonString = root.toString();
assertEquals(OBJECT_MAPPER.readTree(expectedJsonString), OBJECT_MAPPER.readTree(requestBodyString));
+ List<ProvenanceEventRecord> provenanceEvents = runner.getProvenanceEvents();
+ assertEquals(baseUrl.toString(), provenanceEvents.get(0).getTransitUri());
}
static class MockGetHubSpot extends GetHubSpot {
diff --git a/nifi-nar-bundles/nifi-shopify-bundle/nifi-shopify-processors/src/main/java/org/apache/nifi/processors/shopify/GetShopify.java b/nifi-nar-bundles/nifi-shopify-bundle/nifi-shopify-processors/src/main/java/org/apache/nifi/processors/shopify/GetShopify.java
index 9994ab999a..f0aee1514d 100644
--- a/nifi-nar-bundles/nifi-shopify-bundle/nifi-shopify-processors/src/main/java/org/apache/nifi/processors/shopify/GetShopify.java
+++ b/nifi-nar-bundles/nifi-shopify-bundle/nifi-shopify-processors/src/main/java/org/apache/nifi/processors/shopify/GetShopify.java
@@ -327,6 +327,7 @@ public class GetShopify extends AbstractProcessor {
if (objectCountHolder.get() > 0) {
flowFile = session.putAttribute(flowFile, CoreAttributes.MIME_TYPE.key(), "application/json");
session.transfer(flowFile, REL_SUCCESS);
+ session.getProvenanceReporter().receive(flowFile, shopifyRestService.getBaseUriString());
} else {
getLogger().debug("Empty response when requested Shopify resource: [{}]", resourceName);
session.remove(flowFile);
@@ -415,4 +416,6 @@ public class GetShopify extends AbstractProcessor {
throw new ProcessException("Clearing state failed", e);
}
}
+
+
}
diff --git a/nifi-nar-bundles/nifi-shopify-bundle/nifi-shopify-processors/src/main/java/org/apache/nifi/processors/shopify/rest/ShopifyRestService.java b/nifi-nar-bundles/nifi-shopify-bundle/nifi-shopify-processors/src/main/java/org/apache/nifi/processors/shopify/rest/ShopifyRestService.java
index 5bc7ad2f88..f9a8fc1b6a 100644
--- a/nifi-nar-bundles/nifi-shopify-bundle/nifi-shopify-processors/src/main/java/org/apache/nifi/processors/shopify/rest/ShopifyRestService.java
+++ b/nifi-nar-bundles/nifi-shopify-bundle/nifi-shopify-processors/src/main/java/org/apache/nifi/processors/shopify/rest/ShopifyRestService.java
@@ -76,6 +76,10 @@ public class ShopifyRestService {
return retrieveResponse(uri);
}
+ public String getBaseUriString() {
+ return getBaseUri().build().toString();
+ }
+
private HttpResponseEntity retrieveResponse(URI uri) {
return webClientServiceProvider.getWebClientService()
.get()
diff --git a/nifi-nar-bundles/nifi-shopify-bundle/nifi-shopify-processors/src/test/java/org/apache/nifi/processors/shopify/GetShopifyIT.java b/nifi-nar-bundles/nifi-shopify-bundle/nifi-shopify-processors/src/test/java/org/apache/nifi/processors/shopify/GetShopifyTest.java
similarity index 92%
rename from nifi-nar-bundles/nifi-shopify-bundle/nifi-shopify-processors/src/test/java/org/apache/nifi/processors/shopify/GetShopifyIT.java
rename to nifi-nar-bundles/nifi-shopify-bundle/nifi-shopify-processors/src/test/java/org/apache/nifi/processors/shopify/GetShopifyTest.java
index 73ed2a916b..a6f526fe82 100644
--- a/nifi-nar-bundles/nifi-shopify-bundle/nifi-shopify-processors/src/test/java/org/apache/nifi/processors/shopify/GetShopifyIT.java
+++ b/nifi-nar-bundles/nifi-shopify-bundle/nifi-shopify-processors/src/test/java/org/apache/nifi/processors/shopify/GetShopifyTest.java
@@ -24,6 +24,7 @@ import org.apache.nifi.flowfile.attributes.CoreAttributes;
import org.apache.nifi.processors.shopify.model.IncrementalLoadingParameter;
import org.apache.nifi.processors.shopify.model.ResourceType;
import org.apache.nifi.processors.shopify.rest.ShopifyRestService;
+import org.apache.nifi.provenance.ProvenanceEventRecord;
import org.apache.nifi.reporting.InitializationException;
import org.apache.nifi.util.MockFlowFile;
import org.apache.nifi.util.MockProcessContext;
@@ -39,7 +40,6 @@ import org.junit.jupiter.api.Test;
import java.io.IOException;
import java.nio.charset.StandardCharsets;
-import java.time.Instant;
import java.util.List;
import static org.junit.jupiter.api.Assertions.assertEquals;
@@ -49,7 +49,7 @@ import static org.mockito.ArgumentMatchers.any;
import static org.mockito.Mockito.spy;
import static org.mockito.Mockito.verify;
-class GetShopifyIT {
+class GetShopifyTest {
private static final String BASE_URL = "/test/shopify";
private static MockWebServer server;
@@ -101,6 +101,8 @@ class GetShopifyIT {
List<MockFlowFile> flowFiles = runner.getFlowFilesForRelationship(GetShopify.REL_SUCCESS);
flowFiles.get(0).assertAttributeEquals(CoreAttributes.MIME_TYPE.key(), "application/json");
+ List<ProvenanceEventRecord> provenanceEvents = runner.getProvenanceEvents();
+ assertEquals(baseUrl.toString(), provenanceEvents.get(0).getTransitUri());
}
@Test
@@ -124,6 +126,7 @@ class GetShopifyIT {
assertThrows(AssertionError.class, () -> runner.run(1));
assertTrue(((MockProcessContext) runner.getProcessContext()).isYieldCalled(),
"In case of 429 the processor should be yielded.");
+ assertTrue(runner.getProvenanceEvents().isEmpty());
}
@Test
@@ -148,6 +151,7 @@ class GetShopifyIT {
assertTrue(((MockProcessContext) runner.getProcessContext()).isYieldCalled(),
"In case of 404 the processor should be yielded.");
+ assertTrue(runner.getProvenanceEvents().isEmpty());
}
@Test
@@ -157,8 +161,6 @@ class GetShopifyIT {
.setBody(getResourceAsString("collection_listings.json"));
server.enqueue(mockResponse);
- final Instant expectedExecutionTime = Instant.parse("2022-08-16T10:15:30Z");
-
final StandardWebClientServiceProvider standardWebClientServiceProvider =
new StandardWebClientServiceProvider();
final CustomGetShopify customGetShopify = new CustomGetShopify();
@@ -177,6 +179,8 @@ class GetShopifyIT {
List<MockFlowFile> flowFilesForRelationship = runner.getFlowFilesForRelationship(GetShopify.REL_SUCCESS);
assertEquals(1, flowFilesForRelationship.size());
+ List<ProvenanceEventRecord> provenanceEvents = runner.getProvenanceEvents();
+ assertEquals(baseUrl.toString(), provenanceEvents.get(0).getTransitUri());
}
static class CustomGetShopify extends GetShopify {
[nifi] 02/06: NIFI-11146 Proxy usage in BoxFile processors
Posted by tu...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
turcsanyi pushed a commit to branch support/nifi-1.x
in repository https://gitbox.apache.org/repos/asf/nifi.git
commit 20914c1af913919d89a07fe03a11816b38c70f41
Author: krisztina-zsihovszki <zs...@gmail.com>
AuthorDate: Mon Feb 6 18:17:33 2023 +0100
NIFI-11146 Proxy usage in BoxFile processors
This closes #6932.
Signed-off-by: Peter Turcsanyi <tu...@apache.org>
---
.../nifi-box-bundle/nifi-box-services-api/pom.xml | 2 +-
.../nifi-box-bundle/nifi-box-services/pom.xml | 4 ++
.../JsonConfigBasedBoxClientService.java | 44 ++++++++++++++--------
3 files changed, 34 insertions(+), 16 deletions(-)
diff --git a/nifi-nar-bundles/nifi-box-bundle/nifi-box-services-api/pom.xml b/nifi-nar-bundles/nifi-box-bundle/nifi-box-services-api/pom.xml
index ec329c64b2..67f332e44f 100644
--- a/nifi-nar-bundles/nifi-box-bundle/nifi-box-services-api/pom.xml
+++ b/nifi-nar-bundles/nifi-box-bundle/nifi-box-services-api/pom.xml
@@ -29,7 +29,7 @@
<dependency>
<groupId>com.box</groupId>
<artifactId>box-java-sdk</artifactId>
- <version>3.8.2</version>
+ <version>4.0.0</version>
<exclusions>
<exclusion>
<groupId>org.bouncycastle</groupId>
diff --git a/nifi-nar-bundles/nifi-box-bundle/nifi-box-services/pom.xml b/nifi-nar-bundles/nifi-box-bundle/nifi-box-services/pom.xml
index 90cadf875a..0953480005 100644
--- a/nifi-nar-bundles/nifi-box-bundle/nifi-box-services/pom.xml
+++ b/nifi-nar-bundles/nifi-box-bundle/nifi-box-services/pom.xml
@@ -42,6 +42,10 @@
<artifactId>nifi-json-utils</artifactId>
<version>1.21.0-SNAPSHOT</version>
</dependency>
+ <dependency>
+ <groupId>org.apache.nifi</groupId>
+ <artifactId>nifi-proxy-configuration-api</artifactId>
+ </dependency>
<dependency>
<groupId>org.apache.nifi</groupId>
<artifactId>nifi-mock</artifactId>
diff --git a/nifi-nar-bundles/nifi-box-bundle/nifi-box-services/src/main/java/org/apache/nifi/box/controllerservices/JsonConfigBasedBoxClientService.java b/nifi-nar-bundles/nifi-box-bundle/nifi-box-services/src/main/java/org/apache/nifi/box/controllerservices/JsonConfigBasedBoxClientService.java
index f384b015c3..26662bf28c 100644
--- a/nifi-nar-bundles/nifi-box-bundle/nifi-box-services/src/main/java/org/apache/nifi/box/controllerservices/JsonConfigBasedBoxClientService.java
+++ b/nifi-nar-bundles/nifi-box-bundle/nifi-box-services/src/main/java/org/apache/nifi/box/controllerservices/JsonConfigBasedBoxClientService.java
@@ -19,6 +19,16 @@ package org.apache.nifi.box.controllerservices;
import com.box.sdk.BoxAPIConnection;
import com.box.sdk.BoxConfig;
import com.box.sdk.BoxDeveloperEditionAPIConnection;
+import java.io.FileNotFoundException;
+import java.io.FileReader;
+import java.io.IOException;
+import java.io.Reader;
+import java.net.Proxy;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.List;
import org.apache.nifi.annotation.documentation.CapabilityDescription;
import org.apache.nifi.annotation.documentation.Tags;
import org.apache.nifi.annotation.lifecycle.OnEnabled;
@@ -33,16 +43,9 @@ import org.apache.nifi.expression.ExpressionLanguageScope;
import org.apache.nifi.processor.exception.ProcessException;
import org.apache.nifi.processor.util.JsonValidator;
import org.apache.nifi.processor.util.StandardValidators;
+import org.apache.nifi.proxy.ProxyConfiguration;
+import org.apache.nifi.proxy.ProxySpec;
-import java.io.FileNotFoundException;
-import java.io.FileReader;
-import java.io.IOException;
-import java.io.Reader;
-import java.util.ArrayList;
-import java.util.Arrays;
-import java.util.Collection;
-import java.util.Collections;
-import java.util.List;
@CapabilityDescription("Provides Box client objects through which Box API calls can be used.")
@Tags({"box", "client", "provider"})
@@ -75,10 +78,13 @@ public class JsonConfigBasedBoxClientService extends AbstractControllerService i
.expressionLanguageSupported(ExpressionLanguageScope.VARIABLE_REGISTRY)
.build();
+ private static final ProxySpec[] PROXY_SPECS = {ProxySpec.HTTP, ProxySpec.HTTP_AUTH};
+
private static final List<PropertyDescriptor> PROPERTIES = Collections.unmodifiableList(Arrays.asList(
ACCOUNT_ID,
APP_CONFIG_FILE,
- APP_CONFIG_JSON
+ APP_CONFIG_JSON,
+ ProxyConfiguration.createProxyConfigPropertyDescriptor(false, PROXY_SPECS)
));
private volatile BoxAPIConnection boxAPIConnection;
@@ -128,15 +134,16 @@ public class JsonConfigBasedBoxClientService extends AbstractControllerService i
}
private BoxAPIConnection createBoxApiConnection(ConfigurationContext context) {
- BoxAPIConnection api;
+ final BoxAPIConnection api;
- String accountId = context.getProperty(ACCOUNT_ID).evaluateAttributeExpressions().getValue();
+ final String accountId = context.getProperty(ACCOUNT_ID).evaluateAttributeExpressions().getValue();
+ final ProxyConfiguration proxyConfiguration = ProxyConfiguration.getConfiguration(context);
- BoxConfig boxConfig;
+ final BoxConfig boxConfig;
if (context.getProperty(APP_CONFIG_FILE).isSet()) {
String appConfigFile = context.getProperty(APP_CONFIG_FILE).evaluateAttributeExpressions().getValue();
try (
- Reader reader = new FileReader(appConfigFile);
+ Reader reader = new FileReader(appConfigFile)
) {
boxConfig = BoxConfig.readFrom(reader);
} catch (FileNotFoundException e) {
@@ -145,7 +152,7 @@ public class JsonConfigBasedBoxClientService extends AbstractControllerService i
throw new ProcessException("Couldn't read Box config file", e);
}
} else {
- String appConfig = context.getProperty(APP_CONFIG_JSON).evaluateAttributeExpressions().getValue();
+ final String appConfig = context.getProperty(APP_CONFIG_JSON).evaluateAttributeExpressions().getValue();
boxConfig = BoxConfig.readFrom(appConfig);
}
@@ -153,6 +160,13 @@ public class JsonConfigBasedBoxClientService extends AbstractControllerService i
api.asUser(accountId);
+ if (!Proxy.Type.DIRECT.equals(proxyConfiguration.getProxyType())) {
+ api.setProxy(proxyConfiguration.createProxy());
+
+ if (proxyConfiguration.hasCredential()) {
+ api.setProxyBasicAuthentication(proxyConfiguration.getProxyUserName(), proxyConfiguration.getProxyUserPassword());
+ }
+ }
return api;
}
}
[nifi] 06/06: NIFI-11158 PutSalesforceObject processor improvements
Posted by tu...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
turcsanyi pushed a commit to branch support/nifi-1.x
in repository https://gitbox.apache.org/repos/asf/nifi.git
commit 1273152d781b754d0ba0a57506fa44783bb65d1d
Author: krisztina-zsihovszki <zs...@gmail.com>
AuthorDate: Wed Feb 15 12:37:31 2023 +0100
NIFI-11158 PutSalesforceObject processor improvements
This closes #6959.
Reviewed-by: Lehel <le...@hotmail.com>
Reviewed-by: Mark Bathori <ba...@gmail.com>
Signed-off-by: Peter Turcsanyi <tu...@apache.org>
---
.../nifi/util/StandardProcessorTestRunner.java | 13 ++++-
.../main/java/org/apache/nifi/util/TestRunner.java | 7 +++
.../processors/salesforce/PutSalesforceObject.java | 34 ++++++++---
.../salesforce/QuerySalesforceObject.java | 8 ++-
.../processors/salesforce/util/RecordExtender.java | 4 +-
.../salesforce/util/SalesforceRestService.java | 15 ++++-
.../salesforce/PutSalesforceObjectIT.java | 67 +++++++++++++++++++---
.../salesforce/QuerySalesforceObjectIT.java | 2 +
8 files changed, 126 insertions(+), 24 deletions(-)
diff --git a/nifi-mock/src/main/java/org/apache/nifi/util/StandardProcessorTestRunner.java b/nifi-mock/src/main/java/org/apache/nifi/util/StandardProcessorTestRunner.java
index b2b0557dad..99bec62b9e 100644
--- a/nifi-mock/src/main/java/org/apache/nifi/util/StandardProcessorTestRunner.java
+++ b/nifi-mock/src/main/java/org/apache/nifi/util/StandardProcessorTestRunner.java
@@ -40,6 +40,7 @@ import org.apache.nifi.processor.ProcessSessionFactory;
import org.apache.nifi.processor.Processor;
import org.apache.nifi.processor.Relationship;
import org.apache.nifi.provenance.ProvenanceEventRecord;
+import org.apache.nifi.provenance.ProvenanceEventType;
import org.apache.nifi.registry.VariableDescriptor;
import org.apache.nifi.reporting.InitializationException;
import org.apache.nifi.state.MockStateManager;
@@ -72,6 +73,7 @@ import java.util.function.Predicate;
import java.util.stream.Collectors;
import static java.util.Objects.requireNonNull;
+import static java.util.stream.Collectors.toSet;
import static org.junit.jupiter.api.Assertions.assertEquals;
public class StandardProcessorTestRunner implements TestRunner {
@@ -367,7 +369,7 @@ public class StandardProcessorTestRunner implements TestRunner {
.collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue))
)
- .collect(Collectors.toSet());
+ .collect(toSet());
assertEquals(expectedAttributes, actualAttributes);
}
@@ -1056,4 +1058,13 @@ public class StandardProcessorTestRunner implements TestRunner {
public void setRunSchedule(long runSchedule) {
this.runSchedule = runSchedule;
}
+
+ @Override
+ public void assertProvenanceEvent(final ProvenanceEventType eventType) {
+ Set<ProvenanceEventType> expectedEventTypes = Collections.singleton(eventType);
+ Set<ProvenanceEventType> actualEventTypes = getProvenanceEvents().stream()
+ .map(ProvenanceEventRecord::getEventType)
+ .collect(toSet());
+ assertEquals(expectedEventTypes, actualEventTypes);
+ }
}
diff --git a/nifi-mock/src/main/java/org/apache/nifi/util/TestRunner.java b/nifi-mock/src/main/java/org/apache/nifi/util/TestRunner.java
index 7b01ed9709..dbe31e73c8 100644
--- a/nifi-mock/src/main/java/org/apache/nifi/util/TestRunner.java
+++ b/nifi-mock/src/main/java/org/apache/nifi/util/TestRunner.java
@@ -28,6 +28,7 @@ import org.apache.nifi.processor.ProcessSessionFactory;
import org.apache.nifi.processor.Processor;
import org.apache.nifi.processor.Relationship;
import org.apache.nifi.provenance.ProvenanceEventRecord;
+import org.apache.nifi.provenance.ProvenanceEventType;
import org.apache.nifi.reporting.InitializationException;
import org.apache.nifi.state.MockStateManager;
@@ -1062,4 +1063,10 @@ public interface TestRunner {
*/
void setRunSchedule(long runSchedule);
+ /**
+ * Assert that provenance event was created with the specified event type.
+ *
+ * @param eventType Provenance event type
+ */
+ void assertProvenanceEvent(ProvenanceEventType eventType);
}
diff --git a/nifi-nar-bundles/nifi-salesforce-bundle/nifi-salesforce-processors/src/main/java/org/apache/nifi/processors/salesforce/PutSalesforceObject.java b/nifi-nar-bundles/nifi-salesforce-bundle/nifi-salesforce-processors/src/main/java/org/apache/nifi/processors/salesforce/PutSalesforceObject.java
index 00d0c2b783..83605800f9 100644
--- a/nifi-nar-bundles/nifi-salesforce-bundle/nifi-salesforce-processors/src/main/java/org/apache/nifi/processors/salesforce/PutSalesforceObject.java
+++ b/nifi-nar-bundles/nifi-salesforce-bundle/nifi-salesforce-processors/src/main/java/org/apache/nifi/processors/salesforce/PutSalesforceObject.java
@@ -20,7 +20,9 @@ import com.fasterxml.jackson.databind.node.ObjectNode;
import org.apache.nifi.NullSuppression;
import org.apache.nifi.annotation.behavior.InputRequirement;
import org.apache.nifi.annotation.behavior.ReadsAttribute;
+import org.apache.nifi.annotation.behavior.WritesAttribute;
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.annotation.lifecycle.OnScheduled;
import org.apache.nifi.components.PropertyDescriptor;
@@ -64,9 +66,13 @@ import static org.apache.nifi.processors.salesforce.util.CommonSalesforcePropert
@CapabilityDescription("Creates new records for the specified Salesforce sObject. The type of the Salesforce object must be set in the input flowfile's"
+ " 'objectType' attribute. This processor cannot update existing records.")
@ReadsAttribute(attribute = "objectType", description = "The Salesforce object type to upload records to. E.g. Account, Contact, Campaign.")
+@WritesAttribute(attribute = "error.message", description = "The error message returned by Salesforce.")
+@SeeAlso(QuerySalesforceObject.class)
public class PutSalesforceObject extends AbstractProcessor {
private static final int MAX_RECORD_COUNT = 200;
+ private static final String ATTR_OBJECT_TYPE = "objectType";
+ private static final String ATTR_ERROR_MESSAGE = "error.message";
protected static final PropertyDescriptor RECORD_READER_FACTORY = new PropertyDescriptor.Builder()
.name("record-reader")
@@ -138,15 +144,19 @@ public class PutSalesforceObject extends AbstractProcessor {
return;
}
- String objectType = flowFile.getAttribute("objectType");
+ String objectType = flowFile.getAttribute(ATTR_OBJECT_TYPE);
if (objectType == null) {
- throw new ProcessException("Salesforce object type not found among the incoming flowfile attributes");
+ getLogger().error("Salesforce object type not found among the incoming FlowFile attributes");
+ flowFile = session.putAttribute(flowFile, ATTR_ERROR_MESSAGE, "Salesforce object type not found among FlowFile attributes");
+ session.transfer(session.penalize(flowFile), REL_FAILURE);
+ return;
}
RecordReaderFactory readerFactory = context.getProperty(RECORD_READER_FACTORY).asControllerService(RecordReaderFactory.class);
RecordExtender extender;
-
+ long startNanos = System.nanoTime();
+ try {
try (InputStream in = session.read(flowFile);
RecordReader reader = readerFactory.createRecordReader(flowFile, in, getLogger());
ByteArrayOutputStream out = new ByteArrayOutputStream();
@@ -170,24 +180,30 @@ public class PutSalesforceObject extends AbstractProcessor {
out.reset();
}
}
-
if (writer.isActiveRecordSet()) {
processRecords(objectType, out, writer, extender);
}
- session.transfer(flowFile, REL_SUCCESS);
-
+ }
+ session.transfer(flowFile, REL_SUCCESS);
+ long transferMillis = TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - startNanos);
+ session.getProvenanceReporter().send(flowFile, salesforceRestService.getVersionedBaseUrl()+ "/composite/tree/" + objectType, transferMillis);
} catch (MalformedRecordException e) {
getLogger().error("Couldn't read records from input", e);
- session.transfer(flowFile, REL_FAILURE);
+ transferToFailure(session, flowFile, e);
} catch (SchemaNotFoundException e) {
getLogger().error("Couldn't create record writer", e);
- session.transfer(flowFile, REL_FAILURE);
+ transferToFailure(session, flowFile, e);
} catch (Exception e) {
getLogger().error("Failed to put records to Salesforce.", e);
- session.transfer(flowFile, REL_FAILURE);
+ transferToFailure(session, flowFile, e);
}
}
+ private void transferToFailure(ProcessSession session, FlowFile flowFile, Exception e) {
+ flowFile = session.putAttribute(flowFile, ATTR_ERROR_MESSAGE, e.getMessage());
+ session.transfer(session.penalize(flowFile), REL_FAILURE);
+ }
+
private void processRecords(String objectType, ByteArrayOutputStream out, WriteJsonResult writer, RecordExtender extender) throws IOException {
writer.finishRecordSet();
writer.flush();
diff --git a/nifi-nar-bundles/nifi-salesforce-bundle/nifi-salesforce-processors/src/main/java/org/apache/nifi/processors/salesforce/QuerySalesforceObject.java b/nifi-nar-bundles/nifi-salesforce-bundle/nifi-salesforce-processors/src/main/java/org/apache/nifi/processors/salesforce/QuerySalesforceObject.java
index 8235375d69..b5125a9eb1 100644
--- a/nifi-nar-bundles/nifi-salesforce-bundle/nifi-salesforce-processors/src/main/java/org/apache/nifi/processors/salesforce/QuerySalesforceObject.java
+++ b/nifi-nar-bundles/nifi-salesforce-bundle/nifi-salesforce-processors/src/main/java/org/apache/nifi/processors/salesforce/QuerySalesforceObject.java
@@ -26,6 +26,7 @@ import org.apache.nifi.annotation.behavior.WritesAttribute;
import org.apache.nifi.annotation.behavior.WritesAttributes;
import org.apache.nifi.annotation.configuration.DefaultSchedule;
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.annotation.lifecycle.OnScheduled;
import org.apache.nifi.components.PropertyDescriptor;
@@ -104,6 +105,7 @@ import static org.apache.nifi.processors.salesforce.util.CommonSalesforcePropert
@WritesAttribute(attribute = "record.count", description = "Sets the number of records in the FlowFile.")
})
@DefaultSchedule(strategy = SchedulingStrategy.TIMER_DRIVEN, period = "1 min")
+@SeeAlso(PutSalesforceObject.class)
public class QuerySalesforceObject extends AbstractProcessor {
static final PropertyDescriptor SOBJECT_NAME = new PropertyDescriptor.Builder()
@@ -314,7 +316,7 @@ public class QuerySalesforceObject extends AbstractProcessor {
Map<String, String> attributes = new HashMap<>();
AtomicInteger recordCountHolder = new AtomicInteger();
-
+ long startNanos = System.nanoTime();
flowFile = session.write(flowFile, out -> {
try (
InputStream querySObjectResultInputStream = getResultInputStream(nextRecordsUrl, querySObject);
@@ -381,6 +383,10 @@ public class QuerySalesforceObject extends AbstractProcessor {
flowFile = session.putAllAttributes(flowFile, attributes);
session.transfer(flowFile, REL_SUCCESS);
+ long transferMillis = TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - startNanos);
+ session.getProvenanceReporter().receive(flowFile, salesforceRestService.getVersionedBaseUrl() + "/composite/tree/" + sObject,
+ transferMillis);
+
session.adjustCounter("Records Processed", recordCount, false);
getLogger().info("Successfully written {} records for {}", recordCount, flowFile);
}
diff --git a/nifi-nar-bundles/nifi-salesforce-bundle/nifi-salesforce-processors/src/main/java/org/apache/nifi/processors/salesforce/util/RecordExtender.java b/nifi-nar-bundles/nifi-salesforce-bundle/nifi-salesforce-processors/src/main/java/org/apache/nifi/processors/salesforce/util/RecordExtender.java
index 9e8b03e9d5..6af98cd725 100644
--- a/nifi-nar-bundles/nifi-salesforce-bundle/nifi-salesforce-processors/src/main/java/org/apache/nifi/processors/salesforce/util/RecordExtender.java
+++ b/nifi-nar-bundles/nifi-salesforce-bundle/nifi-salesforce-processors/src/main/java/org/apache/nifi/processors/salesforce/util/RecordExtender.java
@@ -66,7 +66,9 @@ public class RecordExtender {
public MapRecord getExtendedRecord(String objectType, int count, Record record) {
- Set<String> rawFieldNames = record.getRawFieldNames();
+ Set<String> rawFieldNames = record.getRawFieldNames().stream()
+ .filter(fieldName -> record.getValue(fieldName) != null)
+ .collect(Collectors.toSet());
Map<String, Object> objectMap = rawFieldNames.stream()
.collect(Collectors.toMap(Function.identity(), record::getValue));
diff --git a/nifi-nar-bundles/nifi-salesforce-bundle/nifi-salesforce-processors/src/main/java/org/apache/nifi/processors/salesforce/util/SalesforceRestService.java b/nifi-nar-bundles/nifi-salesforce-bundle/nifi-salesforce-processors/src/main/java/org/apache/nifi/processors/salesforce/util/SalesforceRestService.java
index cc3f4b2bf8..0affbc441e 100644
--- a/nifi-nar-bundles/nifi-salesforce-bundle/nifi-salesforce-processors/src/main/java/org/apache/nifi/processors/salesforce/util/SalesforceRestService.java
+++ b/nifi-nar-bundles/nifi-salesforce-bundle/nifi-salesforce-processors/src/main/java/org/apache/nifi/processors/salesforce/util/SalesforceRestService.java
@@ -44,7 +44,7 @@ public class SalesforceRestService {
}
public InputStream describeSObject(String sObject) {
- String url = baseUrl + "/services/data/v" + version + "/sobjects/" + sObject + "/describe?maxRecords=1";
+ String url = getVersionedBaseUrl() + "/sobjects/" + sObject + "/describe?maxRecords=1";
Request request = new Request.Builder()
.addHeader("Authorization", "Bearer " + accessTokenProvider.get())
@@ -56,7 +56,7 @@ public class SalesforceRestService {
}
public InputStream query(String query) {
- String url = baseUrl + "/services/data/v" + version + "/query";
+ String url = getVersionedBaseUrl() + "/query";
HttpUrl httpUrl = HttpUrl.get(url).newBuilder()
.addQueryParameter("q", query)
@@ -87,7 +87,7 @@ public class SalesforceRestService {
}
public InputStream postRecord(String sObjectApiName, String body) {
- String url = baseUrl + "/services/data/v" + version + "/composite/tree/" + sObjectApiName;
+ String url = getVersionedBaseUrl() + "/composite/tree/" + sObjectApiName;
HttpUrl httpUrl = HttpUrl.get(url).newBuilder()
.build();
@@ -103,6 +103,10 @@ public class SalesforceRestService {
return request(request);
}
+ public String getVersionedBaseUrl() {
+ return baseUrl + "/services/data/v" + version;
+ }
+
private InputStream request(Request request) {
Response response = null;
try {
@@ -115,6 +119,11 @@ public class SalesforceRestService {
);
}
return response.body().byteStream();
+ } catch (ProcessException e) {
+ if (response != null) {
+ response.close();
+ }
+ throw e;
} catch (Exception e) {
if (response != null) {
response.close();
diff --git a/nifi-nar-bundles/nifi-salesforce-bundle/nifi-salesforce-processors/src/test/java/org/apache/nifi/processors/salesforce/PutSalesforceObjectIT.java b/nifi-nar-bundles/nifi-salesforce-bundle/nifi-salesforce-processors/src/test/java/org/apache/nifi/processors/salesforce/PutSalesforceObjectIT.java
index 390fa8ac0f..2c296a119f 100644
--- a/nifi-nar-bundles/nifi-salesforce-bundle/nifi-salesforce-processors/src/test/java/org/apache/nifi/processors/salesforce/PutSalesforceObjectIT.java
+++ b/nifi-nar-bundles/nifi-salesforce-bundle/nifi-salesforce-processors/src/test/java/org/apache/nifi/processors/salesforce/PutSalesforceObjectIT.java
@@ -20,6 +20,8 @@ import org.apache.nifi.oauth2.StandardOauth2AccessTokenProvider;
import org.apache.nifi.processor.Processor;
import org.apache.nifi.processors.salesforce.util.CommonSalesforceProperties;
import org.apache.nifi.processors.salesforce.util.SalesforceConfigAware;
+import org.apache.nifi.provenance.ProvenanceEventType;
+import org.apache.nifi.reporting.InitializationException;
import org.apache.nifi.serialization.record.MockRecordParser;
import org.apache.nifi.serialization.record.RecordFieldType;
import org.apache.nifi.util.MockFlowFile;
@@ -32,6 +34,7 @@ import java.util.Collections;
import java.util.List;
import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
class PutSalesforceObjectIT implements SalesforceConfigAware {
@@ -62,25 +65,71 @@ class PutSalesforceObjectIT implements SalesforceConfigAware {
reader.addSchemaField("numberOfEmployees", RecordFieldType.STRING);
reader.addSchemaField("industry", RecordFieldType.STRING);
- reader.addRecord("SampleAccount1", "111111", "www.salesforce1.com", "100", "Banking");
- reader.addRecord("SampleAccount2", "222222", "www.salesforce2.com", "200", "Banking");
+ reader.addRecord("SampleAccount1", "111111", "www.salesforce1.com", null, "Banking");
+ reader.addRecord("SampleAccount2", null, "www.salesforce2.com", "200", "Banking");
reader.addRecord("SampleAccount3", "333333", "www.salesforce3.com", "300", "Banking");
- reader.addRecord("SampleAccount4", "444444", "www.salesforce4.com", "400", "Banking");
+ reader.addRecord("SampleAccount4", "444444", null, "400", "Banking");
reader.addRecord("SampleAccount5", "555555", "www.salesforce5.com", "500", "Banking");
runner.enqueue("", Collections.singletonMap("objectType", "Account"));
- runner.addControllerService("reader", reader);
- runner.enableControllerService(reader);
-
- runner.setProperty(CommonSalesforceProperties.API_VERSION, VERSION);
- runner.setProperty(CommonSalesforceProperties.API_URL, BASE_URL);
- runner.setProperty(PutSalesforceObject.RECORD_READER_FACTORY, reader.getIdentifier());
+ configureProcessor(reader);
runner.run();
List<MockFlowFile> results = runner.getFlowFilesForRelationship(PutSalesforceObject.REL_SUCCESS);
assertEquals(1, results.size());
+
+ runner.assertProvenanceEvent(ProvenanceEventType.SEND);
+ }
+
+ @Test
+ void testMissingObjectType() throws Exception {
+ MockRecordParser reader = new MockRecordParser();
+
+ runner.enqueue("");
+
+ configureProcessor(reader);
+
+ runner.run();
+
+ List<MockFlowFile> results = runner.getFlowFilesForRelationship(PutSalesforceObject.REL_FAILURE);
+ assertEquals(1, results.size());
+ assertTrue(runner.getProvenanceEvents().isEmpty());
+
+ List<MockFlowFile> flowFiles = runner.getFlowFilesForRelationship(PutSalesforceObject.REL_FAILURE);
+ MockFlowFile ff0 = flowFiles.get(0);
+ ff0.assertAttributeExists("error.message");
+ }
+
+ @Test
+ void testErrorForInvalidRecordField() throws Exception {
+ MockRecordParser reader = new MockRecordParser();
+ reader.addSchemaField("invalidField", RecordFieldType.STRING);
+ reader.addRecord("invalidField");
+
+ runner.enqueue("", Collections.singletonMap("objectType", "Account"));
+
+ configureProcessor(reader);
+
+ runner.run();
+
+ List<MockFlowFile> results = runner.getFlowFilesForRelationship(PutSalesforceObject.REL_FAILURE);
+ assertEquals(1, results.size());
+ assertTrue(runner.getProvenanceEvents().isEmpty());
+
+ List<MockFlowFile> flowFiles = runner.getFlowFilesForRelationship(PutSalesforceObject.REL_FAILURE);
+ MockFlowFile ff0 = flowFiles.get(0);
+ ff0.assertAttributeExists("error.message");
+ }
+
+ private void configureProcessor(final MockRecordParser reader) throws InitializationException {
+ runner.addControllerService("reader", reader);
+ runner.enableControllerService(reader);
+
+ runner.setProperty(CommonSalesforceProperties.API_VERSION, VERSION);
+ runner.setProperty(CommonSalesforceProperties.API_URL, BASE_URL);
+ runner.setProperty(PutSalesforceObject.RECORD_READER_FACTORY, reader.getIdentifier());
}
}
diff --git a/nifi-nar-bundles/nifi-salesforce-bundle/nifi-salesforce-processors/src/test/java/org/apache/nifi/processors/salesforce/QuerySalesforceObjectIT.java b/nifi-nar-bundles/nifi-salesforce-bundle/nifi-salesforce-processors/src/test/java/org/apache/nifi/processors/salesforce/QuerySalesforceObjectIT.java
index f3f5bd50eb..661df2673a 100644
--- a/nifi-nar-bundles/nifi-salesforce-bundle/nifi-salesforce-processors/src/test/java/org/apache/nifi/processors/salesforce/QuerySalesforceObjectIT.java
+++ b/nifi-nar-bundles/nifi-salesforce-bundle/nifi-salesforce-processors/src/test/java/org/apache/nifi/processors/salesforce/QuerySalesforceObjectIT.java
@@ -20,6 +20,7 @@ import org.apache.nifi.oauth2.StandardOauth2AccessTokenProvider;
import org.apache.nifi.processor.Processor;
import org.apache.nifi.processors.salesforce.util.CommonSalesforceProperties;
import org.apache.nifi.processors.salesforce.util.SalesforceConfigAware;
+import org.apache.nifi.provenance.ProvenanceEventType;
import org.apache.nifi.serialization.RecordSetWriterFactory;
import org.apache.nifi.serialization.record.MockRecordWriter;
import org.apache.nifi.util.MockFlowFile;
@@ -76,5 +77,6 @@ class QuerySalesforceObjectIT implements SalesforceConfigAware {
List<MockFlowFile> results = runner.getFlowFilesForRelationship(QuerySalesforceObject.REL_SUCCESS);
assertNotNull(results.get(0).getContent());
+ runner.assertProvenanceEvent(ProvenanceEventType.RECEIVE);
}
}