This is an automated email from the ASF dual-hosted git repository.
pvillard31 pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/nifi.git
The following commit(s) were added to refs/heads/main by this push:
new 798809a0408 NIFI-14181 Allow FlowFile attributes in
QuerySalesforceObject sObject Name and Field Names (#11756)
798809a0408 is described below
commit 798809a04082d29639645f5d098c7afa036691a2
Author: Nanjesh Ramesh <[email protected]>
AuthorDate: Tue Oct 6 07:35:35 2026 -0700
NIFI-14181 Allow FlowFile attributes in QuerySalesforceObject sObject Name
and Field Names (#11756)
QuerySalesforceObject accepts an optional input FlowFile and already
evaluates
Custom WHERE Condition against it, but the sObject Name and Field Names
properties were declared with ENVIRONMENT expression language scope and
evaluated without a FlowFile. A value such as ${object} therefore resolved
to an
empty string, and the describe call to Salesforce failed with NOT_FOUND.
Declare both properties with FLOWFILE_ATTRIBUTES scope and evaluate them
against
the input FlowFile. Without an input FlowFile the behavior is unchanged.
Because the sObject can now differ between FlowFiles, store the Age Field
timestamp per sObject (last_age_filter.<sObject name, lower case>) instead
of as
one value, so incremental queries work with a dynamic sObject Name. The age
field name must exist on every queried object. State written by earlier
versions under last_age_filter is used once as the starting point for the
first
sObject queried and is then replaced by the per sObject key.
Also evaluate expression language in Age Delay, which was read without
evaluation.
Add tests for attribute based sObject Name and Field Names, for age filters
kept separate per sObject, and for the migration of the previous state.
---
.../salesforce/QuerySalesforceObject.java | 18 +--
.../salesforce/util/IncrementalContext.java | 16 ++-
.../additionalDetails.md | 7 +-
.../salesforce/QuerySalesforceObjectTest.java | 121 +++++++++++++++++++--
4 files changed, 143 insertions(+), 19 deletions(-)
diff --git
a/nifi-extension-bundles/nifi-salesforce-bundle/nifi-salesforce-processors/src/main/java/org/apache/nifi/processors/salesforce/QuerySalesforceObject.java
b/nifi-extension-bundles/nifi-salesforce-bundle/nifi-salesforce-processors/src/main/java/org/apache/nifi/processors/salesforce/QuerySalesforceObject.java
index ddc8c5d39c9..b3f451cd696 100644
---
a/nifi-extension-bundles/nifi-salesforce-bundle/nifi-salesforce-processors/src/main/java/org/apache/nifi/processors/salesforce/QuerySalesforceObject.java
+++
b/nifi-extension-bundles/nifi-salesforce-bundle/nifi-salesforce-processors/src/main/java/org/apache/nifi/processors/salesforce/QuerySalesforceObject.java
@@ -119,8 +119,9 @@ import static
org.apache.nifi.processors.salesforce.util.CommonSalesforcePropert
+ " The processor can accept an optional input FlowFile and reference
the FlowFile attributes in the query."
+ " When 'Include Deleted Records' is true, the processor will include
deleted records (soft-deletes) in the results by using the 'queryAll' API."
+ " The 'IsDeleted' field will be automatically included in the
results when querying deleted records.")
-@Stateful(scopes = Scope.CLUSTER, description = "When 'Age Field' is set,
after performing a query the time of execution is stored. Subsequent queries
will be augmented"
- + " with an additional condition so that only records that are newer
than the stored execution time (adjusted with the optional value of 'Age
Delay') will be retrieved."
+@Stateful(scopes = Scope.CLUSTER, description = "When 'Age Field' is set,
after performing a query the time of execution is stored for the queried
sObject."
+ + " Subsequent queries of the same sObject will be augmented with an
additional condition so that only records that are newer than the stored
execution time"
+ + " (adjusted with the optional value of 'Age Delay') will be
retrieved."
+ " State is stored across the cluster so that this Processor can be
run on Primary Node only and if a new Primary Node is selected,"
+ " the new node can pick up where the previous node left off, without
duplicating the data.")
@WritesAttributes({
@@ -156,7 +157,7 @@ public class QuerySalesforceObject extends
AbstractProcessor {
.name("sObject Name")
.description("The Salesforce sObject to be queried")
.required(true)
- .expressionLanguageSupported(ExpressionLanguageScope.ENVIRONMENT)
+
.expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES)
.addValidator(StandardValidators.NON_BLANK_VALIDATOR)
.dependsOn(QUERY_TYPE, PROPERTY_BASED_QUERY)
.build();
@@ -165,7 +166,7 @@ public class QuerySalesforceObject extends
AbstractProcessor {
.name("Field Names")
.description("Comma-separated list of field names requested from
the sObject to be queried. When this field is left empty, all fields are
queried.")
.required(false)
- .expressionLanguageSupported(ExpressionLanguageScope.ENVIRONMENT)
+
.expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES)
.addValidator(StandardValidators.NON_BLANK_VALIDATOR)
.dependsOn(QUERY_TYPE, PROPERTY_BASED_QUERY)
.build();
@@ -395,15 +396,15 @@ public class QuerySalesforceObject extends
AbstractProcessor {
private void processQuery(ProcessContext context, ProcessSession session,
FlowFile originalFlowFile) {
AtomicReference<String> nextRecordsUrl = new AtomicReference<>();
- String sObject =
context.getProperty(SOBJECT_NAME).evaluateAttributeExpressions().getValue();
- String fields =
context.getProperty(FIELD_NAMES).evaluateAttributeExpressions().getValue();
+ String sObject =
context.getProperty(SOBJECT_NAME).evaluateAttributeExpressions(originalFlowFile).getValue();
+ String fields =
context.getProperty(FIELD_NAMES).evaluateAttributeExpressions(originalFlowFile).getValue();
String customWhereClause =
context.getProperty(CUSTOM_WHERE_CONDITION).evaluateAttributeExpressions(originalFlowFile).getValue();
RecordSetWriterFactory writerFactory =
context.getProperty(RECORD_WRITER).asControllerService(RecordSetWriterFactory.class);
boolean createZeroRecordFlowFiles =
context.getProperty(CREATE_ZERO_RECORD_FILES).asBoolean();
boolean includeDeletedRecords =
context.getProperty(INCLUDE_DELETED_RECORDS).asBoolean();
StateMap state = getState(session);
- IncrementalContext incrementalContext = new
IncrementalContext(context, state);
+ IncrementalContext incrementalContext = new
IncrementalContext(context, state, sObject);
SalesforceSchemaHolder salesForceSchemaHolder =
getConvertedSalesforceSchema(sObject, fields, includeDeletedRecords);
if (StringUtils.isBlank(fields)) {
@@ -479,7 +480,8 @@ public class QuerySalesforceObject extends
AbstractProcessor {
if (incrementalContext.getAgeFilterUpper() != null) {
Map<String, String> newState = new
HashMap<>(state.toMap());
- newState.put(LAST_AGE_FILTER,
incrementalContext.getAgeFilterUpper());
+ newState.remove(LAST_AGE_FILTER);
+ newState.put(incrementalContext.getAgeFilterStateKey(),
incrementalContext.getAgeFilterUpper());
updateState(session, newState);
}
} catch (Exception e) {
diff --git
a/nifi-extension-bundles/nifi-salesforce-bundle/nifi-salesforce-processors/src/main/java/org/apache/nifi/processors/salesforce/util/IncrementalContext.java
b/nifi-extension-bundles/nifi-salesforce-bundle/nifi-salesforce-processors/src/main/java/org/apache/nifi/processors/salesforce/util/IncrementalContext.java
index 207cf9b713b..0c93f7bad55 100644
---
a/nifi-extension-bundles/nifi-salesforce-bundle/nifi-salesforce-processors/src/main/java/org/apache/nifi/processors/salesforce/util/IncrementalContext.java
+++
b/nifi-extension-bundles/nifi-salesforce-bundle/nifi-salesforce-processors/src/main/java/org/apache/nifi/processors/salesforce/util/IncrementalContext.java
@@ -23,6 +23,7 @@ import
org.apache.nifi.processors.salesforce.QuerySalesforceObject;
import java.time.OffsetDateTime;
import java.time.format.DateTimeFormatter;
import java.time.temporal.ChronoUnit;
+import java.util.Locale;
import java.util.Optional;
import java.util.concurrent.TimeUnit;
@@ -32,8 +33,11 @@ public class IncrementalContext {
private final String initialAgeFilter;
private final String ageFilterUpper;
private final String ageFilterLower;
+ private final String ageFilterStateKey;
+
+ public IncrementalContext(ProcessContext context, StateMap state, String
sObject) {
+ ageFilterStateKey = QuerySalesforceObject.LAST_AGE_FILTER + "." +
sObject.toLowerCase(Locale.ROOT);
- public IncrementalContext(ProcessContext context, StateMap state) {
ageField =
context.getProperty(QuerySalesforceObject.AGE_FIELD).evaluateAttributeExpressions().getValue();
if (ageField == null) {
@@ -42,8 +46,10 @@ public class IncrementalContext {
ageFilterUpper = null;
} else {
initialAgeFilter =
context.getProperty(QuerySalesforceObject.INITIAL_AGE_FILTER).evaluateAttributeExpressions().getValue();
- ageFilterLower = state.get(QuerySalesforceObject.LAST_AGE_FILTER);
- Optional<Long> ageDelayMs =
Optional.ofNullable(context.getProperty(QuerySalesforceObject.AGE_DELAY).asTimePeriod(TimeUnit.MILLISECONDS));
+ // State written before the age filter was tracked per sObject is
kept under the plain key and is picked up once
+ ageFilterLower = Optional.ofNullable(state.get(ageFilterStateKey))
+ .orElseGet(() ->
state.get(QuerySalesforceObject.LAST_AGE_FILTER));
+ Optional<Long> ageDelayMs =
Optional.ofNullable(context.getProperty(QuerySalesforceObject.AGE_DELAY).evaluateAttributeExpressions().asTimePeriod(TimeUnit.MILLISECONDS));
OffsetDateTime ageFilterUpperTime = ageDelayMs
.map(delay -> OffsetDateTime.now().minus(delay,
ChronoUnit.MILLIS))
.orElse(OffsetDateTime.now());
@@ -66,4 +72,8 @@ public class IncrementalContext {
public String getAgeFilterLower() {
return ageFilterLower;
}
+
+ public String getAgeFilterStateKey() {
+ return ageFilterStateKey;
+ }
}
diff --git
a/nifi-extension-bundles/nifi-salesforce-bundle/nifi-salesforce-processors/src/main/resources/docs/org.apache.nifi.processors.salesforce.QuerySalesforceObject/additionalDetails.md
b/nifi-extension-bundles/nifi-salesforce-bundle/nifi-salesforce-processors/src/main/resources/docs/org.apache.nifi.processors.salesforce.QuerySalesforceObject/additionalDetails.md
index 273eb4289c0..89fe88f3338 100644
---
a/nifi-extension-bundles/nifi-salesforce-bundle/nifi-salesforce-processors/src/main/resources/docs/org.apache.nifi.processors.salesforce.QuerySalesforceObject/additionalDetails.md
+++
b/nifi-extension-bundles/nifi-salesforce-bundle/nifi-salesforce-processors/src/main/resources/docs/org.apache.nifi.processors.salesforce.QuerySalesforceObject/additionalDetails.md
@@ -56,4 +56,9 @@ than the previous run time but earlier than the current run
time.
The processor uses the Initial Age Filter as a specific timestamp that sets
the beginning of the sliding window from
which processing builds the initial query. The format must adhere to the
Salesforce SOQL standards (see Salesforce
documentation). The Age Delay moves the time of the records to be processed
earlier than the current run time if
-necessary.
\ No newline at end of file
+necessary.
+
+The 'sObject Name' and 'Field Names' properties can reference attributes of an
incoming flowfile. When the Age Field
+property is set, the processor stores the end of the last time window
separately for each queried sObject, so the same
+processor can incrementally query several objects. The Age Field must be a
datetime field that exists on every object
+that is queried, as the name is shared across them.
\ No newline at end of file
diff --git
a/nifi-extension-bundles/nifi-salesforce-bundle/nifi-salesforce-processors/src/test/java/org/apache/nifi/processors/salesforce/QuerySalesforceObjectTest.java
b/nifi-extension-bundles/nifi-salesforce-bundle/nifi-salesforce-processors/src/test/java/org/apache/nifi/processors/salesforce/QuerySalesforceObjectTest.java
index e45c0817e5f..a649624023a 100644
---
a/nifi-extension-bundles/nifi-salesforce-bundle/nifi-salesforce-processors/src/test/java/org/apache/nifi/processors/salesforce/QuerySalesforceObjectTest.java
+++
b/nifi-extension-bundles/nifi-salesforce-bundle/nifi-salesforce-processors/src/test/java/org/apache/nifi/processors/salesforce/QuerySalesforceObjectTest.java
@@ -16,6 +16,7 @@
*/
package org.apache.nifi.processors.salesforce;
+import org.apache.nifi.components.state.Scope;
import org.apache.nifi.oauth2.OAuth2AccessTokenProvider;
import org.apache.nifi.processors.salesforce.rest.SalesforceConfiguration;
import org.apache.nifi.processors.salesforce.rest.SalesforceRestClient;
@@ -28,6 +29,7 @@ import org.apache.nifi.util.TestRunners;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.extension.ExtendWith;
+import org.mockito.ArgumentCaptor;
import org.mockito.Mock;
import org.mockito.junit.jupiter.MockitoExtension;
@@ -37,11 +39,18 @@ import java.time.OffsetTime;
import java.time.ZoneId;
import java.time.ZonedDateTime;
import java.time.format.DateTimeFormatter;
+import java.util.List;
+import java.util.Map;
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.assertTrue;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
@ExtendWith(MockitoExtension.class)
@@ -49,6 +58,8 @@ class QuerySalesforceObjectTest {
private static final DateTimeFormatter DATE_TIME_FORMATTER =
DateTimeFormatter.ofPattern("yyyy-MM-dd'T'HH:mm:ss.SSSX");
private static final DateTimeFormatter TIME_FORMATTER =
DateTimeFormatter.ofPattern("HH:mm:ss.SSSX");
+ private static final String AGE_FIELD = "LastModifiedDate";
+ private static final String ACCOUNT_STATE_KEY =
QuerySalesforceObject.LAST_AGE_FILTER + ".account";
private TestRunner testRunner;
@@ -86,13 +97,7 @@ class QuerySalesforceObjectTest {
String sObjectName = "TestObject";
testRunner.setProperty(QuerySalesforceObject.SOBJECT_NAME,
sObjectName);
- String recordWriterServiceId = "record_writer_service";
- RecordSetWriterFactory recordWriterService =
MockCsvRecordWriter.builder()
- .quoteValues(false)
- .build();
- testRunner.addControllerService(recordWriterServiceId,
recordWriterService);
- testRunner.enableControllerService(recordWriterService);
- testRunner.setProperty(QuerySalesforceObject.RECORD_WRITER,
recordWriterServiceId);
+ setUpRecordWriter();
when(mockSalesforceRestClient.describeSObject(sObjectName)).thenReturn(getResourceAsStream("query/describe_sobject.json"));
when(mockSalesforceRestClient.query(any())).thenReturn(getResourceAsStream("query/query_sobject.json"));
@@ -125,6 +130,108 @@ class QuerySalesforceObjectTest {
assertEquals(expectedTime, fields[3].trim());
}
+ @Test
+ void testSObjectNameEvaluatedWithFlowFileAttributes() throws Exception {
+ String sObjectName = "TestObject";
+ testRunner.setProperty(QuerySalesforceObject.SOBJECT_NAME,
"${object}");
+ setUpRecordWriter();
+
+
when(mockSalesforceRestClient.describeSObject(sObjectName)).thenReturn(getResourceAsStream("query/describe_sobject.json"));
+
when(mockSalesforceRestClient.query(any())).thenReturn(getResourceAsStream("query/query_sobject.json"));
+
+ testRunner.enqueue("", Map.of("object", sObjectName));
+ testRunner.run();
+
+ testRunner.assertTransferCount(QuerySalesforceObject.REL_SUCCESS, 1);
+ testRunner.assertTransferCount(QuerySalesforceObject.REL_ORIGINAL, 1);
+ testRunner.assertTransferCount(QuerySalesforceObject.REL_FAILURE, 0);
+ verify(mockSalesforceRestClient).describeSObject(sObjectName);
+ }
+
+ @Test
+ void testFieldNamesEvaluatedWithFlowFileAttributes() throws Exception {
+ testRunner.setProperty(QuerySalesforceObject.SOBJECT_NAME,
"TestObject");
+ testRunner.setProperty(QuerySalesforceObject.FIELD_NAMES, "${fields}");
+ setUpRecordWriter();
+
+
when(mockSalesforceRestClient.describeSObject("TestObject")).thenReturn(getResourceAsStream("query/describe_sobject.json"));
+
when(mockSalesforceRestClient.query(any())).thenReturn(getResourceAsStream("query/query_sobject.json"));
+
+ testRunner.enqueue("", Map.of("fields", "id__c,date__c"));
+ testRunner.run();
+
+ ArgumentCaptor<String> query = ArgumentCaptor.forClass(String.class);
+ verify(mockSalesforceRestClient).query(query.capture());
+ assertEquals("SELECT id__c,date__c FROM TestObject", query.getValue());
+ testRunner.assertTransferCount(QuerySalesforceObject.REL_SUCCESS, 1);
+ }
+
+ @Test
+ void testAgeFilterStoredSeparatelyForEachSObject() throws Exception {
+ testRunner.setProperty(QuerySalesforceObject.SOBJECT_NAME,
"${object}");
+ testRunner.setProperty(QuerySalesforceObject.AGE_FIELD, AGE_FIELD);
+ setUpRecordWriter();
+
+
when(mockSalesforceRestClient.describeSObject(any())).thenAnswer(invocation ->
getResourceAsStream("query/describe_sobject.json"));
+ when(mockSalesforceRestClient.query(any())).thenAnswer(invocation ->
getResourceAsStream("query/query_sobject.json"));
+
+ testRunner.enqueue("", Map.of("object", "Account"));
+ testRunner.run();
+ String accountAgeFilter =
testRunner.getStateManager().getState(Scope.CLUSTER).get(ACCOUNT_STATE_KEY);
+ assertNotNull(accountAgeFilter);
+
+ testRunner.enqueue("", Map.of("object", "Contact"));
+ testRunner.run();
+
+ testRunner.enqueue("", Map.of("object", "Account"));
+ testRunner.run();
+
+ ArgumentCaptor<String> queries = ArgumentCaptor.forClass(String.class);
+ verify(mockSalesforceRestClient, times(3)).query(queries.capture());
+ List<String> executedQueries = queries.getAllValues();
+
+ assertFalse(executedQueries.get(0).contains(AGE_FIELD + " >= "));
+ assertFalse(executedQueries.get(1).contains(AGE_FIELD + " >= "),
"Contact must not use the age filter stored for Account");
+ assertTrue(executedQueries.get(2).contains(AGE_FIELD + " >= " +
accountAgeFilter));
+
+ Map<String, String> state =
testRunner.getStateManager().getState(Scope.CLUSTER).toMap();
+ assertNotNull(state.get(ACCOUNT_STATE_KEY));
+ assertNotNull(state.get(QuerySalesforceObject.LAST_AGE_FILTER +
".contact"));
+ assertNull(state.get(QuerySalesforceObject.LAST_AGE_FILTER));
+ }
+
+ @Test
+ void testAgeFilterStoredWithoutSObjectIsMigrated() throws Exception {
+ String previousAgeFilter = "2020-01-01T00:00:00.000+0000";
+ testRunner.setProperty(QuerySalesforceObject.SOBJECT_NAME, "Account");
+ testRunner.setProperty(QuerySalesforceObject.AGE_FIELD, AGE_FIELD);
+ setUpRecordWriter();
+
testRunner.getStateManager().setState(Map.of(QuerySalesforceObject.LAST_AGE_FILTER,
previousAgeFilter), Scope.CLUSTER);
+
+
when(mockSalesforceRestClient.describeSObject("Account")).thenReturn(getResourceAsStream("query/describe_sobject.json"));
+
when(mockSalesforceRestClient.query(any())).thenReturn(getResourceAsStream("query/query_sobject.json"));
+
+ testRunner.run();
+
+ ArgumentCaptor<String> query = ArgumentCaptor.forClass(String.class);
+ verify(mockSalesforceRestClient).query(query.capture());
+ assertTrue(query.getValue().contains(AGE_FIELD + " >= " +
previousAgeFilter));
+
+ Map<String, String> state =
testRunner.getStateManager().getState(Scope.CLUSTER).toMap();
+ assertNotNull(state.get(ACCOUNT_STATE_KEY));
+ assertNull(state.get(QuerySalesforceObject.LAST_AGE_FILTER));
+ }
+
+ private void setUpRecordWriter() throws Exception {
+ String recordWriterServiceId = "record_writer_service";
+ RecordSetWriterFactory recordWriterService =
MockCsvRecordWriter.builder()
+ .quoteValues(false)
+ .build();
+ testRunner.addControllerService(recordWriterServiceId,
recordWriterService);
+ testRunner.enableControllerService(recordWriterService);
+ testRunner.setProperty(QuerySalesforceObject.RECORD_WRITER,
recordWriterServiceId);
+ }
+
private InputStream getResourceAsStream(String resourceName) {
return getClass().getClassLoader().getResourceAsStream(resourceName);
}
