You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@nifi.apache.org by ma...@apache.org on 2016/07/14 13:32:57 UTC
[5/6] nifi git commit: NIFI-1157 searched for and resolved all
remaining references to deprecated items that were clearly addressable.
http://git-wip-us.apache.org/repos/asf/nifi/blob/f987b216/nifi-nar-bundles/nifi-hadoop-bundle/nifi-hdfs-processors/src/main/java/org/apache/nifi/processors/hadoop/ListHDFS.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-hadoop-bundle/nifi-hdfs-processors/src/main/java/org/apache/nifi/processors/hadoop/ListHDFS.java b/nifi-nar-bundles/nifi-hadoop-bundle/nifi-hdfs-processors/src/main/java/org/apache/nifi/processors/hadoop/ListHDFS.java
index f5daef2..4cb8d25 100644
--- a/nifi-nar-bundles/nifi-hadoop-bundle/nifi-hdfs-processors/src/main/java/org/apache/nifi/processors/hadoop/ListHDFS.java
+++ b/nifi-nar-bundles/nifi-hadoop-bundle/nifi-hdfs-processors/src/main/java/org/apache/nifi/processors/hadoop/ListHDFS.java
@@ -31,10 +31,8 @@ import org.apache.nifi.annotation.behavior.WritesAttributes;
import org.apache.nifi.annotation.documentation.CapabilityDescription;
import org.apache.nifi.annotation.documentation.SeeAlso;
import org.apache.nifi.annotation.documentation.Tags;
-import org.apache.nifi.annotation.lifecycle.OnScheduled;
import org.apache.nifi.components.PropertyDescriptor;
import org.apache.nifi.components.state.Scope;
-import org.apache.nifi.components.state.StateManager;
import org.apache.nifi.components.state.StateMap;
import org.apache.nifi.distributed.cache.client.DistributedMapCacheClient;
import org.apache.nifi.flowfile.FlowFile;
@@ -46,8 +44,6 @@ import org.apache.nifi.processor.Relationship;
import org.apache.nifi.processor.exception.ProcessException;
import org.apache.nifi.processor.util.StandardValidators;
import org.apache.nifi.processors.hadoop.util.HDFSListing;
-import org.apache.nifi.processors.hadoop.util.HDFSListing.StateKeys;
-import org.apache.nifi.processors.hadoop.util.StringSerDe;
import org.codehaus.jackson.JsonNode;
import org.codehaus.jackson.JsonParseException;
import org.codehaus.jackson.map.JsonMappingException;
@@ -124,7 +120,6 @@ public class ListHDFS extends AbstractHadoopProcessor {
private volatile long latestTimestampListed = -1L;
private volatile long latestTimestampEmitted = -1L;
- private volatile boolean electedPrimaryNodeSinceLastIteration = false;
private volatile long lastRunTimestamp = -1L;
static final String LISTING_TIMESTAMP_KEY = "listing.timestamp";
@@ -176,80 +171,6 @@ public class ListHDFS extends AbstractHadoopProcessor {
}
/**
- * Transitions state from the Distributed cache service to the state manager. This will be
- * removed in NiFi 1.x
- *
- * @param context the ProcessContext
- * @throws IOException if unable to communicate with state manager or controller service
- */
- @Deprecated
- @OnScheduled
- public void moveStateToStateManager(final ProcessContext context) throws IOException {
- final StateManager stateManager = context.getStateManager();
- final StateMap stateMap = stateManager.getState(Scope.CLUSTER);
-
- // Check if we have already stored state in the cluster state manager.
- if (stateMap.getVersion() == -1L) {
- final HDFSListing serviceListing = getListingFromService(context);
- if (serviceListing != null) {
- context.getStateManager().setState(serviceListing.toMap(), Scope.CLUSTER);
- }
- }
- }
-
- @Deprecated
- private HDFSListing getListingFromService(final ProcessContext context) throws IOException {
- final DistributedMapCacheClient client = context.getProperty(DISTRIBUTED_CACHE_SERVICE).asControllerService(DistributedMapCacheClient.class);
- if (client == null) {
- return null;
- }
-
- final String directory = context.getProperty(DIRECTORY).getValue();
- final String remoteValue = client.get(getKey(directory), new StringSerDe(), new StringSerDe());
- if (remoteValue == null) {
- return null;
- }
-
- try {
- return deserialize(remoteValue);
- } catch (final Exception e) {
- getLogger().error("Failed to retrieve state from Distributed Map Cache because the content that was retrieved could not be understood", e);
- return null;
- }
- }
-
- /**
- * Restores state information from the 'old' style of storing state. This is deprecated and will no longer be supported
- * in the 1.x NiFi baseline
- *
- * @param directory the directory that the listing was performed against
- * @param remoteListing the remote listing
- * @return the minimum timestamp that should be used for new entries
- */
- @Deprecated
- private Long restoreTimestampFromOldStateFormat(final String directory, final HDFSListing remoteListing) {
- // No cluster-wide state has been recovered. Just use whatever values we already have.
- if (remoteListing == null) {
- return latestTimestampListed;
- }
-
- // If our local timestamp is already later than the remote listing's timestamp, use our local info.
- Long minTimestamp = latestTimestampListed;
- if (minTimestamp != null && minTimestamp > remoteListing.getLatestTimestamp().getTime()) {
- return minTimestamp;
- }
-
- // Use the remote listing's information.
- if (minTimestamp == null || electedPrimaryNodeSinceLastIteration) {
- this.latestTimestampListed = remoteListing.getLatestTimestamp().getTime();
- this.latestTimestampEmitted = this.latestTimestampListed;
- }
-
- return minTimestamp;
- }
-
-
- /**
* Determines which of the given FileStatus's describes a File that should be listed.
*
* @param statuses the eligible FileStatus objects that we could potentially list
@@ -339,13 +260,7 @@ public class ListHDFS extends AbstractHadoopProcessor {
} else {
// Determine if state is stored in the 'new' format or the 'old' format
final String emittedString = stateMap.get(EMITTED_TIMESTAMP_KEY);
- if (emittedString == null && stateMap.get(StateKeys.TIMESTAMP) != null) {
- // state is stored in the old format with XML
- final Map<String, String> stateValues = stateMap.toMap();
- final HDFSListing stateListing = HDFSListing.fromMap(stateValues);
- getLogger().debug("Found old-style state stored");
- restoreTimestampFromOldStateFormat(directory, stateListing);
- } else if (emittedString == null) {
+ if (emittedString == null) {
latestTimestampEmitted = -1L;
latestTimestampListed = -1L;
getLogger().debug("Found no recognized state keys; assuming no relevant state and resetting listing/emitted time to -1");
http://git-wip-us.apache.org/repos/asf/nifi/blob/f987b216/nifi-nar-bundles/nifi-hbase-bundle/nifi-hbase-processors/src/main/java/org/apache/nifi/hbase/GetHBase.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-hbase-bundle/nifi-hbase-processors/src/main/java/org/apache/nifi/hbase/GetHBase.java b/nifi-nar-bundles/nifi-hbase-bundle/nifi-hbase-processors/src/main/java/org/apache/nifi/hbase/GetHBase.java
index 3cd81a3..6002c3c 100644
--- a/nifi-nar-bundles/nifi-hbase-bundle/nifi-hbase-processors/src/main/java/org/apache/nifi/hbase/GetHBase.java
+++ b/nifi-nar-bundles/nifi-hbase-bundle/nifi-hbase-processors/src/main/java/org/apache/nifi/hbase/GetHBase.java
@@ -34,6 +34,7 @@ import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
+import java.util.concurrent.atomic.AtomicReference;
import java.util.regex.Matcher;
import java.util.regex.Pattern;
@@ -73,7 +74,6 @@ import org.apache.nifi.processor.Relationship;
import org.apache.nifi.processor.exception.ProcessException;
import org.apache.nifi.processor.io.OutputStreamCallback;
import org.apache.nifi.processor.util.StandardValidators;
-import org.apache.nifi.util.ObjectHolder;
@TriggerWhenEmpty
@TriggerSerially
@@ -274,8 +274,8 @@ public class GetHBase extends AbstractProcessor {
final Map<String, Set<String>> cellsMatchingTimestamp = new HashMap<>();
- final ObjectHolder<Long> rowsPulledHolder = new ObjectHolder<>(0L);
- final ObjectHolder<Long> latestTimestampHolder = new ObjectHolder<>(minTime);
+ final AtomicReference<Long> rowsPulledHolder = new AtomicReference<>(0L);
+ final AtomicReference<Long> latestTimestampHolder = new AtomicReference<>(minTime);
hBaseClientService.scan(tableName, columns, filterExpression, minTime, new ResultHandler() {
http://git-wip-us.apache.org/repos/asf/nifi/blob/f987b216/nifi-nar-bundles/nifi-hbase-bundle/nifi-hbase-processors/src/main/java/org/apache/nifi/hbase/PutHBaseJSON.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-hbase-bundle/nifi-hbase-processors/src/main/java/org/apache/nifi/hbase/PutHBaseJSON.java b/nifi-nar-bundles/nifi-hbase-bundle/nifi-hbase-processors/src/main/java/org/apache/nifi/hbase/PutHBaseJSON.java
index 9a57d6e..4c4c207 100644
--- a/nifi-nar-bundles/nifi-hbase-bundle/nifi-hbase-processors/src/main/java/org/apache/nifi/hbase/PutHBaseJSON.java
+++ b/nifi-nar-bundles/nifi-hbase-bundle/nifi-hbase-processors/src/main/java/org/apache/nifi/hbase/PutHBaseJSON.java
@@ -26,6 +26,7 @@ import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
import java.util.Set;
+import java.util.concurrent.atomic.AtomicReference;
import org.apache.commons.lang3.StringUtils;
import org.apache.nifi.annotation.behavior.EventDriven;
@@ -46,7 +47,6 @@ import org.apache.nifi.processor.Relationship;
import org.apache.nifi.processor.exception.ProcessException;
import org.apache.nifi.processor.io.InputStreamCallback;
import org.apache.nifi.processor.util.StandardValidators;
-import org.apache.nifi.util.ObjectHolder;
import org.codehaus.jackson.JsonNode;
import org.codehaus.jackson.map.ObjectMapper;
@@ -166,7 +166,7 @@ public class PutHBaseJSON extends AbstractPutHBase {
// Parse the JSON document
final ObjectMapper mapper = new ObjectMapper();
- final ObjectHolder<JsonNode> rootNodeRef = new ObjectHolder<>(null);
+ final AtomicReference<JsonNode> rootNodeRef = new AtomicReference<>(null);
try {
session.read(flowFile, new InputStreamCallback() {
@Override
@@ -189,13 +189,13 @@ public class PutHBaseJSON extends AbstractPutHBase {
}
final Collection<PutColumn> columns = new ArrayList<>();
- final ObjectHolder<String> rowIdHolder = new ObjectHolder<>(null);
+ final AtomicReference<String> rowIdHolder = new AtomicReference<>(null);
// convert each field/value to a column for the put, skip over nulls and arrays
final Iterator<String> fieldNames = rootNode.getFieldNames();
while (fieldNames.hasNext()) {
final String fieldName = fieldNames.next();
- final ObjectHolder<byte[]> fieldValueHolder = new ObjectHolder<>(null);
+ final AtomicReference<byte[]> fieldValueHolder = new AtomicReference<>(null);
final JsonNode fieldNode = rootNode.get(fieldName);
if (fieldNode.isNull()) {
http://git-wip-us.apache.org/repos/asf/nifi/blob/f987b216/nifi-nar-bundles/nifi-hive-bundle/nifi-hive-processors/src/main/java/org/apache/nifi/processors/hive/SelectHiveQL.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-hive-bundle/nifi-hive-processors/src/main/java/org/apache/nifi/processors/hive/SelectHiveQL.java b/nifi-nar-bundles/nifi-hive-bundle/nifi-hive-processors/src/main/java/org/apache/nifi/processors/hive/SelectHiveQL.java
index 77ded36..9c0ebef 100644
--- a/nifi-nar-bundles/nifi-hive-bundle/nifi-hive-processors/src/main/java/org/apache/nifi/processors/hive/SelectHiveQL.java
+++ b/nifi-nar-bundles/nifi-hive-bundle/nifi-hive-processors/src/main/java/org/apache/nifi/processors/hive/SelectHiveQL.java
@@ -28,6 +28,7 @@ import java.util.HashSet;
import java.util.List;
import java.util.Set;
import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicLong;
import org.apache.nifi.annotation.behavior.EventDriven;
import org.apache.nifi.annotation.behavior.InputRequirement;
@@ -47,7 +48,6 @@ import org.apache.nifi.processor.Relationship;
import org.apache.nifi.processor.exception.ProcessException;
import org.apache.nifi.processor.io.OutputStreamCallback;
import org.apache.nifi.processor.util.StandardValidators;
-import org.apache.nifi.util.LongHolder;
import org.apache.nifi.util.StopWatch;
import org.apache.nifi.util.hive.HiveJdbcCommon;
@@ -157,7 +157,7 @@ public class SelectHiveQL extends AbstractHiveQLProcessor {
try (final Connection con = dbcpService.getConnection();
final Statement st = con.createStatement()) {
- final LongHolder nrOfRows = new LongHolder(0L);
+ final AtomicLong nrOfRows = new AtomicLong(0L);
if (fileToProcess == null) {
fileToProcess = session.create();
}
@@ -182,7 +182,7 @@ public class SelectHiveQL extends AbstractHiveQLProcessor {
});
// set attribute how many rows were selected
- fileToProcess = session.putAttribute(fileToProcess, RESULT_ROW_COUNT, nrOfRows.get().toString());
+ fileToProcess = session.putAttribute(fileToProcess, RESULT_ROW_COUNT, String.valueOf(nrOfRows.get()));
// Set MIME type on output document and add extension
if (AVRO.equals(outputFormat)) {
http://git-wip-us.apache.org/repos/asf/nifi/blob/f987b216/nifi-nar-bundles/nifi-kite-bundle/nifi-kite-processors/src/main/java/org/apache/nifi/processors/kite/ConvertAvroSchema.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-kite-bundle/nifi-kite-processors/src/main/java/org/apache/nifi/processors/kite/ConvertAvroSchema.java b/nifi-nar-bundles/nifi-kite-bundle/nifi-kite-processors/src/main/java/org/apache/nifi/processors/kite/ConvertAvroSchema.java
index ffcd653..a8244d2 100644
--- a/nifi-nar-bundles/nifi-kite-bundle/nifi-kite-processors/src/main/java/org/apache/nifi/processors/kite/ConvertAvroSchema.java
+++ b/nifi-nar-bundles/nifi-kite-bundle/nifi-kite-processors/src/main/java/org/apache/nifi/processors/kite/ConvertAvroSchema.java
@@ -53,7 +53,6 @@ import org.apache.nifi.processor.Relationship;
import org.apache.nifi.processor.exception.ProcessException;
import org.apache.nifi.processor.io.StreamCallback;
import org.apache.nifi.processors.kite.AvroRecordConverter.AvroConversionException;
-import org.apache.nifi.util.LongHolder;
import org.kitesdk.data.DatasetException;
import org.kitesdk.data.DatasetIOException;
import org.kitesdk.data.SchemaNotFoundException;
@@ -63,6 +62,7 @@ import com.google.common.annotations.VisibleForTesting;
import com.google.common.collect.ImmutableList;
import com.google.common.collect.ImmutableSet;
import com.google.common.collect.Lists;
+import java.util.concurrent.atomic.AtomicLong;
@Tags({ "avro", "convert", "kite" })
@CapabilityDescription("Convert records from one Avro schema to another, including support for flattening and simple type conversions")
@@ -291,7 +291,7 @@ public class ConvertAvroSchema extends AbstractKiteProcessor {
failureWriter.setCodec(CodecFactory.snappyCodec());
try {
- final LongHolder written = new LongHolder(0L);
+ final AtomicLong written = new AtomicLong(0L);
final FailureTracker failures = new FailureTracker();
final List<Record> badRecords = Lists.newLinkedList();
http://git-wip-us.apache.org/repos/asf/nifi/blob/f987b216/nifi-nar-bundles/nifi-kite-bundle/nifi-kite-processors/src/main/java/org/apache/nifi/processors/kite/ConvertCSVToAvro.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-kite-bundle/nifi-kite-processors/src/main/java/org/apache/nifi/processors/kite/ConvertCSVToAvro.java b/nifi-nar-bundles/nifi-kite-bundle/nifi-kite-processors/src/main/java/org/apache/nifi/processors/kite/ConvertCSVToAvro.java
index 43b33ff..de4130f 100644
--- a/nifi-nar-bundles/nifi-kite-bundle/nifi-kite-processors/src/main/java/org/apache/nifi/processors/kite/ConvertCSVToAvro.java
+++ b/nifi-nar-bundles/nifi-kite-bundle/nifi-kite-processors/src/main/java/org/apache/nifi/processors/kite/ConvertCSVToAvro.java
@@ -47,7 +47,6 @@ import org.apache.nifi.processor.Relationship;
import org.apache.nifi.processor.exception.ProcessException;
import org.apache.nifi.processor.io.StreamCallback;
import org.apache.nifi.processor.util.StandardValidators;
-import org.apache.nifi.util.LongHolder;
import org.kitesdk.data.DatasetException;
import org.kitesdk.data.DatasetIOException;
import org.kitesdk.data.DatasetRecordException;
@@ -59,6 +58,7 @@ import org.kitesdk.data.spi.filesystem.CSVProperties;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.collect.ImmutableList;
import com.google.common.collect.ImmutableSet;
+import java.util.concurrent.atomic.AtomicLong;
@Tags({"kite", "csv", "avro"})
@InputRequirement(Requirement.INPUT_REQUIRED)
@@ -224,7 +224,7 @@ public class ConvertCSVToAvro extends AbstractKiteProcessor {
writer.setCodec(CodecFactory.snappyCodec());
try {
- final LongHolder written = new LongHolder(0L);
+ final AtomicLong written = new AtomicLong(0L);
final FailureTracker failures = new FailureTracker();
FlowFile badRecords = session.clone(incomingCSV);
http://git-wip-us.apache.org/repos/asf/nifi/blob/f987b216/nifi-nar-bundles/nifi-kite-bundle/nifi-kite-processors/src/main/java/org/apache/nifi/processors/kite/ConvertJSONToAvro.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-kite-bundle/nifi-kite-processors/src/main/java/org/apache/nifi/processors/kite/ConvertJSONToAvro.java b/nifi-nar-bundles/nifi-kite-bundle/nifi-kite-processors/src/main/java/org/apache/nifi/processors/kite/ConvertJSONToAvro.java
index af120bf..6245362 100644
--- a/nifi-nar-bundles/nifi-kite-bundle/nifi-kite-processors/src/main/java/org/apache/nifi/processors/kite/ConvertJSONToAvro.java
+++ b/nifi-nar-bundles/nifi-kite-bundle/nifi-kite-processors/src/main/java/org/apache/nifi/processors/kite/ConvertJSONToAvro.java
@@ -39,7 +39,6 @@ import org.apache.nifi.processor.ProcessSession;
import org.apache.nifi.processor.Relationship;
import org.apache.nifi.processor.exception.ProcessException;
import org.apache.nifi.processor.io.StreamCallback;
-import org.apache.nifi.util.LongHolder;
import org.kitesdk.data.DatasetException;
import org.kitesdk.data.DatasetIOException;
import org.kitesdk.data.DatasetRecordException;
@@ -50,6 +49,7 @@ import org.kitesdk.data.spi.filesystem.JSONFileReader;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.collect.ImmutableList;
import com.google.common.collect.ImmutableSet;
+import java.util.concurrent.atomic.AtomicLong;
@Tags({"kite", "json", "avro"})
@InputRequirement(Requirement.INPUT_REQUIRED)
@@ -132,7 +132,7 @@ public class ConvertJSONToAvro extends AbstractKiteProcessor {
writer.setCodec(CodecFactory.snappyCodec());
try {
- final LongHolder written = new LongHolder(0L);
+ final AtomicLong written = new AtomicLong(0L);
final FailureTracker failures = new FailureTracker();
FlowFile badRecords = session.clone(incomingJSON);
http://git-wip-us.apache.org/repos/asf/nifi/blob/f987b216/nifi-nar-bundles/nifi-media-bundle/nifi-media-processors/src/main/java/org/apache/nifi/processors/image/ExtractImageMetadata.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-media-bundle/nifi-media-processors/src/main/java/org/apache/nifi/processors/image/ExtractImageMetadata.java b/nifi-nar-bundles/nifi-media-bundle/nifi-media-processors/src/main/java/org/apache/nifi/processors/image/ExtractImageMetadata.java
index 1dd9241..06dd52b 100644
--- a/nifi-nar-bundles/nifi-media-bundle/nifi-media-processors/src/main/java/org/apache/nifi/processors/image/ExtractImageMetadata.java
+++ b/nifi-nar-bundles/nifi-media-bundle/nifi-media-processors/src/main/java/org/apache/nifi/processors/image/ExtractImageMetadata.java
@@ -44,13 +44,13 @@ import org.apache.nifi.processor.Relationship;
import org.apache.nifi.processor.exception.ProcessException;
import org.apache.nifi.processor.io.InputStreamCallback;
import org.apache.nifi.processor.util.StandardValidators;
-import org.apache.nifi.util.ObjectHolder;
import com.drew.imaging.ImageMetadataReader;
import com.drew.imaging.ImageProcessingException;
import com.drew.metadata.Directory;
import com.drew.metadata.Metadata;
import com.drew.metadata.Tag;
+import java.util.concurrent.atomic.AtomicReference;
@InputRequirement(Requirement.INPUT_REQUIRED)
@Tags({"Exif", "Exchangeable", "image", "file", "format", "JPG", "GIF", "PNG", "BMP", "metadata","IPTC", "XMP"})
@@ -116,7 +116,7 @@ public class ExtractImageMetadata extends AbstractProcessor {
}
final ComponentLog logger = this.getLogger();
- final ObjectHolder<Metadata> value = new ObjectHolder<>(null);
+ final AtomicReference<Metadata> value = new AtomicReference<>(null);
final Integer max = context.getProperty(MAX_NUMBER_OF_ATTRIBUTES).asInteger();
try {
http://git-wip-us.apache.org/repos/asf/nifi/blob/f987b216/nifi-nar-bundles/nifi-provenance-repository-bundle/nifi-persistent-provenance-repository/src/main/java/org/apache/nifi/provenance/PersistentProvenanceRepository.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-provenance-repository-bundle/nifi-persistent-provenance-repository/src/main/java/org/apache/nifi/provenance/PersistentProvenanceRepository.java b/nifi-nar-bundles/nifi-provenance-repository-bundle/nifi-persistent-provenance-repository/src/main/java/org/apache/nifi/provenance/PersistentProvenanceRepository.java
index c6c4b15..0d7886d 100644
--- a/nifi-nar-bundles/nifi-provenance-repository-bundle/nifi-persistent-provenance-repository/src/main/java/org/apache/nifi/provenance/PersistentProvenanceRepository.java
+++ b/nifi-nar-bundles/nifi-provenance-repository-bundle/nifi-persistent-provenance-repository/src/main/java/org/apache/nifi/provenance/PersistentProvenanceRepository.java
@@ -113,7 +113,6 @@ import java.util.stream.Collectors;
public class PersistentProvenanceRepository implements ProvenanceEventRepository {
- public static final String DEPRECATED_CLASS_NAME = "nifi.controller.repository.provenance.PersistentProvenanceRepository";
public static final String EVENT_CATEGORY = "Provenance Repository";
private static final String FILE_EXTENSION = ".prov";
private static final String TEMP_FILE_SUFFIX = ".prov.part";
http://git-wip-us.apache.org/repos/asf/nifi/blob/f987b216/nifi-nar-bundles/nifi-provenance-repository-bundle/nifi-volatile-provenance-repository/src/main/java/org/apache/nifi/provenance/VolatileProvenanceRepository.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-provenance-repository-bundle/nifi-volatile-provenance-repository/src/main/java/org/apache/nifi/provenance/VolatileProvenanceRepository.java b/nifi-nar-bundles/nifi-provenance-repository-bundle/nifi-volatile-provenance-repository/src/main/java/org/apache/nifi/provenance/VolatileProvenanceRepository.java
index 6e7c2f2..7445c5d 100644
--- a/nifi-nar-bundles/nifi-provenance-repository-bundle/nifi-volatile-provenance-repository/src/main/java/org/apache/nifi/provenance/VolatileProvenanceRepository.java
+++ b/nifi-nar-bundles/nifi-provenance-repository-bundle/nifi-volatile-provenance-repository/src/main/java/org/apache/nifi/provenance/VolatileProvenanceRepository.java
@@ -35,7 +35,6 @@ import org.apache.nifi.provenance.search.QueryResult;
import org.apache.nifi.provenance.search.QuerySubmission;
import org.apache.nifi.provenance.search.SearchTerm;
import org.apache.nifi.provenance.search.SearchableField;
-import org.apache.nifi.util.IntegerHolder;
import org.apache.nifi.util.NiFiProperties;
import org.apache.nifi.util.RingBuffer;
import org.apache.nifi.util.RingBuffer.Filter;
@@ -611,7 +610,7 @@ public class VolatileProvenanceRepository implements ProvenanceEventRepository {
@Override
public void run() {
// Retrieve the most recent results and count the total number of matches
- final IntegerHolder matchingCount = new IntegerHolder(0);
+ final AtomicInteger matchingCount = new AtomicInteger(0);
final List<ProvenanceEventRecord> matchingRecords = new ArrayList<>(maxRecords);
ringBuffer.forEach(new ForEachEvaluator<ProvenanceEventRecord>() {
@Override
http://git-wip-us.apache.org/repos/asf/nifi/blob/f987b216/nifi-nar-bundles/nifi-site-to-site-reporting-bundle/nifi-site-to-site-reporting-task/src/main/java/org/apache/nifi/reporting/SiteToSiteProvenanceReportingTask.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-site-to-site-reporting-bundle/nifi-site-to-site-reporting-task/src/main/java/org/apache/nifi/reporting/SiteToSiteProvenanceReportingTask.java b/nifi-nar-bundles/nifi-site-to-site-reporting-bundle/nifi-site-to-site-reporting-task/src/main/java/org/apache/nifi/reporting/SiteToSiteProvenanceReportingTask.java
index 268b734..a6eb662 100644
--- a/nifi-nar-bundles/nifi-site-to-site-reporting-bundle/nifi-site-to-site-reporting-task/src/main/java/org/apache/nifi/reporting/SiteToSiteProvenanceReportingTask.java
+++ b/nifi-nar-bundles/nifi-site-to-site-reporting-bundle/nifi-site-to-site-reporting-task/src/main/java/org/apache/nifi/reporting/SiteToSiteProvenanceReportingTask.java
@@ -49,10 +49,8 @@ import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
-import java.util.HashSet;
import java.util.List;
import java.util.Map;
-import java.util.Set;
import java.util.TimeZone;
import java.util.UUID;
import java.util.concurrent.TimeUnit;
http://git-wip-us.apache.org/repos/asf/nifi/blob/f987b216/nifi-nar-bundles/nifi-solr-bundle/nifi-solr-processors/src/main/java/org/apache/nifi/processors/solr/PutSolrContentStream.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-solr-bundle/nifi-solr-processors/src/main/java/org/apache/nifi/processors/solr/PutSolrContentStream.java b/nifi-nar-bundles/nifi-solr-bundle/nifi-solr-processors/src/main/java/org/apache/nifi/processors/solr/PutSolrContentStream.java
index ca16286..76a70d4 100644
--- a/nifi-nar-bundles/nifi-solr-bundle/nifi-solr-processors/src/main/java/org/apache/nifi/processors/solr/PutSolrContentStream.java
+++ b/nifi-nar-bundles/nifi-solr-bundle/nifi-solr-processors/src/main/java/org/apache/nifi/processors/solr/PutSolrContentStream.java
@@ -32,6 +32,7 @@ import java.util.Set;
import java.util.SortedMap;
import java.util.TreeMap;
import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicReference;
import org.apache.nifi.annotation.behavior.DynamicProperty;
import org.apache.nifi.annotation.behavior.InputRequirement;
@@ -47,7 +48,6 @@ import org.apache.nifi.processor.Relationship;
import org.apache.nifi.processor.exception.ProcessException;
import org.apache.nifi.processor.io.InputStreamCallback;
import org.apache.nifi.processor.util.StandardValidators;
-import org.apache.nifi.util.ObjectHolder;
import org.apache.nifi.util.StopWatch;
import org.apache.solr.client.solrj.SolrServerException;
import org.apache.solr.client.solrj.request.ContentStreamUpdateRequest;
@@ -166,8 +166,8 @@ public class PutSolrContentStream extends SolrProcessor {
return;
}
- final ObjectHolder<Exception> error = new ObjectHolder<>(null);
- final ObjectHolder<Exception> connectionError = new ObjectHolder<>(null);
+ final AtomicReference<Exception> error = new AtomicReference<>(null);
+ final AtomicReference<Exception> connectionError = new AtomicReference<>(null);
final boolean isSolrCloud = SOLR_TYPE_CLOUD.equals(context.getProperty(SOLR_TYPE).getValue());
final String collection = context.getProperty(COLLECTION).evaluateAttributeExpressions(flowFile).getValue();
http://git-wip-us.apache.org/repos/asf/nifi/blob/f987b216/nifi-nar-bundles/nifi-splunk-bundle/nifi-splunk-processors/src/main/java/org/apache/nifi/processors/splunk/PutSplunk.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-splunk-bundle/nifi-splunk-processors/src/main/java/org/apache/nifi/processors/splunk/PutSplunk.java b/nifi-nar-bundles/nifi-splunk-bundle/nifi-splunk-processors/src/main/java/org/apache/nifi/processors/splunk/PutSplunk.java
index c461b0f..9e4aa12 100644
--- a/nifi-nar-bundles/nifi-splunk-bundle/nifi-splunk-processors/src/main/java/org/apache/nifi/processors/splunk/PutSplunk.java
+++ b/nifi-nar-bundles/nifi-splunk-bundle/nifi-splunk-processors/src/main/java/org/apache/nifi/processors/splunk/PutSplunk.java
@@ -39,7 +39,6 @@ import org.apache.nifi.stream.io.ByteArrayOutputStream;
import org.apache.nifi.stream.io.ByteCountingInputStream;
import org.apache.nifi.stream.io.StreamUtils;
import org.apache.nifi.stream.io.util.NonThreadSafeCircularBuffer;
-import org.apache.nifi.util.LongHolder;
import javax.net.ssl.SSLContext;
import java.io.IOException;
@@ -50,6 +49,7 @@ import java.util.Arrays;
import java.util.Collection;
import java.util.List;
import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicLong;
@InputRequirement(InputRequirement.Requirement.INPUT_REQUIRED)
@Tags({"splunk", "logs", "tcp", "udp"})
@@ -219,7 +219,7 @@ public class PutSplunk extends AbstractPutEventProcessor {
// some pattern. We can use this to search for the delimiter as we read through the stream of bytes in the FlowFile
final NonThreadSafeCircularBuffer buffer = new NonThreadSafeCircularBuffer(delimiterBytes);
- final LongHolder messagesSent = new LongHolder(0L);
+ final AtomicLong messagesSent = new AtomicLong(0L);
final FlowFileMessageBatch messageBatch = new FlowFileMessageBatch(session, flowFile);
activeBatches.add(messageBatch);
http://git-wip-us.apache.org/repos/asf/nifi/blob/f987b216/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/AbstractJsonPathProcessor.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/AbstractJsonPathProcessor.java b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/AbstractJsonPathProcessor.java
index 2f1ff43..21172e9 100644
--- a/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/AbstractJsonPathProcessor.java
+++ b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/AbstractJsonPathProcessor.java
@@ -31,7 +31,6 @@ import org.apache.nifi.processor.ProcessSession;
import org.apache.nifi.processor.io.InputStreamCallback;
import org.apache.nifi.processors.standard.util.JsonPathExpressionValidator;
import org.apache.nifi.stream.io.BufferedInputStream;
-import org.apache.nifi.util.ObjectHolder;
import java.io.IOException;
import java.io.InputStream;
@@ -39,6 +38,7 @@ import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Objects;
+import java.util.concurrent.atomic.AtomicReference;
/**
* Provides common functionality used for processors interacting and manipulating JSON data via JsonPath.
@@ -73,7 +73,7 @@ public abstract class AbstractJsonPathProcessor extends AbstractProcessor {
static DocumentContext validateAndEstablishJsonContext(ProcessSession processSession, FlowFile flowFile) {
// Parse the document once into an associated context to support multiple path evaluations if specified
- final ObjectHolder<DocumentContext> contextHolder = new ObjectHolder<>(null);
+ final AtomicReference<DocumentContext> contextHolder = new AtomicReference<>(null);
processSession.read(flowFile, new InputStreamCallback() {
@Override
public void process(InputStream in) throws IOException {
http://git-wip-us.apache.org/repos/asf/nifi/blob/f987b216/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/BinFiles.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/BinFiles.java b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/BinFiles.java
deleted file mode 100644
index 3df4317..0000000
--- a/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/BinFiles.java
+++ /dev/null
@@ -1,354 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.apache.nifi.processors.standard;
-
-import java.io.IOException;
-import java.util.ArrayList;
-import java.util.Collection;
-import java.util.List;
-import java.util.Queue;
-import java.util.concurrent.LinkedBlockingQueue;
-import java.util.concurrent.TimeUnit;
-
-import org.apache.nifi.annotation.lifecycle.OnScheduled;
-import org.apache.nifi.annotation.lifecycle.OnStopped;
-import org.apache.nifi.components.PropertyDescriptor;
-import org.apache.nifi.components.ValidationContext;
-import org.apache.nifi.components.ValidationResult;
-import org.apache.nifi.flowfile.FlowFile;
-import org.apache.nifi.logging.ComponentLog;
-import org.apache.nifi.processor.AbstractSessionFactoryProcessor;
-import org.apache.nifi.processor.DataUnit;
-import org.apache.nifi.processor.ProcessContext;
-import org.apache.nifi.processor.ProcessSession;
-import org.apache.nifi.processor.ProcessSessionFactory;
-import org.apache.nifi.processor.Relationship;
-import org.apache.nifi.processor.exception.ProcessException;
-import org.apache.nifi.processor.util.StandardValidators;
-import org.apache.nifi.processors.standard.util.Bin;
-import org.apache.nifi.processors.standard.util.BinManager;
-import org.apache.nifi.processors.standard.util.FlowFileSessionWrapper;
-
-/**
- * Base class for file-binning processors, including MergeContent.
- *
- * @deprecated As of release 0.5.0, replaced by
- * {@link org.apache.nifi.processor.util.bin.BinFiles}
- */
-@Deprecated
-public abstract class BinFiles extends AbstractSessionFactoryProcessor {
-
- public static final PropertyDescriptor MIN_SIZE = new PropertyDescriptor.Builder()
- .name("Minimum Group Size")
- .description("The minimum size of for the bundle")
- .required(true)
- .defaultValue("0 B")
- .addValidator(StandardValidators.DATA_SIZE_VALIDATOR)
- .build();
- public static final PropertyDescriptor MAX_SIZE = new PropertyDescriptor.Builder()
- .name("Maximum Group Size")
- .description("The maximum size for the bundle. If not specified, there is no maximum.")
- .required(false)
- .addValidator(StandardValidators.DATA_SIZE_VALIDATOR)
- .build();
-
- public static final PropertyDescriptor MIN_ENTRIES = new PropertyDescriptor.Builder()
- .name("Minimum Number of Entries")
- .description("The minimum number of files to include in a bundle")
- .required(true)
- .defaultValue("1")
- .addValidator(StandardValidators.POSITIVE_INTEGER_VALIDATOR)
- .build();
- public static final PropertyDescriptor MAX_ENTRIES = new PropertyDescriptor.Builder()
- .name("Maximum Number of Entries")
- .description("The maximum number of files to include in a bundle. If not specified, there is no maximum.")
- .required(false)
- .addValidator(StandardValidators.POSITIVE_INTEGER_VALIDATOR)
- .build();
-
- public static final PropertyDescriptor MAX_BIN_COUNT = new PropertyDescriptor.Builder()
- .name("Maximum number of Bins")
- .description("Specifies the maximum number of bins that can be held in memory at any one time")
- .defaultValue("100")
- .required(true)
- .addValidator(StandardValidators.POSITIVE_INTEGER_VALIDATOR)
- .build();
-
- public static final PropertyDescriptor MAX_BIN_AGE = new PropertyDescriptor.Builder()
- .name("Max Bin Age")
- .description("The maximum age of a Bin that will trigger a Bin to be complete. Expected format is <duration> <time unit> "
- + "where <duration> is a positive integer and time unit is one of seconds, minutes, hours")
- .required(false)
- .addValidator(StandardValidators.createTimePeriodValidator(1, TimeUnit.SECONDS, Integer.MAX_VALUE, TimeUnit.SECONDS))
- .build();
-
- public static final Relationship REL_ORIGINAL = new Relationship.Builder()
- .name("original")
- .description("The FlowFiles that were used to create the bundle")
- .build();
- public static final Relationship REL_FAILURE = new Relationship.Builder()
- .name("failure")
- .description("If the bundle cannot be created, all FlowFiles that would have been used to created the bundle will be transferred to failure")
- .build();
-
- private final BinManager binManager = new BinManager();
- private final Queue<Bin> readyBins = new LinkedBlockingQueue<>();
-
- @OnStopped
- public final void resetState() {
- binManager.purge();
-
- Bin bin;
- while ((bin = readyBins.poll()) != null) {
- for (final FlowFileSessionWrapper wrapper : bin.getContents()) {
- wrapper.getSession().rollback();
- }
- }
- }
-
- /**
- * Allows general pre-processing of a flow file before it is offered to a bin. This is called before getGroupId().
- *
- * @param context context
- * @param session session
- * @param flowFile flowFile
- * @return The flow file, possibly altered
- */
- protected abstract FlowFile preprocessFlowFile(final ProcessContext context, final ProcessSession session, final FlowFile flowFile);
-
- /**
- * Returns a group ID representing a bin. This allows flow files to be binned into like groups.
- *
- * @param context context
- * @param flowFile flowFile
- * @return The appropriate group ID
- */
- protected abstract String getGroupId(final ProcessContext context, final FlowFile flowFile);
-
- /**
- * Performs any additional setup of the bin manager. Called during the OnScheduled phase.
- *
- * @param binManager The bin manager
- * @param context context
- */
- protected abstract void setUpBinManager(BinManager binManager, ProcessContext context);
-
- /**
- * Processes a single bin. Implementing class is responsible for committing each session
- *
- * @param unmodifiableBin A reference to a single bin of flow file/session wrappers
- * @param binContents A copy of the contents of the bin
- * @param context The context
- * @param session The session that created the bin
- * @return Return true if the input bin was already committed. E.g., in case of a failure, the implementation may choose to transfer all binned files to Failure and commit their sessions. If
- * false, the processBins() method will transfer the files to Original and commit the sessions
- *
- * @throws ProcessException if any problem arises while processing a bin of FlowFiles. All flow files in the bin will be transferred to failure and the ProcessSession provided by the 'session'
- * argument rolled back
- */
- protected abstract boolean processBin(Bin unmodifiableBin, List<FlowFileSessionWrapper> binContents, ProcessContext context, ProcessSession session) throws ProcessException;
-
- /**
- * Allows additional custom validation to be done. This will be called from the parent's customValidation method.
- *
- * @param context The context
- * @return Validation results indicating problems
- */
- protected Collection<ValidationResult> additionalCustomValidation(final ValidationContext context) {
- return new ArrayList<>();
- }
-
- @Override
- public final void onTrigger(final ProcessContext context, final ProcessSessionFactory sessionFactory) throws ProcessException {
- final int flowFilesBinned = binFlowFiles(context, sessionFactory);
- getLogger().debug("Binned {} FlowFiles", new Object[]{flowFilesBinned});
-
- if (!isScheduled()) {
- return;
- }
-
- final int binsMigrated = migrateBins(context);
- final int binsProcessed = processBins(context, sessionFactory);
- //If we accomplished nothing then let's yield
- if (flowFilesBinned == 0 && binsMigrated == 0 && binsProcessed == 0) {
- context.yield();
- }
- }
-
- private int migrateBins(final ProcessContext context) {
- int added = 0;
- for (final Bin bin : binManager.removeReadyBins(true)) {
- this.readyBins.add(bin);
- added++;
- }
-
- // if we have created all of the bins that are allowed, go ahead and remove the oldest one. If we don't do
- // this, then we will simply wait for it to expire because we can't get any more FlowFiles into the
- // bins. So we may as well expire it now.
- if (added == 0 && binManager.getBinCount() >= context.getProperty(MAX_BIN_COUNT).asInteger()) {
- final Bin bin = binManager.removeOldestBin();
- if (bin != null) {
- added++;
- this.readyBins.add(bin);
- }
- }
- return added;
- }
-
- private int processBins(final ProcessContext context, final ProcessSessionFactory sessionFactory) {
- final Bin bin = readyBins.poll();
- if (bin == null) {
- return 0;
- }
-
- final List<Bin> bins = new ArrayList<>();
- bins.add(bin);
-
- final ComponentLog logger = getLogger();
- final ProcessSession session = sessionFactory.createSession();
-
- final List<FlowFileSessionWrapper> binCopy = new ArrayList<>(bin.getContents());
-
- boolean binAlreadyCommitted = false;
- try {
- binAlreadyCommitted = this.processBin(bin, binCopy, context, session);
- } catch (final ProcessException e) {
- logger.error("Failed to process bundle of {} files due to {}", new Object[]{binCopy.size(), e});
-
- for (final FlowFileSessionWrapper wrapper : binCopy) {
- wrapper.getSession().transfer(wrapper.getFlowFile(), REL_FAILURE);
- wrapper.getSession().commit();
- }
- session.rollback();
- return 1;
- }
-
- // we first commit the bundle's session before the originals' sessions because if we are restarted or crash
- // between commits, we favor data redundancy over data loss. Since we have no Distributed Transaction capability
- // across multiple sessions, we cannot guarantee atomicity across the sessions
- session.commit();
- // If this bin's session has been committed, move on.
- if (!binAlreadyCommitted) {
- for (final FlowFileSessionWrapper wrapper : bin.getContents()) {
- wrapper.getSession().transfer(wrapper.getFlowFile(), REL_ORIGINAL);
- wrapper.getSession().commit();
- }
- }
-
- return 1;
- }
-
- private int binFlowFiles(final ProcessContext context, final ProcessSessionFactory sessionFactory) {
- int flowFilesBinned = 0;
- while (binManager.getBinCount() <= context.getProperty(MAX_BIN_COUNT).asInteger().intValue()) {
- if (!isScheduled()) {
- break;
- }
-
- final ProcessSession session = sessionFactory.createSession();
- FlowFile flowFile = session.get();
- if (flowFile == null) {
- break;
- }
-
- flowFile = this.preprocessFlowFile(context, session, flowFile);
-
- String groupId = this.getGroupId(context, flowFile);
-
- final boolean binned = binManager.offer(groupId, flowFile, session);
-
- // could not be added to a bin -- probably too large by itself, so create a separate bin for just this guy.
- if (!binned) {
- Bin bin = new Bin(0, Long.MAX_VALUE, 0, Integer.MAX_VALUE, null);
- bin.offer(flowFile, session);
- this.readyBins.add(bin);
- }
-
- flowFilesBinned++;
- }
-
- return flowFilesBinned;
- }
-
- @OnScheduled
- public final void onScheduled(final ProcessContext context) throws IOException {
- binManager.setMinimumSize(context.getProperty(MIN_SIZE).asDataSize(DataUnit.B).longValue());
-
- if (context.getProperty(MAX_BIN_AGE).isSet()) {
- binManager.setMaxBinAge(context.getProperty(MAX_BIN_AGE).asTimePeriod(TimeUnit.SECONDS).intValue());
- } else {
- binManager.setMaxBinAge(Integer.MAX_VALUE);
- }
-
- if (context.getProperty(MAX_SIZE).isSet()) {
- binManager.setMaximumSize(context.getProperty(MAX_SIZE).asDataSize(DataUnit.B).longValue());
- } else {
- binManager.setMaximumSize(Long.MAX_VALUE);
- }
-
- binManager.setMinimumEntries(context.getProperty(MIN_ENTRIES).asInteger());
-
- if (context.getProperty(MAX_ENTRIES).isSet()) {
- binManager.setMaximumEntries(context.getProperty(MAX_ENTRIES).asInteger().intValue());
- } else {
- binManager.setMaximumEntries(Integer.MAX_VALUE);
- }
-
- this.setUpBinManager(binManager, context);
- }
-
- @Override
- protected final Collection<ValidationResult> customValidate(final ValidationContext context) {
- final List<ValidationResult> problems = new ArrayList<>(super.customValidate(context));
-
- final long minBytes = context.getProperty(MIN_SIZE).asDataSize(DataUnit.B).longValue();
- final Double maxBytes = context.getProperty(MAX_SIZE).asDataSize(DataUnit.B);
-
- if (maxBytes != null && maxBytes.longValue() < minBytes) {
- problems.add(
- new ValidationResult.Builder()
- .subject(MIN_SIZE.getName())
- .input(context.getProperty(MIN_SIZE).getValue())
- .valid(false)
- .explanation("Min Size must be less than or equal to Max Size")
- .build()
- );
- }
-
- final Long min = context.getProperty(MIN_ENTRIES).asLong();
- final Long max = context.getProperty(MAX_ENTRIES).asLong();
-
- if (min != null && max != null) {
- if (min > max) {
- problems.add(
- new ValidationResult.Builder().subject(MIN_ENTRIES.getName())
- .input(context.getProperty(MIN_ENTRIES).getValue())
- .valid(false)
- .explanation("Min Entries must be less than or equal to Max Entries")
- .build()
- );
- }
- }
-
- Collection<ValidationResult> otherProblems = this.additionalCustomValidation(context);
- if (otherProblems != null) {
- problems.addAll(otherProblems);
- }
-
- return problems;
- }
-}
http://git-wip-us.apache.org/repos/asf/nifi/blob/f987b216/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/CompressContent.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/CompressContent.java b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/CompressContent.java
index e4e6cf4..1f4dd45 100644
--- a/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/CompressContent.java
+++ b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/CompressContent.java
@@ -28,6 +28,7 @@ import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicReference;
import org.apache.commons.compress.compressors.CompressorStreamFactory;
import org.apache.commons.compress.compressors.bzip2.BZip2CompressorInputStream;
@@ -55,7 +56,6 @@ import org.apache.nifi.processor.io.StreamCallback;
import org.apache.nifi.stream.io.BufferedInputStream;
import org.apache.nifi.stream.io.BufferedOutputStream;
import org.apache.nifi.stream.io.GZIPOutputStream;
-import org.apache.nifi.util.ObjectHolder;
import org.apache.nifi.util.StopWatch;
import org.tukaani.xz.LZMA2Options;
import org.tukaani.xz.XZInputStream;
@@ -203,7 +203,7 @@ public class CompressContent extends AbstractProcessor {
}
final String compressionFormat = compressionFormatValue;
- final ObjectHolder<String> mimeTypeRef = new ObjectHolder<>(null);
+ final AtomicReference<String> mimeTypeRef = new AtomicReference<>(null);
final StopWatch stopWatch = new StopWatch(true);
final String fileExtension;
http://git-wip-us.apache.org/repos/asf/nifi/blob/f987b216/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/ConvertJSONToSQL.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/ConvertJSONToSQL.java b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/ConvertJSONToSQL.java
index 3eb44cb..cbc301a 100755
--- a/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/ConvertJSONToSQL.java
+++ b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/ConvertJSONToSQL.java
@@ -33,6 +33,7 @@ import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.UUID;
+import java.util.concurrent.atomic.AtomicReference;
import org.apache.nifi.annotation.behavior.InputRequirement;
import org.apache.nifi.annotation.behavior.SideEffectFree;
@@ -57,7 +58,6 @@ import org.apache.nifi.processor.exception.ProcessException;
import org.apache.nifi.processor.io.InputStreamCallback;
import org.apache.nifi.processor.io.OutputStreamCallback;
import org.apache.nifi.processor.util.StandardValidators;
-import org.apache.nifi.util.ObjectHolder;
import org.codehaus.jackson.JsonNode;
import org.codehaus.jackson.map.ObjectMapper;
import org.codehaus.jackson.node.ArrayNode;
@@ -276,7 +276,7 @@ public class ConvertJSONToSQL extends AbstractProcessor {
// Parse the JSON document
final ObjectMapper mapper = new ObjectMapper();
- final ObjectHolder<JsonNode> rootNodeRef = new ObjectHolder<>(null);
+ final AtomicReference<JsonNode> rootNodeRef = new AtomicReference<>(null);
try {
session.read(flowFile, new InputStreamCallback() {
@Override
http://git-wip-us.apache.org/repos/asf/nifi/blob/f987b216/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/EvaluateJsonPath.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/EvaluateJsonPath.java b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/EvaluateJsonPath.java
index 18259d1..813a07d 100644
--- a/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/EvaluateJsonPath.java
+++ b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/EvaluateJsonPath.java
@@ -52,12 +52,12 @@ import org.apache.nifi.processor.Relationship;
import org.apache.nifi.processor.exception.ProcessException;
import org.apache.nifi.processor.io.OutputStreamCallback;
import org.apache.nifi.stream.io.BufferedOutputStream;
-import org.apache.nifi.util.ObjectHolder;
import com.jayway.jsonpath.DocumentContext;
import com.jayway.jsonpath.InvalidJsonException;
import com.jayway.jsonpath.JsonPath;
import com.jayway.jsonpath.PathNotFoundException;
+import java.util.concurrent.atomic.AtomicReference;
@EventDriven
@SideEffectFree
@@ -277,7 +277,7 @@ public class EvaluateJsonPath extends AbstractJsonPathProcessor {
final JsonPath jsonPathExp = attributeJsonPathEntry.getValue();
final String pathNotFound = processContext.getProperty(PATH_NOT_FOUND).getValue();
- final ObjectHolder<Object> resultHolder = new ObjectHolder<>(null);
+ final AtomicReference<Object> resultHolder = new AtomicReference<>(null);
try {
final Object result = documentContext.read(jsonPathExp);
if (returnType.equals(RETURN_TYPE_SCALAR) && !isJsonScalar(result)) {
http://git-wip-us.apache.org/repos/asf/nifi/blob/f987b216/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/EvaluateXPath.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/EvaluateXPath.java b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/EvaluateXPath.java
index 10d9af9..4ff7e0e 100644
--- a/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/EvaluateXPath.java
+++ b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/EvaluateXPath.java
@@ -75,7 +75,6 @@ import org.apache.nifi.processor.io.InputStreamCallback;
import org.apache.nifi.processor.io.OutputStreamCallback;
import org.apache.nifi.stream.io.BufferedInputStream;
import org.apache.nifi.stream.io.BufferedOutputStream;
-import org.apache.nifi.util.ObjectHolder;
import org.xml.sax.InputSource;
import net.sf.saxon.lib.NamespaceConstant;
@@ -271,8 +270,8 @@ public class EvaluateXPath extends AbstractProcessor {
flowFileLoop:
for (FlowFile flowFile : flowFiles) {
- final ObjectHolder<Throwable> error = new ObjectHolder<>(null);
- final ObjectHolder<Source> sourceRef = new ObjectHolder<>(null);
+ final AtomicReference<Throwable> error = new AtomicReference<>(null);
+ final AtomicReference<Source> sourceRef = new AtomicReference<>(null);
session.read(flowFile, new InputStreamCallback() {
@Override
@@ -402,7 +401,7 @@ public class EvaluateXPath extends AbstractProcessor {
final ComponentLog logger = getLogger();
- final ObjectHolder<TransformerException> error = new ObjectHolder<>(null);
+ final AtomicReference<TransformerException> error = new AtomicReference<>(null);
transformer.setErrorListener(new ErrorListener() {
@Override
public void warning(final TransformerException exception) throws TransformerException {
http://git-wip-us.apache.org/repos/asf/nifi/blob/f987b216/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/EvaluateXQuery.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/EvaluateXQuery.java b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/EvaluateXQuery.java
index 51a7db0..b8ff2eb 100644
--- a/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/EvaluateXQuery.java
+++ b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/EvaluateXQuery.java
@@ -29,6 +29,7 @@ import java.util.List;
import java.util.Map;
import java.util.Properties;
import java.util.Set;
+import java.util.concurrent.atomic.AtomicReference;
import javax.xml.parsers.DocumentBuilderFactory;
import javax.xml.transform.OutputKeys;
@@ -66,7 +67,6 @@ import org.apache.nifi.processor.io.OutputStreamCallback;
import org.apache.nifi.processor.util.StandardValidators;
import org.apache.nifi.stream.io.BufferedInputStream;
import org.apache.nifi.stream.io.BufferedOutputStream;
-import org.apache.nifi.util.ObjectHolder;
import org.w3c.dom.Document;
import org.xml.sax.InputSource;
@@ -264,8 +264,8 @@ public class EvaluateXQuery extends AbstractProcessor {
return;
}
- final ObjectHolder<Throwable> error = new ObjectHolder<>(null);
- final ObjectHolder<XdmNode> sourceRef = new ObjectHolder<>(null);
+ final AtomicReference<Throwable> error = new AtomicReference<>(null);
+ final AtomicReference<XdmNode> sourceRef = new AtomicReference<>(null);
session.read(flowFile, new InputStreamCallback() {
@Override
http://git-wip-us.apache.org/repos/asf/nifi/blob/f987b216/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/ExecuteSQL.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/ExecuteSQL.java b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/ExecuteSQL.java
index 95893d8..1c514b4 100644
--- a/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/ExecuteSQL.java
+++ b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/ExecuteSQL.java
@@ -29,6 +29,7 @@ import java.util.HashSet;
import java.util.List;
import java.util.Set;
import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicLong;
import org.apache.commons.io.IOUtils;
import org.apache.nifi.annotation.behavior.EventDriven;
@@ -51,7 +52,6 @@ import org.apache.nifi.processor.io.InputStreamCallback;
import org.apache.nifi.processor.io.OutputStreamCallback;
import org.apache.nifi.processor.util.StandardValidators;
import org.apache.nifi.processors.standard.util.JdbcCommon;
-import org.apache.nifi.util.LongHolder;
import org.apache.nifi.util.StopWatch;
@EventDriven
@@ -180,7 +180,7 @@ public class ExecuteSQL extends AbstractProcessor {
try (final Connection con = dbcpService.getConnection();
final Statement st = con.createStatement()) {
st.setQueryTimeout(queryTimeout); // timeout in seconds
- final LongHolder nrOfRows = new LongHolder(0L);
+ final AtomicLong nrOfRows = new AtomicLong(0L);
if (fileToProcess == null) {
fileToProcess = session.create();
}
@@ -198,7 +198,7 @@ public class ExecuteSQL extends AbstractProcessor {
});
// set attribute how many rows were selected
- fileToProcess = session.putAttribute(fileToProcess, RESULT_ROW_COUNT, nrOfRows.get().toString());
+ fileToProcess = session.putAttribute(fileToProcess, RESULT_ROW_COUNT, String.valueOf(nrOfRows.get()));
logger.info("{} contains {} Avro records; transferring to 'success'",
new Object[]{fileToProcess, nrOfRows.get()});
http://git-wip-us.apache.org/repos/asf/nifi/blob/f987b216/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/HashContent.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/HashContent.java b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/HashContent.java
index cea3803..ade251c 100644
--- a/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/HashContent.java
+++ b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/HashContent.java
@@ -27,6 +27,7 @@ import java.util.Collections;
import java.util.HashSet;
import java.util.List;
import java.util.Set;
+import java.util.concurrent.atomic.AtomicReference;
import org.apache.nifi.annotation.behavior.EventDriven;
import org.apache.nifi.annotation.behavior.InputRequirement;
@@ -48,7 +49,6 @@ import org.apache.nifi.processor.io.InputStreamCallback;
import org.apache.nifi.processor.util.StandardValidators;
import org.apache.nifi.stream.io.NullOutputStream;
import org.apache.nifi.stream.io.StreamUtils;
-import org.apache.nifi.util.ObjectHolder;
@EventDriven
@SupportsBatching
@@ -129,7 +129,7 @@ public class HashContent extends AbstractProcessor {
return;
}
- final ObjectHolder<String> hashValueHolder = new ObjectHolder<>(null);
+ final AtomicReference<String> hashValueHolder = new AtomicReference<>(null);
try {
session.read(flowFile, new InputStreamCallback() {
http://git-wip-us.apache.org/repos/asf/nifi/blob/f987b216/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/IdentifyMimeType.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/IdentifyMimeType.java b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/IdentifyMimeType.java
index f4189c3..5965cfd 100644
--- a/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/IdentifyMimeType.java
+++ b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/IdentifyMimeType.java
@@ -22,6 +22,7 @@ import java.io.InputStream;
import java.util.Collections;
import java.util.HashSet;
import java.util.Set;
+import java.util.concurrent.atomic.AtomicReference;
import org.apache.nifi.annotation.behavior.EventDriven;
import org.apache.nifi.annotation.behavior.InputRequirement;
@@ -40,7 +41,6 @@ import org.apache.nifi.processor.ProcessSession;
import org.apache.nifi.processor.ProcessorInitializationContext;
import org.apache.nifi.processor.Relationship;
import org.apache.nifi.processor.io.InputStreamCallback;
-import org.apache.nifi.util.ObjectHolder;
import org.apache.tika.config.TikaConfig;
import org.apache.tika.detect.Detector;
import org.apache.tika.io.TikaInputStream;
@@ -117,7 +117,7 @@ public class IdentifyMimeType extends AbstractProcessor {
}
final ComponentLog logger = getLogger();
- final ObjectHolder<String> mimeTypeRef = new ObjectHolder<>(null);
+ final AtomicReference<String> mimeTypeRef = new AtomicReference<>(null);
final String filename = flowFile.getAttribute(CoreAttributes.FILENAME.key());
session.read(flowFile, new InputStreamCallback() {
http://git-wip-us.apache.org/repos/asf/nifi/blob/f987b216/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/JmsConsumer.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/JmsConsumer.java b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/JmsConsumer.java
index cf968a6..71e2500 100644
--- a/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/JmsConsumer.java
+++ b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/JmsConsumer.java
@@ -41,6 +41,7 @@ import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
import javax.jms.JMSException;
import javax.jms.MapMessage;
@@ -60,7 +61,6 @@ import org.apache.nifi.processors.standard.util.JmsFactory;
import org.apache.nifi.processors.standard.util.JmsProcessingSummary;
import org.apache.nifi.processors.standard.util.WrappedMessageConsumer;
import org.apache.nifi.stream.io.BufferedOutputStream;
-import org.apache.nifi.util.IntegerHolder;
import org.apache.nifi.util.StopWatch;
public abstract class JmsConsumer extends AbstractProcessor {
@@ -181,7 +181,7 @@ public abstract class JmsConsumer extends AbstractProcessor {
throws Exception {
// Currently not very useful, because always one Message == one FlowFile
- final IntegerHolder msgsThisFlowFile = new IntegerHolder(1);
+ final AtomicInteger msgsThisFlowFile = new AtomicInteger(1);
FlowFile flowFile = session.create();
try {
http://git-wip-us.apache.org/repos/asf/nifi/blob/f987b216/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/MergeContent.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/MergeContent.java b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/MergeContent.java
index ce216d7..545559e 100644
--- a/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/MergeContent.java
+++ b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/MergeContent.java
@@ -34,6 +34,7 @@ import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.TreeMap;
+import java.util.concurrent.atomic.AtomicReference;
import java.util.regex.Pattern;
import java.util.zip.ZipEntry;
import java.util.zip.ZipOutputStream;
@@ -85,7 +86,6 @@ import org.apache.nifi.util.FlowFilePackager;
import org.apache.nifi.util.FlowFilePackagerV1;
import org.apache.nifi.util.FlowFilePackagerV2;
import org.apache.nifi.util.FlowFilePackagerV3;
-import org.apache.nifi.util.ObjectHolder;
@SideEffectFree
@TriggerWhenEmpty
@@ -557,7 +557,7 @@ public class MergeContent extends BinFiles {
}
FlowFile bundle = session.create(parentFlowFiles);
- final ObjectHolder<String> bundleMimeTypeRef = new ObjectHolder<>(null);
+ final AtomicReference<String> bundleMimeTypeRef = new AtomicReference<>(null);
bundle = session.write(bundle, new OutputStreamCallback() {
@Override
public void process(final OutputStream out) throws IOException {
@@ -884,8 +884,8 @@ public class MergeContent extends BinFiles {
public FlowFile merge(ProcessContext context, final ProcessSession session, final List<FlowFileSessionWrapper> wrappers) {
final Map<String, byte[]> metadata = new TreeMap<>();
- final ObjectHolder<Schema> schema = new ObjectHolder<>(null);
- final ObjectHolder<String> inputCodec = new ObjectHolder<>(null);
+ final AtomicReference<Schema> schema = new AtomicReference<>(null);
+ final AtomicReference<String> inputCodec = new AtomicReference<>(null);
final DataFileWriter<GenericRecord> writer = new DataFileWriter<>(new GenericDatumWriter<GenericRecord>());
// we don't pass the parents to the #create method because the parents belong to different sessions
http://git-wip-us.apache.org/repos/asf/nifi/blob/f987b216/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/PostHTTP.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/PostHTTP.java b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/PostHTTP.java
index d3713a1..2239995 100644
--- a/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/PostHTTP.java
+++ b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/PostHTTP.java
@@ -117,7 +117,6 @@ import org.apache.nifi.util.FlowFilePackagerV1;
import org.apache.nifi.util.FlowFilePackagerV2;
import org.apache.nifi.util.FlowFilePackagerV3;
import org.apache.nifi.util.FormatUtils;
-import org.apache.nifi.util.ObjectHolder;
import org.apache.nifi.util.StopWatch;
import org.apache.nifi.util.StringUtils;
import com.sun.jersey.api.client.ClientResponse.Status;
@@ -466,7 +465,7 @@ public class PostHTTP extends AbstractProcessor {
CloseableHttpClient client = null;
final String transactionId = UUID.randomUUID().toString();
- final ObjectHolder<String> dnHolder = new ObjectHolder<>("none");
+ final AtomicReference<String> dnHolder = new AtomicReference<>("none");
while (true) {
FlowFile flowFile = session.get();
if (flowFile == null) {
http://git-wip-us.apache.org/repos/asf/nifi/blob/f987b216/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/PutFileTransfer.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/PutFileTransfer.java b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/PutFileTransfer.java
index f41941d..5b9f5b4 100644
--- a/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/PutFileTransfer.java
+++ b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/PutFileTransfer.java
@@ -30,7 +30,6 @@ import org.apache.nifi.processor.io.InputStreamCallback;
import org.apache.nifi.processors.standard.util.FileInfo;
import org.apache.nifi.processors.standard.util.FileTransfer;
import org.apache.nifi.processors.standard.util.SFTPTransfer;
-import org.apache.nifi.util.ObjectHolder;
import org.apache.nifi.util.StopWatch;
import java.io.File;
@@ -40,6 +39,7 @@ import java.util.Collections;
import java.util.HashSet;
import java.util.Set;
import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicReference;
/**
* Base class for PutFTP & PutSFTP
@@ -123,7 +123,7 @@ public abstract class PutFileTransfer<T extends FileTransfer> extends AbstractPr
beforePut(flowFile, context, transfer);
final FlowFile flowFileToTransfer = flowFile;
- final ObjectHolder<String> fullPathRef = new ObjectHolder<>(null);
+ final AtomicReference<String> fullPathRef = new AtomicReference<>(null);
session.read(flowFile, new InputStreamCallback() {
@Override
public void process(final InputStream in) throws IOException {
http://git-wip-us.apache.org/repos/asf/nifi/blob/f987b216/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/PutSyslog.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/PutSyslog.java b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/PutSyslog.java
index 373e402..412d8ab 100644
--- a/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/PutSyslog.java
+++ b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/PutSyslog.java
@@ -40,7 +40,6 @@ import org.apache.nifi.processor.util.put.sender.SSLSocketChannelSender;
import org.apache.nifi.processor.util.put.sender.SocketChannelSender;
import org.apache.nifi.processors.standard.syslog.SyslogParser;
import org.apache.nifi.ssl.SSLContextService;
-import org.apache.nifi.util.ObjectHolder;
import org.apache.nifi.util.StopWatch;
import javax.net.ssl.SSLContext;
@@ -55,6 +54,7 @@ import java.util.Set;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicReference;
import java.util.regex.Matcher;
import java.util.regex.Pattern;
@@ -321,7 +321,7 @@ public class PutSyslog extends AbstractSyslogProcessor {
final String port = context.getProperty(PORT).getValue();
final String host = context.getProperty(HOSTNAME).getValue();
final String transitUri = new StringBuilder().append(protocol).append("://").append(host).append(":").append(port).toString();
- final ObjectHolder<IOException> exceptionHolder = new ObjectHolder<>(null);
+ final AtomicReference<IOException> exceptionHolder = new AtomicReference<>(null);
final Charset charSet = Charset.forName(context.getProperty(CHARSET).getValue());
try {
http://git-wip-us.apache.org/repos/asf/nifi/blob/f987b216/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/QueryDatabaseTable.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/QueryDatabaseTable.java b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/QueryDatabaseTable.java
index ec81446..e749696 100644
--- a/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/QueryDatabaseTable.java
+++ b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/QueryDatabaseTable.java
@@ -41,7 +41,6 @@ import org.apache.nifi.processor.exception.ProcessException;
import org.apache.nifi.processor.io.OutputStreamCallback;
import org.apache.nifi.processor.util.StandardValidators;
import org.apache.nifi.processors.standard.util.JdbcCommon;
-import org.apache.nifi.util.LongHolder;
import org.apache.nifi.util.StopWatch;
import java.io.IOException;
@@ -93,6 +92,7 @@ import static java.sql.Types.TIMESTAMP;
import static java.sql.Types.TINYINT;
import static java.sql.Types.VARBINARY;
import static java.sql.Types.VARCHAR;
+import java.util.concurrent.atomic.AtomicLong;
@EventDriven
@InputRequirement(Requirement.INPUT_FORBIDDEN)
@@ -296,7 +296,7 @@ public class QueryDatabaseTable extends AbstractSessionFactoryProcessor {
final Integer queryTimeout = context.getProperty(QUERY_TIMEOUT).asTimePeriod(TimeUnit.SECONDS).intValue();
st.setQueryTimeout(queryTimeout); // timeout in seconds
- final LongHolder nrOfRows = new LongHolder(0L);
+ final AtomicLong nrOfRows = new AtomicLong(0L);
fileToProcess = session.create();
fileToProcess = session.write(fileToProcess, new OutputStreamCallback() {
@@ -317,7 +317,7 @@ public class QueryDatabaseTable extends AbstractSessionFactoryProcessor {
if (nrOfRows.get() > 0) {
// set attribute how many rows were selected
- fileToProcess = session.putAttribute(fileToProcess, RESULT_ROW_COUNT, nrOfRows.get().toString());
+ fileToProcess = session.putAttribute(fileToProcess, RESULT_ROW_COUNT, String.valueOf(nrOfRows.get()));
logger.info("{} contains {} Avro records; transferring to 'success'",
new Object[]{fileToProcess, nrOfRows.get()});
http://git-wip-us.apache.org/repos/asf/nifi/blob/f987b216/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/RouteOnContent.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/RouteOnContent.java b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/RouteOnContent.java
index 1bf235b..64877f2 100644
--- a/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/RouteOnContent.java
+++ b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/RouteOnContent.java
@@ -26,6 +26,7 @@ import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
+import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;
import java.util.regex.Pattern;
@@ -51,7 +52,6 @@ import org.apache.nifi.processor.Relationship;
import org.apache.nifi.processor.io.InputStreamCallback;
import org.apache.nifi.processor.util.StandardValidators;
import org.apache.nifi.stream.io.StreamUtils;
-import org.apache.nifi.util.IntegerHolder;
@EventDriven
@SideEffectFree
@@ -182,7 +182,7 @@ public class RouteOnContent extends AbstractProcessor {
final Set<Relationship> destinations = new HashSet<>();
flowFileDestinationMap.put(flowFile, destinations);
- final IntegerHolder bufferedByteCount = new IntegerHolder(0);
+ final AtomicInteger bufferedByteCount = new AtomicInteger(0);
session.read(flowFile, new InputStreamCallback() {
@Override
public void process(final InputStream in) throws IOException {
http://git-wip-us.apache.org/repos/asf/nifi/blob/f987b216/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/ScanContent.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/ScanContent.java b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/ScanContent.java
index 59d1cbd..1f635f9 100644
--- a/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/ScanContent.java
+++ b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/ScanContent.java
@@ -54,7 +54,6 @@ import org.apache.nifi.processor.exception.ProcessException;
import org.apache.nifi.processor.io.InputStreamCallback;
import org.apache.nifi.processor.util.StandardValidators;
import org.apache.nifi.stream.io.BufferedInputStream;
-import org.apache.nifi.util.ObjectHolder;
import org.apache.nifi.util.file.monitor.LastModifiedMonitor;
import org.apache.nifi.util.file.monitor.SynchronousFileWatcher;
import org.apache.nifi.util.search.Search;
@@ -219,7 +218,7 @@ public class ScanContent extends AbstractProcessor {
}
final Search<byte[]> finalSearch = search;
- final ObjectHolder<SearchTerm<byte[]>> termRef = new ObjectHolder<>(null);
+ final AtomicReference<SearchTerm<byte[]>> termRef = new AtomicReference<>(null);
termRef.set(null);
session.read(flowFile, new InputStreamCallback() {
http://git-wip-us.apache.org/repos/asf/nifi/blob/f987b216/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/SplitText.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/SplitText.java b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/SplitText.java
index 0f1abfd..81c06f8 100644
--- a/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/SplitText.java
+++ b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/SplitText.java
@@ -32,6 +32,9 @@ import java.util.Map;
import java.util.Set;
import java.util.UUID;
import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicLong;
+import java.util.concurrent.atomic.AtomicReference;
import org.apache.nifi.annotation.behavior.EventDriven;
import org.apache.nifi.annotation.behavior.InputRequirement;
@@ -63,9 +66,6 @@ import org.apache.nifi.stream.io.BufferedOutputStream;
import org.apache.nifi.stream.io.ByteArrayOutputStream;
import org.apache.nifi.stream.io.ByteCountingInputStream;
import org.apache.nifi.stream.io.ByteCountingOutputStream;
-import org.apache.nifi.util.IntegerHolder;
-import org.apache.nifi.util.LongHolder;
-import org.apache.nifi.util.ObjectHolder;
@EventDriven
@SideEffectFree
@@ -412,7 +412,7 @@ public class SplitText extends AbstractProcessor {
final String headerMarker = context.getProperty(HEADER_MARKER).getValue();
final boolean includeLineDelimiter = !context.getProperty(REMOVE_TRAILING_NEWLINES).asBoolean();
- final ObjectHolder<String> errorMessage = new ObjectHolder<>(null);
+ final AtomicReference<String> errorMessage = new AtomicReference<>(null);
final ArrayList<SplitInfo> splitInfos = new ArrayList<>();
final long startNanos = System.nanoTime();
@@ -479,8 +479,8 @@ public class SplitText extends AbstractProcessor {
if (headerInfoLineCount > 0) {
// if we have header lines, create a new FlowFile, copy the header lines to that file,
// and then start copying lines
- final IntegerHolder linesCopied = new IntegerHolder(0);
- final LongHolder bytesCopied = new LongHolder(0L);
+ final AtomicInteger linesCopied = new AtomicInteger(0);
+ final AtomicLong bytesCopied = new AtomicLong(0L);
FlowFile splitFile = session.create(flowFile);
try {
splitFile = session.write(splitFile, new OutputStreamCallback() {
http://git-wip-us.apache.org/repos/asf/nifi/blob/f987b216/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/SplitXml.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/SplitXml.java b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/SplitXml.java
index 055ffd8..4764ea8 100644
--- a/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/SplitXml.java
+++ b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/SplitXml.java
@@ -27,6 +27,7 @@ import java.util.Map;
import java.util.Map.Entry;
import java.util.Set;
import java.util.TreeMap;
+import java.util.concurrent.atomic.AtomicBoolean;
import javax.xml.parsers.ParserConfigurationException;
import javax.xml.parsers.SAXParser;
@@ -53,7 +54,6 @@ import org.apache.nifi.processor.io.OutputStreamCallback;
import org.apache.nifi.processor.util.StandardValidators;
import org.apache.nifi.processors.standard.util.XmlElementNotifier;
import org.apache.nifi.stream.io.BufferedInputStream;
-import org.apache.nifi.util.BooleanHolder;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.xml.sax.Attributes;
@@ -160,7 +160,7 @@ public class SplitXml extends AbstractProcessor {
}
}, depth);
- final BooleanHolder failed = new BooleanHolder(false);
+ final AtomicBoolean failed = new AtomicBoolean(false);
session.read(original, new InputStreamCallback() {
@Override
public void process(final InputStream rawIn) throws IOException {
http://git-wip-us.apache.org/repos/asf/nifi/blob/f987b216/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/TailFile.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/TailFile.java b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/TailFile.java
index b3ece4b..1ebf736 100644
--- a/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/TailFile.java
+++ b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/TailFile.java
@@ -39,6 +39,7 @@ import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicLong;
import java.util.zip.CRC32;
import java.util.zip.CheckedInputStream;
import java.util.zip.Checksum;
@@ -68,7 +69,6 @@ import org.apache.nifi.processor.util.StandardValidators;
import org.apache.nifi.stream.io.ByteArrayOutputStream;
import org.apache.nifi.stream.io.NullOutputStream;
import org.apache.nifi.stream.io.StreamUtils;
-import org.apache.nifi.util.LongHolder;
// note: it is important that this Processor is not marked as @SupportsBatching because the session commits must complete before persisting state locally; otherwise, data loss may occur
@TriggerSerially
@@ -463,7 +463,7 @@ public class TailFile extends AbstractProcessor {
FlowFile flowFile = session.create();
final FileChannel fileReader = reader;
- final LongHolder positionHolder = new LongHolder(position);
+ final AtomicLong positionHolder = new AtomicLong(position);
flowFile = session.write(flowFile, new OutputStreamCallback() {
@Override
public void process(final OutputStream rawOut) throws IOException {