This is an automated email from the ASF dual-hosted git repository.
exceptionfactory pushed a commit to branch support/nifi-1.x
in repository https://gitbox.apache.org/repos/asf/nifi.git
The following commit(s) were added to refs/heads/support/nifi-1.x by this push:
new 49596efa09 NIFI-11945 Add Cache Entry Identifier property to
DeduplicateRecord
49596efa09 is described below
commit 49596efa094b5a4d6133e4af6555b23b007c106a
Author: p-kimberley <[email protected]>
AuthorDate: Sat Aug 12 12:38:12 2023 +1000
NIFI-11945 Add Cache Entry Identifier property to DeduplicateRecord
This closes #7603
Signed-off-by: David Handermann <[email protected]>
(cherry picked from commit a9ac8fb7073ef89e1f1105775de2fdb56b8da0b3)
---
.../processors/standard/DeduplicateRecord.java | 175 +++++++++------------
.../nifi/processors/standard/MockCacheService.java | 30 +++-
.../processors/standard/TestDeduplicateRecord.java | 147 +++++++++++++++--
.../nifi/processors/standard/TestListSFTP.java | 4 +-
4 files changed, 236 insertions(+), 120 deletions(-)
diff --git
a/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/DeduplicateRecord.java
b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/DeduplicateRecord.java
index b055a75273..0c1ea0e93e 100644
---
a/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/DeduplicateRecord.java
+++
b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/DeduplicateRecord.java
@@ -67,7 +67,6 @@ import org.apache.nifi.serialization.record.Record;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
-import java.io.Serializable;
import java.nio.charset.Charset;
import java.nio.charset.StandardCharsets;
import java.security.MessageDigest;
@@ -81,7 +80,6 @@ import java.util.Map;
import java.util.Set;
import java.util.stream.Collectors;
-import static java.util.stream.Collectors.toList;
import static org.apache.commons.codec.binary.StringUtils.getBytesUtf8;
@EventDriven
@@ -96,14 +94,15 @@ import static
org.apache.commons.codec.binary.StringUtils.getBytesUtf8;
)
@Tags({"text", "record", "update", "change", "replace", "modify", "distinct",
"unique",
"filter", "hash", "dupe", "duplicate", "dedupe"})
-@CapabilityDescription("This processor attempts to deduplicate a record set in
memory using either a hashset or a bloom filter. " +
- "It operates on a per-file basis rather than across an entire data set
that spans multiple files.")
-@WritesAttribute(attribute = "record.count", description = "The number of
records processed.")
+@CapabilityDescription("This processor de-duplicates individual records within
a record set. " +
+ "It can operate on a per-file basis using an in-memory hashset or
bloom filter. " +
+ "When configured with a distributed map cache, it de-duplicates
records across multiple files.")
+@WritesAttribute(attribute = DeduplicateRecord.RECORD_COUNT_ATTRIBUTE,
description = "Number of records written to the destination FlowFile.")
@DynamicProperty(
- name = "RecordPath",
- value = "An expression language statement used to determine how the
RecordPath is resolved. " +
- "The following variables are availible: ${field.name},
${field.value}, ${field.type}",
- description = "The name of each user-defined property must be a valid
RecordPath.")
+ name = "Name of the property.",
+ value = "A valid RecordPath to the record field to be included in the
cache key used for deduplication.",
+ description = "A record's cache key is generated by combining the name
of each dynamic property with its evaluated record value " +
+ "(as specified by the corresponding RecordPath).")
@SeeAlso(classNames = {
"org.apache.nifi.distributed.cache.client.DistributedMapCacheClientService",
"org.apache.nifi.distributed.cache.server.map.DistributedMapCacheServer",
@@ -111,10 +110,8 @@ import static
org.apache.commons.codec.binary.StringUtils.getBytesUtf8;
})
public class DeduplicateRecord extends AbstractProcessor {
public static final char JOIN_CHAR = '~';
-
- private static final String FIELD_NAME = "field.name";
- private static final String FIELD_VALUE = "field.value";
- private static final String FIELD_TYPE = "field.type";
+ public static final String RECORD_COUNT_ATTRIBUTE = "record.count";
+ public static final String RECORD_HASH_VALUE_ATTRIBUTE =
"record.hash.value";
private volatile RecordPathCache recordPathCache;
private volatile List<PropertyDescriptor> dynamicProperties;
@@ -122,17 +119,18 @@ public class DeduplicateRecord extends AbstractProcessor {
// VALUES
static final AllowableValue NONE_ALGORITHM_VALUE = new
AllowableValue("none", "None",
- "Do not use a hashing algorithm. The value of resolved RecordPaths
will be combined with tildes (~) to form the unique record key. " +
+ "Do not use a hashing algorithm. The value of resolved RecordPaths
will be combined with a delimiter " +
+ "(" + DeduplicateRecord.JOIN_CHAR + ") to form the unique
cache key. " +
"This may use significantly more storage depending on the
size and shape or your data.");
static final AllowableValue SHA256_ALGORITHM_VALUE = new
AllowableValue(MessageDigestAlgorithms.SHA_256, "SHA-256",
- "The SHA-256 cryptographic hash algorithm.");
+ "SHA-256 cryptographic hashing algorithm.");
static final AllowableValue SHA512_ALGORITHM_VALUE = new
AllowableValue(MessageDigestAlgorithms.SHA_512, "SHA-512",
- "The SHA-512 cryptographic hash algorithm.");
+ "SHA-512 cryptographic hashing algorithm.");
static final AllowableValue HASH_SET_VALUE = new
AllowableValue("hash-set", "HashSet",
"Exactly matches records seen before with 100% accuracy at the
expense of more storage usage. " +
"Stores the filter data in a single cache entry in the
distributed cache, and is loaded entirely into memory during duplicate
detection. " +
- "This filter is preferred for small to medium data sets
and offers high performance loaded into memory when this processor is
running.");
+ "This filter is preferred for small to medium data sets
and offers high performance, being loaded into memory when this processor is
running.");
static final AllowableValue BLOOM_FILTER_VALUE = new
AllowableValue("bloom-filter", "BloomFilter",
"Space-efficient data structure ideal for large data sets using
probability to determine if a record was seen previously. " +
"False positive matches are possible, but false negatives
are not – in other words, a query returns either \"possibly in the set\" or
\"definitely not in the set\". " +
@@ -162,9 +160,9 @@ public class DeduplicateRecord extends AbstractProcessor {
static final PropertyDescriptor DEDUPLICATION_STRATEGY = new
PropertyDescriptor.Builder()
.name("deduplication-strategy")
.displayName("Deduplication Strategy")
- .description("The strategy to use for detecting and isolating
duplicate records. The option for doing it " +
- "across a single data file will operate in memory, whereas
the one for going across the enter repository " +
- "will require a distributed map cache.")
+ .description("The strategy to use for detecting and routing
duplicate records. The option for detecting " +
+ "duplicates across a single FlowFile operates in-memory,
whereas detection spanning multiple FlowFiles " +
+ "utilises a distributed map cache.")
.allowableValues(OPTION_SINGLE_FILE, OPTION_MULTIPLE_FILES)
.defaultValue(OPTION_SINGLE_FILE.getValue())
.required(true)
@@ -173,10 +171,8 @@ public class DeduplicateRecord extends AbstractProcessor {
static final PropertyDescriptor DISTRIBUTED_MAP_CACHE = new
PropertyDescriptor.Builder()
.name("distributed-map-cache")
.displayName("Distributed Map Cache client")
- .description("This configuration is required when the
deduplication strategy is set to 'multiple files.' The map " +
- "cache will be used to check a data source such as HBase
or Redis for entries indicating that a record has " +
- "been processed before. This option requires a downstream
process that uses PutDistributedMapCache to write " +
- "an entry to the cache data source once the record has
been processed to indicate that it has been handled before.")
+ .description("This property is required when the deduplication
strategy is set to 'multiple files.' The map " +
+ "cache will for each record, atomically check whether the
cache key exists and if not, set it.")
.identifiesControllerService(DistributedMapCacheClient.class)
.required(false)
.addValidator(Validator.VALID)
@@ -186,20 +182,32 @@ public class DeduplicateRecord extends AbstractProcessor {
static final PropertyDescriptor CACHE_IDENTIFIER = new
PropertyDescriptor.Builder()
.name("cache-identifier")
.displayName("Cache Identifier")
- .description("This option defines a record path operation to use
for defining the cache identifier. It can be used " +
- "in addition to the hash settings. This field will have
the expression language attribute \"record.hash.value\" " +
- "available to it to use with it to generate the record
path operation.")
+ .description("An optional expression language field that overrides
the record's computed cache key. " +
+ "This field has an additional attribute available: ${" +
RECORD_HASH_VALUE_ATTRIBUTE + "}, " +
+ "which contains the cache key derived from dynamic
properties (if set) or record fields.")
.expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES)
.required(false)
.addValidator(Validator.VALID)
.dependsOn(DEDUPLICATION_STRATEGY, OPTION_MULTIPLE_FILES)
.build();
+ static final PropertyDescriptor PUT_CACHE_IDENTIFIER = new
PropertyDescriptor.Builder()
+ .name("put-cache-identifier")
+ .displayName("Cache the Entry Identifier")
+ .description("For each record, check whether the cache identifier
exists in the distributed map cache. " +
+ "If it doesn't exist and this property is true, put the
identifier to the cache.")
+ .required(true)
+ .allowableValues("true", "false")
+ .dependsOn(DISTRIBUTED_MAP_CACHE)
+ .defaultValue("false")
+ .build();
+
static final PropertyDescriptor INCLUDE_ZERO_RECORD_FLOWFILES = new
PropertyDescriptor.Builder()
.name("include-zero-record-flowfiles")
.displayName("Include Zero Record FlowFiles")
- .description("When converting an incoming FlowFile, if the
conversion results in no data, "
- + "this property specifies whether or not a FlowFile will
be sent to the corresponding relationship")
+ .description("If a FlowFile sent to either the duplicate or
non-duplicate relationships contains no records, " +
+ "a value of `false` in this property causes the FlowFile
to be dropped. Otherwise, the empty FlowFile " +
+ "is emitted.")
.expressionLanguageSupported(ExpressionLanguageScope.NONE)
.allowableValues("true", "false")
.defaultValue("true")
@@ -209,7 +217,7 @@ public class DeduplicateRecord extends AbstractProcessor {
static final PropertyDescriptor RECORD_HASHING_ALGORITHM = new
PropertyDescriptor.Builder()
.name("record-hashing-algorithm")
.displayName("Record Hashing Algorithm")
- .description("The algorithm used to hash the combined set of
resolved RecordPath values for cache storage.")
+ .description("The algorithm used to hash the cache key.")
.allowableValues(
NONE_ALGORITHM_VALUE,
SHA256_ALGORITHM_VALUE,
@@ -265,17 +273,17 @@ public class DeduplicateRecord extends AbstractProcessor {
static final Relationship REL_DUPLICATE = new Relationship.Builder()
.name("duplicate")
- .description("Records detected as duplicates in the FlowFile
content will be routed to this relationship")
+ .description("Records detected as duplicates are routed to this
relationship.")
.build();
static final Relationship REL_NON_DUPLICATE = new Relationship.Builder()
.name("non-duplicate")
- .description("If the record was not found in the cache, it will be
routed to this relationship")
+ .description("Records not found in the cache are routed to this
relationship.")
.build();
static final Relationship REL_ORIGINAL = new Relationship.Builder()
.name("original")
- .description("The original input FlowFile is sent to this
relationship unless there is a fatal error in the processing.")
+ .description("The original input FlowFile is sent to this
relationship unless a fatal error occurs.")
.build();
static final Relationship REL_FAILURE = new Relationship.Builder()
@@ -293,6 +301,7 @@ public class DeduplicateRecord extends AbstractProcessor {
descriptors.add(DEDUPLICATION_STRATEGY);
descriptors.add(DISTRIBUTED_MAP_CACHE);
descriptors.add(CACHE_IDENTIFIER);
+ descriptors.add(PUT_CACHE_IDENTIFIER);
descriptors.add(RECORD_READER);
descriptors.add(RECORD_WRITER);
descriptors.add(INCLUDE_ZERO_RECORD_FLOWFILES);
@@ -324,11 +333,11 @@ public class DeduplicateRecord extends AbstractProcessor {
protected PropertyDescriptor getSupportedDynamicPropertyDescriptor(final
String propertyDescriptorName) {
return new PropertyDescriptor.Builder()
.name(propertyDescriptorName)
- .description("Specifies a value to use from the record that
matches the RecordPath: '" +
- propertyDescriptorName + "' which is used together
with other specified " +
- "record path values to determine the uniqueness of a
record. " +
- "Expression Language may reference variables
'field.name', 'field.type', and 'field.value' " +
- "to access information about the field and the value
of the field being evaluated.")
+ .description("The property's value is a RecordPath, which is
evaluated when testing whether a record is a duplicate or not. " +
+ "Multiple dynamic properties are supported. The key
used to de-duplicate a record is determined by concatenating " +
+ "the following for each dynamic property: property
name, a fixed delimiter and the evaluated RecordPath for the record. " +
+ "If a hashing algorithm is configured, the key is
hashed prior to being used in the state cache. "
+ )
.required(false)
.dynamic(true)
.addValidator(new RecordPathValidator())
@@ -338,7 +347,6 @@ public class DeduplicateRecord extends AbstractProcessor {
@Override
protected Collection<ValidationResult> customValidate(final
ValidationContext context) {
- RecordPathValidator recordPathValidator = new RecordPathValidator();
List<ValidationResult> validationResults = new ArrayList<>();
boolean useSingleFile =
context.getProperty(DEDUPLICATION_STRATEGY).getValue().equals(OPTION_SINGLE_FILE.getValue());
@@ -397,7 +405,7 @@ public class DeduplicateRecord extends AbstractProcessor {
private FilterWrapper getFilter(ProcessContext context) {
if (useInMemoryStrategy) {
boolean useHashSet = context.getProperty(FILTER_TYPE).getValue()
- .equals(context.getProperty(HASH_SET_VALUE.getValue()));
+
.equals(context.getProperty(HASH_SET_VALUE.getValue()).getValue());
final int filterCapacity =
context.getProperty(FILTER_CAPACITY_HINT).asInteger();
return useHashSet
? new HashSetFilterWrapper(new HashSet<>(filterCapacity))
@@ -407,7 +415,7 @@ public class DeduplicateRecord extends AbstractProcessor {
context.getProperty(BLOOM_FILTER_FPP).asDouble()
));
} else {
- return new DistributedMapCacheClientWrapper(mapCacheClient);
+ return new DistributedMapCacheClientWrapper(mapCacheClient,
context.getProperty(PUT_CACHE_IDENTIFIER).asBoolean());
}
}
@@ -447,7 +455,7 @@ public class DeduplicateRecord extends AbstractProcessor {
final MessageDigest messageDigest =
recordHashingAlgorithm.equals(NONE_ALGORITHM_VALUE.getValue())
? null
: DigestUtils.getDigest(recordHashingAlgorithm);
- final Boolean matchWholeRecord =
context.getProperties().keySet().stream().noneMatch(p -> p.isDynamic());
+ final boolean matchWholeRecord =
context.getProperties().keySet().stream().noneMatch(PropertyDescriptor::isDynamic);
nonDuplicatesWriter.beginRecordSet();
duplicatesWriter.beginRecordSet();
@@ -459,26 +467,21 @@ public class DeduplicateRecord extends AbstractProcessor {
if (matchWholeRecord) {
recordValue =
Joiner.on(JOIN_CHAR).join(record.getValues());
} else {
- recordValue = executeDynamicRecordPaths(context, record,
flowFile);
+ recordValue = evaluateKeyFromDynamicProperties(context,
record, flowFile);
}
- String recordHash = messageDigest != null
- ?
Hex.encodeHexString(messageDigest.digest(getBytesUtf8(recordValue)))
- : recordValue;
- messageDigest.reset();
+ // Hash the record value if a hashing algorithm is specified
+ String recordHash = recordValue;
+ if (messageDigest != null) {
+ recordHash =
Hex.encodeHexString(messageDigest.digest(getBytesUtf8(recordValue)));
+ messageDigest.reset();
+ }
+ // If a cache entry identifier is specified, apply any
expressions to determine the final cache value
if (!useInMemoryStrategy &&
context.getProperty(CACHE_IDENTIFIER).isSet()) {
Map<String, String> additional = new HashMap<>();
- additional.put("record.hash.value", recordHash);
- String rawPath =
context.getProperty(CACHE_IDENTIFIER).evaluateAttributeExpressions(flowFile,
additional).getValue();
- RecordPath compiled = recordPathCache.getCompiled(rawPath);
- RecordPathResult result = compiled.evaluate(record);
- FieldValue fieldValue =
result.getSelectedFields().findFirst().get();
- if (fieldValue.getValue() == null) {
- throw new ProcessException(String.format("The path
\"%s\" failed to create an ID value at record index %d", rawPath, index));
- }
-
- recordHash = fieldValue.getValue().toString();
+ additional.put(RECORD_HASH_VALUE_ATTRIBUTE, recordHash);
+ recordHash =
context.getProperty(CACHE_IDENTIFIER).evaluateAttributeExpressions(flowFile,
additional).getValue();
}
if (filter.contains(recordHash)) {
@@ -532,9 +535,8 @@ public class DeduplicateRecord extends AbstractProcessor {
if (!includeZeroRecordFlowFiles && writeResult.getRecordCount() == 0) {
session.remove(outputFlowFile);
} else {
- Map<String, String> attributes = new HashMap<>();
- attributes.putAll(writeResult.getAttributes());
- attributes.put("record.count",
String.valueOf(writeResult.getRecordCount()));
+ Map<String, String> attributes = new
HashMap<>(writeResult.getAttributes());
+ attributes.put(RECORD_COUNT_ATTRIBUTE,
String.valueOf(writeResult.getRecordCount()));
attributes.put(CoreAttributes.MIME_TYPE.key(), mimeType);
outputFlowFile = session.putAllAttributes(outputFlowFile,
attributes);
if (getLogger().isDebugEnabled()) {
@@ -546,7 +548,7 @@ public class DeduplicateRecord extends AbstractProcessor {
}
}
- private String executeDynamicRecordPaths(ProcessContext context, Record
record, FlowFile flowFile) {
+ private String evaluateKeyFromDynamicProperties(ProcessContext context,
Record record, FlowFile flowFile) {
final List<String> fieldValues = new ArrayList<>();
for (final PropertyDescriptor propertyDescriptor : dynamicProperties) {
final String value =
context.getProperty(propertyDescriptor).evaluateAttributeExpressions(flowFile).getValue();
@@ -554,11 +556,14 @@ public class DeduplicateRecord extends AbstractProcessor {
final RecordPathResult result = recordPath.evaluate(record);
final List<FieldValue> selectedFields =
result.getSelectedFields().collect(Collectors.toList());
+ // Add the name of the dynamic property
fieldValues.add(propertyDescriptor.getName());
+ // Add any non-null record field values
fieldValues.addAll(selectedFields.stream()
+ .filter(f -> f.getValue() != null)
.map(f -> f.getValue().toString())
- .collect(toList())
+ .collect(Collectors.toList())
);
}
@@ -566,14 +571,6 @@ public class DeduplicateRecord extends AbstractProcessor {
}
private abstract static class FilterWrapper {
- public static FilterWrapper create(Object filter) {
- if (filter instanceof HashSet) {
- return new HashSetFilterWrapper((HashSet<String>) filter);
- } else {
- return new BloomFilterWrapper((BloomFilter<String>) filter);
- }
- }
-
public abstract boolean contains(String value);
public abstract void put(String value);
@@ -618,16 +615,23 @@ public class DeduplicateRecord extends AbstractProcessor {
}
private static class DistributedMapCacheClientWrapper extends
FilterWrapper {
- private DistributedMapCacheClient client;
+ private final DistributedMapCacheClient client;
+ private final boolean putCacheIdentifier;
- public DistributedMapCacheClientWrapper(DistributedMapCacheClient
client) {
+ public DistributedMapCacheClientWrapper(final
DistributedMapCacheClient client,
+ final boolean
putCacheIdentifier) {
this.client = client;
+ this.putCacheIdentifier = putCacheIdentifier;
}
@Override
public boolean contains(String value) {
try {
- return client.containsKey(value, STRING_SERIALIZER);
+ if (putCacheIdentifier) {
+ return !client.putIfAbsent(value, "", STRING_SERIALIZER,
STRING_SERIALIZER);
+ } else {
+ return client.containsKey(value, STRING_SERIALIZER);
+ }
} catch (IOException e) {
throw new ProcessException("Distributed Map lookup failed", e);
}
@@ -635,32 +639,9 @@ public class DeduplicateRecord extends AbstractProcessor {
@Override
public void put(String value) {
- /*
- * This needs to be a noop because this process will be used
upstream of the systems that would write the records
- * that power the map cache.
- */
+ // Do nothing as the key is queried and put to the cache
atomically in the `contains` method.
}
}
private static final Serializer<String> STRING_SERIALIZER = (value,
output) -> output.write(value.getBytes(StandardCharsets.UTF_8));
- private static final Serializer<Boolean> BOOLEAN_SERIALIZER = (value,
output) -> output.write((byte) (value ? 1 : 0));
-
- private static class CacheValue implements Serializable {
-
- private final Serializable filter;
- private final long entryTimeMS;
-
- public CacheValue(Serializable filter, long entryTimeMS) {
- this.filter = filter;
- this.entryTimeMS = entryTimeMS;
- }
-
- public Serializable getFilter() {
- return filter;
- }
-
- public long getEntryTimeMS() {
- return entryTimeMS;
- }
- }
}
diff --git
a/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/java/org/apache/nifi/processors/standard/MockCacheService.java
b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/java/org/apache/nifi/processors/standard/MockCacheService.java
index c2f231efe5..f6428b4ffa 100644
---
a/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/java/org/apache/nifi/processors/standard/MockCacheService.java
+++
b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/java/org/apache/nifi/processors/standard/MockCacheService.java
@@ -25,21 +25,34 @@ import java.io.IOException;
import java.util.HashMap;
import java.util.Map;
-final class MockCacheService<K, V> extends AbstractControllerService
implements DistributedMapCacheClient {
- private Map storage;
+final class MockCacheService extends AbstractControllerService implements
DistributedMapCacheClient {
+ private Map<Object, Object> storage;
public MockCacheService() {
storage = new HashMap<>();
}
+ /**
+ * Puts the value to the cache with the specified key, if it doesn't
already exist
+ * @param key the key for into the map
+ * @param value the value to add to the map if and only if the key is
absent
+ * @param keySerializer key serializer
+ * @param valueSerializer value serializer
+ * @return true if the value was added to the cache, false if it already
exists
+ */
@Override
public <K, V> boolean putIfAbsent(K key, V value, Serializer<K>
keySerializer, Serializer<V> valueSerializer) throws IOException {
- return false;
+ return storage.putIfAbsent(key, value) == null;
}
@Override
public <K, V> V getAndPutIfAbsent(K key, V value, Serializer<K>
keySerializer, Serializer<V> valueSerializer, Deserializer<V>
valueDeserializer) throws IOException {
- return null;
+ if (storage.containsKey(key)) {
+ return (V) storage.get(key);
+ } else {
+ storage.put(key, value);
+ return null;
+ }
}
@Override
@@ -54,7 +67,7 @@ final class MockCacheService<K, V> extends
AbstractControllerService implements
@Override
public <K, V> V get(K key, Serializer<K> keySerializer, Deserializer<V>
valueDeserializer) throws IOException {
- return null;
+ return (V) storage.get(key);
}
@Override
@@ -64,7 +77,12 @@ final class MockCacheService<K, V> extends
AbstractControllerService implements
@Override
public <K> boolean remove(K key, Serializer<K> serializer) throws
IOException {
- return false;
+ if (storage.containsKey(key)) {
+ storage.remove(key);
+ return true;
+ } else {
+ return false;
+ }
}
@Override
diff --git
a/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/java/org/apache/nifi/processors/standard/TestDeduplicateRecord.java
b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/java/org/apache/nifi/processors/standard/TestDeduplicateRecord.java
index 9f90c3186c..1ccc82bf87 100644
---
a/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/java/org/apache/nifi/processors/standard/TestDeduplicateRecord.java
+++
b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/java/org/apache/nifi/processors/standard/TestDeduplicateRecord.java
@@ -193,22 +193,138 @@ public class TestDeduplicateRecord {
runner.assertNotValid();
}
- public static final String FIRST_KEY =
DigestUtils.sha256Hex(String.join(String.valueOf(DeduplicateRecord.JOIN_CHAR),
Arrays.asList(
+ public static final String FIRST_KEY =
String.join(String.valueOf(DeduplicateRecord.JOIN_CHAR), Arrays.asList(
"John", "Q", "Smith"
- )));
- public static final String SECOND_KEY =
DigestUtils.sha256Hex(String.join(String.valueOf(DeduplicateRecord.JOIN_CHAR),
Arrays.asList(
+ ));
+ public static final String SECOND_KEY =
String.join(String.valueOf(DeduplicateRecord.JOIN_CHAR), Arrays.asList(
"Jack", "Z", "Brown"
- )));
+ ));
+ public static final String FIRST_KEY_HASHED =
DigestUtils.sha256Hex(FIRST_KEY);
+ public static final String SECOND_KEY_HASHED =
DigestUtils.sha256Hex(SECOND_KEY);
@Test
- public void testDeduplicateWithDMC() throws Exception {
- DistributedMapCacheClient dmc = new MockCacheService<>();
+ public void testDeduplicateWithDMCNoPutIdentifier() throws Exception {
+ DistributedMapCacheClient dmc = new MockCacheService();
runner.addControllerService("dmc", dmc);
runner.setProperty(DeduplicateRecord.DISTRIBUTED_MAP_CACHE, "dmc");
runner.setProperty(DeduplicateRecord.DEDUPLICATION_STRATEGY,
DeduplicateRecord.OPTION_MULTIPLE_FILES.getValue());
runner.enableControllerService(dmc);
runner.assertValid();
+ dmc.put(FIRST_KEY_HASHED, true, null, null);
+
+ reader.addRecord("John", "Q", "Smith");
+ reader.addRecord("Jack", "Z", "Brown");
+ reader.addRecord("Jack", "Z", "Brown");
+ reader.addRecord("Jane", "X", "Doe");
+
+ runner.enqueue("");
+ runner.run();
+
+ doCountTests(0, 1, 1, 1, 3, 1);
+ }
+
+ @Test
+ public void testDeduplicateWithDMCPutIdentifier() throws Exception {
+ DistributedMapCacheClient dmc = new MockCacheService();
+ runner.addControllerService("dmc", dmc);
+ runner.setProperty(DeduplicateRecord.DISTRIBUTED_MAP_CACHE, "dmc");
+ runner.setProperty(DeduplicateRecord.PUT_CACHE_IDENTIFIER, "true");
+ runner.setProperty(DeduplicateRecord.DEDUPLICATION_STRATEGY,
DeduplicateRecord.OPTION_MULTIPLE_FILES.getValue());
+ runner.enableControllerService(dmc);
+ runner.assertValid();
+
+ dmc.put(FIRST_KEY_HASHED, true, null, null);
+
+ reader.addRecord("John", "Q", "Smith");
+ reader.addRecord("Jack", "Z", "Brown");
+ reader.addRecord("Jack", "Z", "Brown");
+ reader.addRecord("Jane", "X", "Doe");
+
+ runner.enqueue("");
+ runner.run();
+
+ doCountTests(0, 1, 1, 1, 2, 2);
+ }
+
+ @Test
+ public void testDeduplicateWithDMCAndOneRecordPath() throws Exception {
+ DistributedMapCacheClient dmc = new MockCacheService();
+ runner.addControllerService("dmc", dmc);
+ runner.setProperty(DeduplicateRecord.DISTRIBUTED_MAP_CACHE, "dmc");
+ runner.setProperty(DeduplicateRecord.PUT_CACHE_IDENTIFIER, "true");
+ runner.setProperty(DeduplicateRecord.DEDUPLICATION_STRATEGY,
DeduplicateRecord.OPTION_MULTIPLE_FILES.getValue());
+ runner.setProperty("first_name", "/firstName");
+ runner.enableControllerService(dmc);
+ runner.assertValid();
+
+ reader.addRecord("John", "Q", "Smith");
+ reader.addRecord("Jack", "Z", "Brown");
+ reader.addRecord("Jack", "Z", "Brown");
+ reader.addRecord("Jane", "X", "Doe");
+
+ runner.enqueue("");
+ runner.run();
+
+ doCountTests(0, 1, 1, 1, 3, 1);
+ }
+
+ @Test
+ public void testDeduplicateWithDMCAndMultipleRecordPaths() throws
Exception {
+ DistributedMapCacheClient dmc = new MockCacheService();
+ runner.addControllerService("dmc", dmc);
+ runner.setProperty(DeduplicateRecord.DISTRIBUTED_MAP_CACHE, "dmc");
+ runner.setProperty(DeduplicateRecord.PUT_CACHE_IDENTIFIER, "true");
+ runner.setProperty(DeduplicateRecord.DEDUPLICATION_STRATEGY,
DeduplicateRecord.OPTION_MULTIPLE_FILES.getValue());
+ runner.setProperty("first_name", "/firstName");
+ runner.setProperty("middle_name", "/middleName");
+ runner.setProperty("invalid_property", "/missingProperty");
+ runner.enableControllerService(dmc);
+ runner.assertValid();
+
+ reader.addRecord("John", "Q", "Smith");
+ reader.addRecord("Jack", "Z", "Brown");
+ reader.addRecord("Jack", "Z", "Brown");
+ reader.addRecord("Jane", "X", "Doe");
+
+ runner.enqueue("");
+ runner.run();
+
+ doCountTests(0, 1, 1, 1, 3, 1);
+ }
+
+ @Test
+ public void testDeduplicateWithDMCAndNullField() throws Exception {
+ DistributedMapCacheClient dmc = new MockCacheService();
+ runner.addControllerService("dmc", dmc);
+ runner.setProperty(DeduplicateRecord.DISTRIBUTED_MAP_CACHE, "dmc");
+ runner.setProperty(DeduplicateRecord.PUT_CACHE_IDENTIFIER, "true");
+ runner.setProperty(DeduplicateRecord.DEDUPLICATION_STRATEGY,
DeduplicateRecord.OPTION_MULTIPLE_FILES.getValue());
+ runner.setProperty("middle_name", "/middleName");
+ runner.setProperty("last_name", "/lastName");
+ runner.enableControllerService(dmc);
+ runner.assertValid();
+
+ reader.addRecord("Jack", "Z", null);
+ reader.addRecord("Jack", "Z", "Brown");
+ reader.addRecord("Jack", "R", "Brown");
+
+ runner.enqueue("");
+ runner.run();
+
+ doCountTests(0, 1, 1, 1, 3, 0);
+ }
+
+ @Test
+ public void testDeduplicateNoHashing() throws Exception {
+ DistributedMapCacheClient dmc = new MockCacheService();
+ runner.addControllerService("dmc", dmc);
+ runner.setProperty(DeduplicateRecord.DISTRIBUTED_MAP_CACHE, "dmc");
+ runner.setProperty(DeduplicateRecord.DEDUPLICATION_STRATEGY,
DeduplicateRecord.OPTION_MULTIPLE_FILES.getValue());
+ runner.setProperty(DeduplicateRecord.RECORD_HASHING_ALGORITHM,
DeduplicateRecord.NONE_ALGORITHM_VALUE);
+ runner.enableControllerService(dmc);
+ runner.assertValid();
+
dmc.put(FIRST_KEY, true, null, null);
dmc.put(SECOND_KEY, true, null, null);
@@ -225,16 +341,17 @@ public class TestDeduplicateRecord {
@Test
public void testDeduplicateWithDMCAndCacheIdentifier() throws Exception {
- DistributedMapCacheClient dmc = new MockCacheService<>();
+ DistributedMapCacheClient dmc = new MockCacheService();
runner.addControllerService("dmc", dmc);
runner.setProperty(DeduplicateRecord.DISTRIBUTED_MAP_CACHE, "dmc");
runner.setProperty(DeduplicateRecord.DEDUPLICATION_STRATEGY,
DeduplicateRecord.OPTION_MULTIPLE_FILES.getValue());
- runner.setProperty(DeduplicateRecord.CACHE_IDENTIFIER,
"concat('${user.name}', '${record.hash.value}')");
+ runner.setProperty(DeduplicateRecord.CACHE_IDENTIFIER,
"${user.name}-${" + DeduplicateRecord.RECORD_HASH_VALUE_ATTRIBUTE + "}");
+ runner.setProperty(DeduplicateRecord.PUT_CACHE_IDENTIFIER, "true");
runner.enableControllerService(dmc);
runner.assertValid();
- dmc.put(String.format("john.smith-%s", FIRST_KEY), true, null, null);
- dmc.put(String.format("john.smith-%s", SECOND_KEY), true, null, null);
+ dmc.put(String.format("john.smith-%s", FIRST_KEY_HASHED), true, null,
null);
+ dmc.put(String.format("john.smith-%s", SECOND_KEY_HASHED), true, null,
null);
reader.addRecord("John", "Q", "Smith");
reader.addRecord("Jack", "Z", "Brown");
@@ -242,7 +359,7 @@ public class TestDeduplicateRecord {
reader.addRecord("Jane", "X", "Doe");
Map<String, String> attrs = new HashMap<>();
- attrs.put("user.name", "john.smith-");
+ attrs.put("user.name", "john.smith");
runner.enqueue("", attrs);
runner.run();
@@ -251,19 +368,19 @@ public class TestDeduplicateRecord {
}
void doCountTests(int failure, int original, int duplicates, int
notDuplicates, int notDupeCount, int dupeCount) {
+ runner.assertTransferCount(DeduplicateRecord.REL_FAILURE, failure);
+ runner.assertTransferCount(DeduplicateRecord.REL_ORIGINAL, original);
runner.assertTransferCount(DeduplicateRecord.REL_DUPLICATE,
duplicates);
runner.assertTransferCount(DeduplicateRecord.REL_NON_DUPLICATE,
notDuplicates);
- runner.assertTransferCount(DeduplicateRecord.REL_ORIGINAL, original);
- runner.assertTransferCount(DeduplicateRecord.REL_FAILURE, failure);
List<MockFlowFile> duplicateFlowFile =
runner.getFlowFilesForRelationship(DeduplicateRecord.REL_DUPLICATE);
if (duplicateFlowFile != null) {
- assertEquals(String.valueOf(dupeCount),
duplicateFlowFile.get(0).getAttribute("record.count"));
+ assertEquals(String.valueOf(dupeCount),
duplicateFlowFile.get(0).getAttribute(DeduplicateRecord.RECORD_COUNT_ATTRIBUTE));
}
List<MockFlowFile> nonDuplicateFlowFile =
runner.getFlowFilesForRelationship(DeduplicateRecord.REL_NON_DUPLICATE);
if (nonDuplicateFlowFile != null) {
- assertEquals(String.valueOf(notDupeCount),
nonDuplicateFlowFile.get(0).getAttribute("record.count"));
+ assertEquals(String.valueOf(notDupeCount),
nonDuplicateFlowFile.get(0).getAttribute(DeduplicateRecord.RECORD_COUNT_ATTRIBUTE));
}
}
diff --git
a/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/java/org/apache/nifi/processors/standard/TestListSFTP.java
b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/java/org/apache/nifi/processors/standard/TestListSFTP.java
index f9077f8fbd..d0abf64e10 100644
---
a/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/java/org/apache/nifi/processors/standard/TestListSFTP.java
+++
b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/java/org/apache/nifi/processors/standard/TestListSFTP.java
@@ -141,7 +141,7 @@ public class TestListSFTP {
runner.setProperty(AbstractListProcessor.RECORD_WRITER,
"csv-record-writer");
runner.setProperty(AbstractListProcessor.LISTING_STRATEGY,
AbstractListProcessor.BY_ENTITIES);
runner.enableControllerService(recordWriter);
- DistributedMapCacheClient dmc = new MockCacheService<>();
+ DistributedMapCacheClient dmc = new MockCacheService();
runner.addControllerService("dmc", dmc);
runner.setProperty(ListedEntityTracker.TRACKING_STATE_CACHE, "dmc");
runner.enableControllerService(dmc);
@@ -158,7 +158,7 @@ public class TestListSFTP {
runner.setProperty(AbstractListProcessor.RECORD_WRITER,
"csv-record-writer");
runner.setProperty(AbstractListProcessor.LISTING_STRATEGY,
AbstractListProcessor.BY_ENTITIES);
runner.enableControllerService(recordWriter);
- DistributedMapCacheClient dmc = new MockCacheService<>();
+ DistributedMapCacheClient dmc = new MockCacheService();
runner.addControllerService("dmc", dmc);
runner.setProperty(ListedEntityTracker.TRACKING_STATE_CACHE, "dmc");
runner.enableControllerService(dmc);
