This is an automated email from the ASF dual-hosted git repository.
pvillard 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 35219052e1 NIFI-13869: Enhance QuerySalesforceObject Processor to
Support Querying Deleted Records
35219052e1 is described below
commit 35219052e13c9a2a2f3a40e125832a872067d1b1
Author: Nicolae Puica <[email protected]>
AuthorDate: Tue Oct 22 14:46:47 2024 +0200
NIFI-13869: Enhance QuerySalesforceObject Processor to Support Querying
Deleted Records
- Added a new boolean property `Include Deleted Records` to allow users to
include deleted (soft-deleted) records in Salesforce queries.
- Modified the query construction to include the `IsDeleted` field and use
the `queryAll` API endpoint when `Include Deleted Records` is enabled.
- Updated `SalesforceRestClient` to support the `queryAll` API by adding a
new `queryAll` method.
- Ensured backward compatibility by defaulting `Include Deleted Records` to
`false`, maintaining existing behavior unless explicitly enabled.
- Updated processor documentation and annotations to reflect the new
functionality.
- Adjusted state management to reset state when `Include Deleted Records`
property is modified.
Signed-off-by: Pierre Villard <[email protected]>
This closes #9434.
---
.../salesforce/QuerySalesforceObject.java | 52 +++++++++++++++++-----
.../salesforce/rest/SalesforceRestClient.java | 8 ++++
2 files changed, 48 insertions(+), 12 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 9906bbee22..82e8b54b37 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
@@ -111,7 +111,9 @@ import static
org.apache.nifi.processors.salesforce.util.CommonSalesforcePropert
+ " It's also possible to define an initial cutoff value for the age,
filtering out all older records"
+ " even for the first run. In case of 'Property Based Query' this
processor should run on the Primary Node only."
+ " FlowFile attribute 'record.count' indicates how many records were
retrieved and written to the output."
- + " The processor can accept an optional input FlowFile and reference
the FlowFile attributes in the query.")
+ + " 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."
+ " 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,"
@@ -232,6 +234,16 @@ public class QuerySalesforceObject extends
AbstractProcessor {
.dependsOn(QUERY_TYPE, PROPERTY_BASED_QUERY)
.build();
+ static final PropertyDescriptor INCLUDE_DELETED_RECORDS = new
PropertyDescriptor.Builder()
+ .name("include-deleted-records")
+ .displayName("Include Deleted Records")
+ .description("If true, the processor will include deleted records
(IsDeleted = true) in the query results. When enabled, the processor will use
the 'queryAll' API.")
+ .required(true)
+ .defaultValue("false")
+ .allowableValues("true", "false")
+ .dependsOn(QUERY_TYPE, PROPERTY_BASED_QUERY)
+ .build();
+
static final Relationship REL_SUCCESS = new Relationship.Builder()
.name("success")
.description("For FlowFiles created as a result of a successful
query.")
@@ -306,6 +318,7 @@ public class QuerySalesforceObject extends
AbstractProcessor {
INITIAL_AGE_FILTER,
AGE_DELAY,
CUSTOM_WHERE_CONDITION,
+ INCLUDE_DELETED_RECORDS,
READ_TIMEOUT,
CREATE_ZERO_RECORD_FILES,
TOKEN_PROVIDER
@@ -339,7 +352,8 @@ public class QuerySalesforceObject extends
AbstractProcessor {
|| descriptor.equals(SOBJECT_NAME)
|| descriptor.equals(AGE_FIELD)
|| descriptor.equals(INITIAL_AGE_FILTER)
- || descriptor.equals(CUSTOM_WHERE_CONDITION))
+ || descriptor.equals(CUSTOM_WHERE_CONDITION)
+ || descriptor.equals(INCLUDE_DELETED_RECORDS))
) {
getLogger().debug("A property that require resetting state was
modified - {} oldValue {} newValue {}",
descriptor.getDisplayName(), oldValue, newValue);
@@ -365,10 +379,11 @@ public class QuerySalesforceObject extends
AbstractProcessor {
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);
- SalesforceSchemaHolder salesForceSchemaHolder =
getConvertedSalesforceSchema(sObject, fields);
+ SalesforceSchemaHolder salesForceSchemaHolder =
getConvertedSalesforceSchema(sObject, fields, includeDeletedRecords);
if (StringUtils.isBlank(fields)) {
fields = salesForceSchemaHolder.getSalesforceObject().getFields()
@@ -377,6 +392,14 @@ public class QuerySalesforceObject extends
AbstractProcessor {
.collect(Collectors.joining(","));
}
+ // Add IsDeleted to fields if Include Deleted Records is true
+ if (includeDeletedRecords) {
+ List<String> fieldList =
Arrays.stream(fields.split("\\s*,\\s*")).collect(Collectors.toList());
+ if (fieldList.stream().noneMatch(f ->
f.equalsIgnoreCase("IsDeleted"))) {
+ fields = fields + ", IsDeleted";
+ }
+ }
+
String querySObject = new SalesforceQueryBuilder(incrementalContext)
.buildQuery(sObject, fields, customWhereClause);
@@ -396,7 +419,7 @@ public class QuerySalesforceObject extends
AbstractProcessor {
AtomicInteger recordCountHolder = new AtomicInteger();
try {
outgoingFlowFile = session.write(outgoingFlowFile,
processRecordsCallback(session, nextRecordsUrl, writerFactory, state,
incrementalContext,
- salesForceSchemaHolder, querySObject,
originalAttributes, attributes, recordCountHolder));
+ salesForceSchemaHolder, querySObject,
originalAttributes, attributes, recordCountHolder, includeDeletedRecords));
int recordCount = recordCountHolder.get();
if (createZeroRecordFlowFiles || recordCount != 0) {
@@ -428,10 +451,10 @@ public class QuerySalesforceObject extends
AbstractProcessor {
private OutputStreamCallback processRecordsCallback(ProcessSession
session, AtomicReference<String> nextRecordsUrl, RecordSetWriterFactory
writerFactory,
StateMap state,
IncrementalContext incrementalContext, SalesforceSchemaHolder
salesForceSchemaHolder,
String querySObject,
Map<String, String> originalAttributes, Map<String, String> attributes,
- AtomicInteger
recordCountHolder) {
+ AtomicInteger
recordCountHolder, boolean includeDeletedRecords) {
return out -> {
try {
- handleRecordSet(out, nextRecordsUrl, querySObject,
writerFactory, salesForceSchemaHolder, originalAttributes, attributes,
recordCountHolder);
+ handleRecordSet(out, nextRecordsUrl, querySObject,
writerFactory, salesForceSchemaHolder, originalAttributes, attributes,
recordCountHolder, includeDeletedRecords);
if (incrementalContext.getAgeFilterUpper() != null) {
Map<String, String> newState = new
HashMap<>(state.toMap());
@@ -446,9 +469,9 @@ public class QuerySalesforceObject extends
AbstractProcessor {
private void handleRecordSet(OutputStream out, AtomicReference<String>
nextRecordsUrl, String querySObject, RecordSetWriterFactory writerFactory,
SalesforceSchemaHolder
salesForceSchemaHolder, Map<String, String> originalAttributes, Map<String,
String> attributes,
- AtomicInteger recordCountHolder) throws
Exception {
+ AtomicInteger recordCountHolder, boolean
includeDeletedRecords) throws Exception {
try (
- InputStream querySObjectResultInputStream =
getResultInputStream(nextRecordsUrl.get(), querySObject);
+ InputStream querySObjectResultInputStream =
getResultInputStream(nextRecordsUrl.get(), querySObject, includeDeletedRecords);
JsonTreeRowRecordReader jsonReader =
createJsonReader(querySObjectResultInputStream,
salesForceSchemaHolder.getRecordSchema());
RecordSetWriter writer = createRecordSetWriter(writerFactory,
originalAttributes, out, salesForceSchemaHolder.getRecordSchema())
) {
@@ -506,8 +529,9 @@ public class QuerySalesforceObject extends
AbstractProcessor {
AtomicBoolean isOriginalTransferred = new AtomicBoolean(false);
List<FlowFile> outgoingFlowFiles = new ArrayList<>();
long startNanos = System.nanoTime();
+ boolean includeDeletedRecords =
context.getProperty(INCLUDE_DELETED_RECORDS).asBoolean();
do {
- try (InputStream response =
getResultInputStream(nextRecordsUrl.get(), customQuery)) {
+ try (InputStream response =
getResultInputStream(nextRecordsUrl.get(), customQuery, includeDeletedRecords))
{
FlowFile outgoingFlowFile = createOutgoingFlowFile(session,
originalFlowFile);
outgoingFlowFiles.add(outgoingFlowFile);
outgoingFlowFile = session.write(outgoingFlowFile,
parseCustomQueryResponse(response, nextRecordsUrl, totalSize));
@@ -574,14 +598,18 @@ public class QuerySalesforceObject extends
AbstractProcessor {
.equals(value) && jsonParser.nextToken() != null;
}
- private InputStream getResultInputStream(String nextRecordsUrl, String
querySObject) {
+ private InputStream getResultInputStream(String nextRecordsUrl, String
querySObject, boolean includeDeletedRecords) {
if (nextRecordsUrl == null) {
- return salesforceRestService.query(querySObject);
+ if (includeDeletedRecords) {
+ return salesforceRestService.queryAll(querySObject);
+ } else {
+ return salesforceRestService.query(querySObject);
+ }
}
return salesforceRestService.getNextRecords(nextRecordsUrl);
}
- private SalesforceSchemaHolder getConvertedSalesforceSchema(String
sObject, String fields) {
+ private SalesforceSchemaHolder getConvertedSalesforceSchema(String
sObject, String fields, boolean includeDeletedRecords) {
try (InputStream describeSObjectResult =
salesforceRestService.describeSObject(sObject)) {
return convertSchema(describeSObjectResult, fields);
} catch (IOException e) {
diff --git
a/nifi-extension-bundles/nifi-salesforce-bundle/nifi-salesforce-processors/src/main/java/org/apache/nifi/processors/salesforce/rest/SalesforceRestClient.java
b/nifi-extension-bundles/nifi-salesforce-bundle/nifi-salesforce-processors/src/main/java/org/apache/nifi/processors/salesforce/rest/SalesforceRestClient.java
index 26cb6c420a..5fd0ff326d 100644
---
a/nifi-extension-bundles/nifi-salesforce-bundle/nifi-salesforce-processors/src/main/java/org/apache/nifi/processors/salesforce/rest/SalesforceRestClient.java
+++
b/nifi-extension-bundles/nifi-salesforce-bundle/nifi-salesforce-processors/src/main/java/org/apache/nifi/processors/salesforce/rest/SalesforceRestClient.java
@@ -57,6 +57,14 @@ public class SalesforceRestClient {
return executeRequest(request);
}
+ public InputStream queryAll(String query) {
+ HttpUrl httpUrl = HttpUrl.get(getUrl("/queryAll")).newBuilder()
+ .addQueryParameter("q", query)
+ .build();
+ Request request = buildGetRequest(httpUrl.toString());
+ return executeRequest(request);
+ }
+
public InputStream getNextRecords(String nextRecordsUrl) {
HttpUrl httpUrl = HttpUrl.get(configuration.getInstanceUrl() +
nextRecordsUrl).newBuilder().build();
Request request = buildGetRequest(httpUrl.toString());
