Apache
/
nifi
mirror of https://github.com/apache/nifi.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

NIFI-13200: Do not allow 'filename' or 'path' attribute to removed from FlowFile 

This closes #8791

Signed-off-by: Joseph Witt <joewitt@apache.org>
This commit is contained in:
Mark Payne 2024-05-09 16:37:28 -04:00 committed by Joseph Witt
parent 72b45f3114
commit 9cf5f947e8
No known key found for this signature in database
GPG Key ID: 9093BF854F811A1A
1 changed files with 47 additions and 36 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

83
nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/controller/repository/StandardProcessSession.java
View File

@ -16,40 +16,6 @@
*/
package org.apache.nifi.controller.repository;
import java.io.BufferedInputStream;
import java.io.BufferedOutputStream;
import java.io.ByteArrayInputStream;
import java.io.Closeable;
import java.io.EOFException;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.nio.file.Files;
import java.nio.file.Path;
import java.util.ArrayList;
import java.util.BitSet;
import java.util.Collection;
import java.util.Collections;
import java.util.Comparator;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.LinkedHashSet;
import java.util.List;
import java.util.Map;
import java.util.NoSuchElementException;
import java.util.Objects;
import java.util.Optional;
import java.util.Set;
import java.util.UUID;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;
import java.util.function.BiConsumer;
import java.util.function.BiFunction;
import java.util.function.Consumer;
import java.util.regex.Pattern;
import java.util.stream.Collectors;
import org.apache.nifi.components.state.Scope;
import org.apache.nifi.components.state.StateManager;
import org.apache.nifi.components.state.StateMap;
@ -104,6 +70,41 @@ import org.apache.nifi.util.FormatUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.io.BufferedInputStream;
import java.io.BufferedOutputStream;
import java.io.ByteArrayInputStream;
import java.io.Closeable;
import java.io.EOFException;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.nio.file.Files;
import java.nio.file.Path;
import java.util.ArrayList;
import java.util.BitSet;
import java.util.Collection;
import java.util.Collections;
import java.util.Comparator;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.LinkedHashSet;
import java.util.List;
import java.util.Map;
import java.util.NoSuchElementException;
import java.util.Objects;
import java.util.Optional;
import java.util.Set;
import java.util.UUID;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;
import java.util.function.BiConsumer;
import java.util.function.BiFunction;
import java.util.function.Consumer;
import java.util.regex.Pattern;
import java.util.stream.Collectors;
/**
* <p>
* Provides a ProcessSession that ensures all accesses, changes and transfers
@ -116,6 +117,12 @@ import org.slf4j.LoggerFactory;
* <p/>
*/
public class StandardProcessSession implements ProcessSession, ProvenanceEventEnricher {
private static final Set<String> REQUIRED_ATTRIBUTES = Set.of(
CoreAttributes.UUID.key(),
CoreAttributes.FILENAME.key(),
CoreAttributes.PATH.key()
);
private static final long VERSION_INCREMENT = 1;
private static final String INITIAL_VERSION = String.valueOf(VERSION_INCREMENT);
private static final AtomicLong idGenerator = new AtomicLong(0L);
@ -2271,7 +2278,7 @@ public class StandardProcessSession implements ProcessSession, ProvenanceEventEn
verifyTaskActive();
flowFile = validateRecordState(flowFile);
if (CoreAttributes.UUID.key().equals(key)) {
if (REQUIRED_ATTRIBUTES.contains(key)) {
return flowFile;
}
@ -2295,6 +2302,10 @@ public class StandardProcessSession implements ProcessSession, ProvenanceEventEn
final Map<String, String> updatedAttrs = new HashMap<>();
for (final String key : keys) {
if (REQUIRED_ATTRIBUTES.contains(key)) {
continue;
}
if (CoreAttributes.UUID.key().equals(key)) {
continue;
}
@ -2321,7 +2332,7 @@ public class StandardProcessSession implements ProcessSession, ProvenanceEventEn
final Map<String, String> removed = new HashMap<>();
for (final String key : curAttrs.keySet()) {
if (CoreAttributes.UUID.key().equals(key)) {
if (REQUIRED_ATTRIBUTES.contains(key)) {
continue;
}