You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@nifi.apache.org by tp...@apache.org on 2023/03/08 13:39:45 UTC
[nifi] 01/04: NIFI-11147: Query all fields in QuerySalesforceObject
This is an automated email from the ASF dual-hosted git repository.
tpalfy pushed a commit to branch support/nifi-1.x
in repository https://gitbox.apache.org/repos/asf/nifi.git
commit e0c1f0a89b9595b7a4d40b51737e5f9da876c07f
Author: Lehel Boér <Le...@hotmail.com>
AuthorDate: Wed Feb 15 18:37:15 2023 +0100
NIFI-11147: Query all fields in QuerySalesforceObject
Fix review comments
---
.../salesforce/QuerySalesforceObject.java | 40 +++++++++++++++------
.../util/SalesforceToRecordSchemaConverter.java | 28 ++++++++-------
.../SalesforceToRecordSchemaConverterTest.java | 42 +++++++++++++++-------
.../salesforce/util/TestRecordExtender.java | 6 ++--
4 files changed, 78 insertions(+), 38 deletions(-)
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 b5125a9eb1..d831f56948 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
@@ -16,6 +16,8 @@
*/
package org.apache.nifi.processors.salesforce;
+import org.apache.camel.component.salesforce.api.dto.SObjectDescription;
+import org.apache.camel.component.salesforce.api.dto.SObjectField;
import org.apache.nifi.annotation.behavior.InputRequirement;
import org.apache.nifi.annotation.behavior.InputRequirement.Requirement;
import org.apache.nifi.annotation.behavior.PrimaryNodeOnly;
@@ -60,6 +62,7 @@ import org.apache.nifi.serialization.record.Record;
import org.apache.nifi.serialization.record.RecordField;
import org.apache.nifi.serialization.record.RecordFieldType;
import org.apache.nifi.serialization.record.RecordSchema;
+import org.apache.nifi.util.StringUtils;
import java.io.IOException;
import java.io.InputStream;
@@ -80,6 +83,7 @@ import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;
import java.util.function.BiPredicate;
+import java.util.stream.Collectors;
import static org.apache.nifi.processors.salesforce.util.CommonSalesforceProperties.API_URL;
import static org.apache.nifi.processors.salesforce.util.CommonSalesforceProperties.API_VERSION;
@@ -120,8 +124,8 @@ public class QuerySalesforceObject extends AbstractProcessor {
static final PropertyDescriptor FIELD_NAMES = new PropertyDescriptor.Builder()
.name("field-names")
.displayName("Field Names")
- .description("Comma-separated list of field names requested from the sObject to be queried")
- .required(true)
+ .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.VARIABLE_REGISTRY)
.addValidator(StandardValidators.NON_BLANK_VALIDATOR)
.build();
@@ -295,7 +299,14 @@ public class QuerySalesforceObject extends AbstractProcessor {
ageFilterUpper = ageFilterUpperTime.format(DateTimeFormatter.ISO_OFFSET_DATE_TIME);
}
- ConvertedSalesforceSchema convertedSalesforceSchema = getConvertedSalesforceSchema(sObject, fields);
+ SalesforceSchemaHolder salesForceSchemaHolder = getConvertedSalesforceSchema(sObject, fields);
+
+ if (StringUtils.isBlank(fields)) {
+ fields = salesForceSchemaHolder.getSalesforceObject().getFields()
+ .stream()
+ .map(SObjectField::getName)
+ .collect(Collectors.joining(","));
+ }
String querySObject = buildQuery(
sObject,
@@ -324,7 +335,7 @@ public class QuerySalesforceObject extends AbstractProcessor {
JsonTreeRowRecordReader jsonReader = new JsonTreeRowRecordReader(
querySObjectResultInputStream,
getLogger(),
- convertedSalesforceSchema.recordSchema,
+ salesForceSchemaHolder.recordSchema,
DATE_FORMAT,
TIME_FORMAT,
DATE_TIME_FORMAT,
@@ -338,7 +349,7 @@ public class QuerySalesforceObject extends AbstractProcessor {
getLogger(),
writerFactory.getSchema(
originalAttributes,
- convertedSalesforceSchema.recordSchema
+ salesForceSchemaHolder.recordSchema
),
out,
originalAttributes
@@ -400,7 +411,7 @@ public class QuerySalesforceObject extends AbstractProcessor {
return salesforceRestService.getNextRecords(nextRecordsUrl.get());
}
- private ConvertedSalesforceSchema getConvertedSalesforceSchema(String sObject, String fields) {
+ private SalesforceSchemaHolder getConvertedSalesforceSchema(String sObject, String fields) {
try (InputStream describeSObjectResult = salesforceRestService.describeSObject(sObject)) {
return convertSchema(describeSObjectResult, fields);
} catch (IOException e) {
@@ -416,9 +427,10 @@ public class QuerySalesforceObject extends AbstractProcessor {
}
}
- protected ConvertedSalesforceSchema convertSchema(InputStream describeSObjectResult, String fields) {
+ protected SalesforceSchemaHolder convertSchema(InputStream describeSObjectResult, String fieldsOfInterest) {
try {
- RecordSchema recordSchema = salesForceToRecordSchemaConverter.convertSchema(describeSObjectResult, fields);
+ SObjectDescription salesforceObject = salesForceToRecordSchemaConverter.getSalesforceObject(describeSObjectResult);
+ RecordSchema recordSchema = salesForceToRecordSchemaConverter.convertSchema(salesforceObject, fieldsOfInterest);
RecordSchema querySObjectResultSchema = new SimpleRecordSchema(Collections.singletonList(
new RecordField(STARTING_FIELD_NAME, RecordFieldType.ARRAY.getArrayDataType(
@@ -428,7 +440,7 @@ public class QuerySalesforceObject extends AbstractProcessor {
))
));
- return new ConvertedSalesforceSchema(querySObjectResultSchema, recordSchema);
+ return new SalesforceSchemaHolder(querySObjectResultSchema, recordSchema, salesforceObject);
} catch (IOException e) {
throw new ProcessException("SObject to Record schema conversion failed", e);
}
@@ -471,13 +483,19 @@ public class QuerySalesforceObject extends AbstractProcessor {
return queryBuilder.toString();
}
- static class ConvertedSalesforceSchema {
+ static class SalesforceSchemaHolder {
RecordSchema querySObjectResultSchema;
RecordSchema recordSchema;
+ SObjectDescription salesforceObject;
- public ConvertedSalesforceSchema(RecordSchema querySObjectResultSchema, RecordSchema recordSchema) {
+ public SalesforceSchemaHolder(RecordSchema querySObjectResultSchema, RecordSchema recordSchema, SObjectDescription salesforceObject) {
this.querySObjectResultSchema = querySObjectResultSchema;
this.recordSchema = recordSchema;
+ this.salesforceObject = salesforceObject;
+ }
+
+ public SObjectDescription getSalesforceObject() {
+ return salesforceObject;
}
}
}
diff --git a/nifi-nar-bundles/nifi-salesforce-bundle/nifi-salesforce-processors/src/main/java/org/apache/nifi/processors/salesforce/util/SalesforceToRecordSchemaConverter.java b/nifi-nar-bundles/nifi-salesforce-bundle/nifi-salesforce-processors/src/main/java/org/apache/nifi/processors/salesforce/util/SalesforceToRecordSchemaConverter.java
index f4396415ea..5bc6637161 100644
--- a/nifi-nar-bundles/nifi-salesforce-bundle/nifi-salesforce-processors/src/main/java/org/apache/nifi/processors/salesforce/util/SalesforceToRecordSchemaConverter.java
+++ b/nifi-nar-bundles/nifi-salesforce-bundle/nifi-salesforce-processors/src/main/java/org/apache/nifi/processors/salesforce/util/SalesforceToRecordSchemaConverter.java
@@ -24,6 +24,7 @@ import org.apache.nifi.serialization.SimpleRecordSchema;
import org.apache.nifi.serialization.record.RecordField;
import org.apache.nifi.serialization.record.RecordFieldType;
import org.apache.nifi.serialization.record.RecordSchema;
+import org.apache.nifi.util.StringUtils;
import java.io.IOException;
import java.io.InputStream;
@@ -33,27 +34,30 @@ import java.util.List;
import java.util.stream.Collectors;
public class SalesforceToRecordSchemaConverter {
-
+ private static final ObjectMapper OBJECT_MAPPER = JsonUtils.createObjectMapper();
private final String dateFormat;
private final String dateTimeFormat;
private final String timeFormat;
- private final ObjectMapper objectMapper;
public SalesforceToRecordSchemaConverter(String dateFormat, String dateTimeFormat, String timeFormat) {
this.dateFormat = dateFormat;
this.dateTimeFormat = dateTimeFormat;
this.timeFormat = timeFormat;
- objectMapper = JsonUtils.createObjectMapper();
}
- public RecordSchema convertSchema(final InputStream describeSOjbectResultJsonString, final String fieldNamesOfInterest) throws IOException {
+ public SObjectDescription getSalesforceObject(InputStream salesforceObjectResultJsonString) throws IOException {
+ return OBJECT_MAPPER.readValue(salesforceObjectResultJsonString, SObjectDescription.class);
+ }
- final SObjectDescription sObjectDescription = objectMapper.readValue(describeSOjbectResultJsonString, SObjectDescription.class);
- final List<String> listOfFieldNamesOfInterest = Arrays.asList(fieldNamesOfInterest.toLowerCase().split("\\s*,\\s*"));
- final List<SObjectField> fields = sObjectDescription.getFields()
- .stream()
- .filter(sObjectField -> listOfFieldNamesOfInterest.contains(sObjectField.getName().toLowerCase()))
- .collect(Collectors.toList());
+ public RecordSchema convertSchema(SObjectDescription salesforceObject, String fieldNamesOfInterest) {
+ List<SObjectField> fields = salesforceObject.getFields();
+ if (StringUtils.isNotBlank(fieldNamesOfInterest)) {
+ final List<String> listOfFieldNamesOfInterest = Arrays.asList(fieldNamesOfInterest.toLowerCase().split("\\s*,\\s*"));
+ fields = fields
+ .stream()
+ .filter(sObjectField -> listOfFieldNamesOfInterest.contains(sObjectField.getName().toLowerCase()))
+ .collect(Collectors.toList());
+ }
final List<RecordField> recordFields = new ArrayList<>();
@@ -110,8 +114,8 @@ public class SalesforceToRecordSchemaConverter {
recordFields.add(new RecordField(field.getName(), RecordFieldType.RECORD.getRecordDataType(locationSchema), field.getDefaultValue(), field.isNillable()));
break;
default:
- throw new IllegalArgumentException(String.format("Could not create determine schema for '%s'. Could not convert field '%s' of soap type '%s'.",
- sObjectDescription.getName(), field.getName(), soapType));
+ throw new IllegalArgumentException(String.format("Could not determine schema for '%s'. Could not convert field '%s' of soap type '%s'.",
+ salesforceObject.getName(), field.getName(), soapType));
}
}
diff --git a/nifi-nar-bundles/nifi-salesforce-bundle/nifi-salesforce-processors/src/test/java/org/apache/nifi/processors/salesforce/util/SalesforceToRecordSchemaConverterTest.java b/nifi-nar-bundles/nifi-salesforce-bundle/nifi-salesforce-processors/src/test/java/org/apache/nifi/processors/salesforce/util/SalesforceToRecordSchemaConverterTest.java
index c0bc3a80c1..408bd62394 100644
--- a/nifi-nar-bundles/nifi-salesforce-bundle/nifi-salesforce-processors/src/test/java/org/apache/nifi/processors/salesforce/util/SalesforceToRecordSchemaConverterTest.java
+++ b/nifi-nar-bundles/nifi-salesforce-bundle/nifi-salesforce-processors/src/test/java/org/apache/nifi/processors/salesforce/util/SalesforceToRecordSchemaConverterTest.java
@@ -17,6 +17,7 @@
package org.apache.nifi.processors.salesforce.util;
import com.fasterxml.jackson.databind.exc.MismatchedInputException;
+import org.apache.camel.component.salesforce.api.dto.SObjectDescription;
import org.apache.nifi.serialization.SimpleRecordSchema;
import org.apache.nifi.serialization.record.RecordField;
import org.apache.nifi.serialization.record.RecordFieldType;
@@ -29,7 +30,6 @@ import java.io.IOException;
import java.io.InputStream;
import java.nio.charset.StandardCharsets;
import java.util.Arrays;
-import java.util.Collections;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertThrows;
@@ -69,7 +69,8 @@ class SalesforceToRecordSchemaConverterTest {
));
try (final InputStream sfSchema = readFile(TEST_PATH + salesforceSchemaFileName)) {
- final RecordSchema actual = converter.convertSchema(sfSchema, fieldNames);
+ final SObjectDescription salesforceObject = converter.getSalesforceObject(sfSchema);
+ final RecordSchema actual = converter.convertSchema(salesforceObject, fieldNames);
assertEquals(expected, actual);
}
@@ -102,7 +103,8 @@ class SalesforceToRecordSchemaConverterTest {
));
try (final InputStream sfSchema = readFile(TEST_PATH + salesforceSchemaFileName)) {
- final RecordSchema actual = converter.convertSchema(sfSchema, fieldNames);
+ final SObjectDescription salesforceObject = converter.getSalesforceObject(sfSchema);
+ final RecordSchema actual = converter.convertSchema(salesforceObject, fieldNames);
assertEquals(expected, actual);
}
@@ -119,21 +121,36 @@ class SalesforceToRecordSchemaConverterTest {
));
try (final InputStream sfSchema = readFile(TEST_PATH + salesforceSchemaFileName)) {
- final RecordSchema actual = converter.convertSchema(sfSchema, fieldNames);
+ final SObjectDescription salesforceObject = converter.getSalesforceObject(sfSchema);
+ final RecordSchema actual = converter.convertSchema(salesforceObject, fieldNames);
assertEquals(expected, actual);
}
}
@Test
- void testSelectEmptyFields() throws IOException {
+ void testSelectAllFields() throws IOException {
final String salesforceSchemaFileName = "simple_sf_schema.json";
final String fieldNames = "";
- final RecordSchema expected = new SimpleRecordSchema(Collections.emptyList());
+ final RecordSchema expected = new SimpleRecordSchema(Arrays.asList(
+ new RecordField("ExampleInt", RecordFieldType.INT.getDataType()),
+ new RecordField("ExampleLong", RecordFieldType.LONG.getDataType()),
+ new RecordField("ExampleDouble", RecordFieldType.DOUBLE.getDataType()),
+ new RecordField("ExampleBoolean", RecordFieldType.BOOLEAN.getDataType()),
+ new RecordField("ExampleID", RecordFieldType.STRING.getDataType()),
+ new RecordField("ExampleString", RecordFieldType.STRING.getDataType()),
+ new RecordField("ExampleJson", RecordFieldType.STRING.getDataType()),
+ new RecordField("ExampleBase64Binary", RecordFieldType.STRING.getDataType()),
+ new RecordField("ExampleAnyType", RecordFieldType.STRING.getDataType()),
+ new RecordField("ExampleDate", RecordFieldType.DATE.getDataType("yyyy-mm-dd")),
+ new RecordField("ExampleDateTime", RecordFieldType.TIMESTAMP.getDataType("yyyy-mm-dd / hh:mm:ss")),
+ new RecordField("ExampleTime", RecordFieldType.TIME.getDataType("hh:mm:ss"))
+ ));
try (final InputStream sfSchema = readFile(TEST_PATH + salesforceSchemaFileName)) {
- final RecordSchema actual = converter.convertSchema(sfSchema, fieldNames);
+ final SObjectDescription salesforceObject = converter.getSalesforceObject(sfSchema);
+ final RecordSchema actual = converter.convertSchema(salesforceObject, fieldNames);
assertEquals(expected, actual);
}
@@ -142,22 +159,23 @@ class SalesforceToRecordSchemaConverterTest {
@Test
void testConvertEmptySchema() throws IOException {
try (final InputStream sfSchema = new ByteArrayInputStream("".getBytes(StandardCharsets.UTF_8))) {
- assertThrows(MismatchedInputException.class, () -> converter.convertSchema(sfSchema, "ExampleField"));
+ assertThrows(MismatchedInputException.class, () -> converter.getSalesforceObject(sfSchema));
}
}
@Test
- void testConvertNullSchema() {
+ void testConvertNullSchema() throws IOException {
final InputStream sfSchema = null;
- assertThrows(IllegalArgumentException.class, () -> converter.convertSchema(sfSchema, "ExampleField"));
+ assertThrows(IllegalArgumentException.class, () -> converter.getSalesforceObject(sfSchema));
}
@Test
void testConvertUnknownDataType() throws IOException {
try (final InputStream sfSchema = readFile(TEST_PATH + "unknown_type_sf_schema.json")) {
final String fieldNames = "FieldWithUnknownType";
- final IllegalArgumentException exception = assertThrows(IllegalArgumentException.class, () -> converter.convertSchema(sfSchema, fieldNames));
- final String errorMessage = "Could not create determine schema for 'SObjectWithUnknownFieldType'. Could not convert field 'FieldWithUnknownType' of soap type 'xsd:unknown'.";
+ final SObjectDescription salesforceObject = converter.getSalesforceObject(sfSchema);
+ final IllegalArgumentException exception = assertThrows(IllegalArgumentException.class, () -> converter.convertSchema(salesforceObject, fieldNames));
+ final String errorMessage = "Could not determine schema for 'SObjectWithUnknownFieldType'. Could not convert field 'FieldWithUnknownType' of soap type 'xsd:unknown'.";
assertEquals(errorMessage, exception.getMessage());
}
}
diff --git a/nifi-nar-bundles/nifi-salesforce-bundle/nifi-salesforce-processors/src/test/java/org/apache/nifi/processors/salesforce/util/TestRecordExtender.java b/nifi-nar-bundles/nifi-salesforce-bundle/nifi-salesforce-processors/src/test/java/org/apache/nifi/processors/salesforce/util/TestRecordExtender.java
index a55f10b8c1..bd48c5b5b6 100644
--- a/nifi-nar-bundles/nifi-salesforce-bundle/nifi-salesforce-processors/src/test/java/org/apache/nifi/processors/salesforce/util/TestRecordExtender.java
+++ b/nifi-nar-bundles/nifi-salesforce-bundle/nifi-salesforce-processors/src/test/java/org/apache/nifi/processors/salesforce/util/TestRecordExtender.java
@@ -94,15 +94,15 @@ class TestRecordExtender {
int referenceId = 0;
String objectType = "Account";
- MapRecord testRecord = new MapRecord(ORIGINAL_SCHEMA, new HashMap<String, Object>() {{
+ MapRecord testRecord = new MapRecord(ORIGINAL_SCHEMA, new HashMap<>() {{
put("testRecordField1", "testRecordValue1");
put("testRecordField2", "testRecordValue2");
}});
- MapRecord expectedRecord = new MapRecord(EXPECTED_EXTENDED_SCHEMA, new HashMap<String, Object>() {{
+ MapRecord expectedRecord = new MapRecord(EXPECTED_EXTENDED_SCHEMA, new HashMap<>() {{
put("attributes",
- new MapRecord(ATTRIBUTES_RECORD_SCHEMA, new HashMap<String, Object>() {{
+ new MapRecord(ATTRIBUTES_RECORD_SCHEMA, new HashMap<>() {{
put("type", objectType);
put("referenceId", referenceId);
}})