This is an automated email from the ASF dual-hosted git repository.
exceptionfactory pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/nifi.git
The following commit(s) were added to refs/heads/main by this push:
new 1d119f89b7d NIFI-16280 Pass component and Connector context through
repository APIs (#11613)
1d119f89b7d is described below
commit 1d119f89b7d7253c6306a8d235a35efd86470cb6
Author: Mark Payne <[email protected]>
AuthorDate: Thu Sep 3 13:15:50 2026 -0400
NIFI-16280 Pass component and Connector context through repository APIs
(#11613)
Session-driven content creates and FlowFile updates now carry component,
Connector, and loss-tolerance context. The Content Repository reports its
own appendable-claim size. Implements NIP-40.
Signed-off-by: David Handermann <[email protected]>
---
.../src/main/asciidoc/administration-guide.adoc | 4 +-
.../repository/ContentClaimCreationContext.java | 27 ++---
.../controller/repository/ContentRepository.java | 20 ++++
.../controller/repository/FlowFileRepository.java | 8 ++
.../repository/FlowFileUpdateContext.java | 31 +++---
.../nifi/controller/repository/LossTolerance.java | 32 +++---
.../StandardContentClaimCreationContext.java | 47 +++++++++
.../repository/StandardFlowFileUpdateContext.java | 31 +++---
.../repository/io/ContentClaimInputStream.java | 5 +-
.../repository/io/ContentClaimOutputStream.java | 0
.../repository/metrics/PerformanceTracker.java | 0
.../metrics/PerformanceTrackingInputStream.java | 0
.../repository/AbstractRepositoryContext.java | 17 ++++
.../controller/repository/RepositoryContext.java | 4 +
.../repository/StandardProcessSession.java | 16 +--
.../apache/nifi/groups/StandardProcessGroup.java | 28 +++---
.../nifi/groups/StandardProcessGroupTest.java | 52 +++++++++-
.../java/org/apache/nifi/groups/ProcessGroup.java | 29 +++---
.../org/apache/nifi/controller/FlowController.java | 14 +--
.../repository/FileSystemRepository.java | 5 +
.../repository/NonPurgeableContentRepository.java | 10 ++
.../repository/StandardRepositoryContext.java | 8 +-
.../claim/StandardContentClaimWriteCache.java | 11 ++-
.../scheduling/RepositoryContextFactory.java | 8 +-
.../NonPurgeableContentRepositoryTest.java | 60 ++++++++++++
.../repository/StandardProcessSessionIT.java | 109 ++++++++++++++++++++-
.../claim/TestStandardContentClaimWriteCache.java | 100 ++++++++++++++-----
.../StatelessContentClaimWriteCache.java | 7 +-
.../repository/StatelessRepositoryContext.java | 2 +-
29 files changed, 531 insertions(+), 154 deletions(-)
diff --git a/nifi-docs/src/main/asciidoc/administration-guide.adoc
b/nifi-docs/src/main/asciidoc/administration-guide.adoc
index df67ba22a1e..5836a0d8b94 100644
--- a/nifi-docs/src/main/asciidoc/administration-guide.adoc
+++ b/nifi-docs/src/main/asciidoc/administration-guide.adoc
@@ -3059,7 +3059,9 @@ FlowFile Repository, if also on that disk, could become
corrupt. To avoid this s
individual FlowFile as a separate file in the content repository. Doing so
would be very detrimental to performance, if each 120 byte FlowFile, for
instance, was written to its own file. Instead,
we continue writing to the same file until it reaches some threshold. This
property configures that threshold. Setting the value too small can result in
poor performance due to reading from and
writing to too many files. However, a file can only be deleted from the
content repository once there are no longer any FlowFiles pointing to it.
Therefore, setting the value too large can result
-in data remaining in the content repository for much longer, potentially
leading to the content repository running out of disk space. The default value
is `50 KB`.
+in data remaining in the content repository for much longer, potentially
leading to the content repository running out of disk space. The default value
is `50 KB`. Each Content Repository
+implementation reports the threshold that it wants the framework to use, so
this property is honored only by the File System Content Repository. An
alternate implementation may report a different
+threshold and ignore this property.
|`nifi.content.repository.directory.default`*|The location of the Content
Repository. The default value is `./content_repository`. +
+
*NOTE*: Multiple content repositories can be specified by using the
`nifi.content.repository.directory.` prefix with unique suffixes and separate
paths as values. +
diff --git
a/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/controller/repository/io/ContentClaimOutputStream.java
b/nifi-framework-api/src/main/java/org/apache/nifi/controller/repository/ContentClaimCreationContext.java
similarity index 58%
copy from
nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/controller/repository/io/ContentClaimOutputStream.java
copy to
nifi-framework-api/src/main/java/org/apache/nifi/controller/repository/ContentClaimCreationContext.java
index 3c4d7785f00..09a6356dc18 100644
---
a/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/controller/repository/io/ContentClaimOutputStream.java
+++
b/nifi-framework-api/src/main/java/org/apache/nifi/controller/repository/ContentClaimCreationContext.java
@@ -14,24 +14,27 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-
-package org.apache.nifi.controller.repository.io;
+package org.apache.nifi.controller.repository;
import org.apache.nifi.controller.repository.claim.ContentClaim;
-import java.io.IOException;
-import java.io.OutputStream;
+/**
+ * Identifies the component creating a {@link ContentClaim} and that
component's Connector and loss tolerance.
+ */
+public interface ContentClaimCreationContext {
-public abstract class ContentClaimOutputStream extends OutputStream {
+ /**
+ * @return the identifier of the component creating the claim; never
{@code null}
+ */
+ String getComponentIdentifier();
/**
- * Creates a new Content Claim that is backed by this OutputStream. This
allows the caller to
- * create a new Content Claim but ensure that they keep writing to the
same OutputStream, which can
- * significantly improve performance.
- *
- * @return a new ContentClaim
- * @throws IOException if unable to finalize the current ContentClaim or
create a new one
+ * @return the owning Connector identifier, or {@code null} if the
component is not in a Connector-managed flow
*/
- public abstract ContentClaim newContentClaim() throws IOException;
+ String getConnectorIdentifier();
+ /**
+ * @return the loss tolerance of the content being written; never {@code
null}
+ */
+ LossTolerance getLossTolerance();
}
diff --git
a/nifi-framework-api/src/main/java/org/apache/nifi/controller/repository/ContentRepository.java
b/nifi-framework-api/src/main/java/org/apache/nifi/controller/repository/ContentRepository.java
index 3327f7a36ba..cae8cb8d65f 100644
---
a/nifi-framework-api/src/main/java/org/apache/nifi/controller/repository/ContentRepository.java
+++
b/nifi-framework-api/src/main/java/org/apache/nifi/controller/repository/ContentRepository.java
@@ -33,6 +33,11 @@ import java.util.Set;
*/
public interface ContentRepository {
+ /**
+ * Default for {@link #getMaxAppendableClaimBytes()}, matching {@code
nifi.content.claim.max.appendable.size}.
+ */
+ long DEFAULT_MAX_APPENDABLE_CLAIM_BYTES = 51_200L;
+
/**
* Initializes the Content Repository, providing to it the
* ContentRepositoryContext.
@@ -94,6 +99,21 @@ public interface ContentRepository {
*/
ContentClaim create(boolean lossTolerant) throws IOException;
+ /**
+ * Creates a new content claim for the given context. The default
implementation delegates to {@link #create(boolean)}.
+ */
+ default ContentClaim create(final ContentClaimCreationContext context)
throws IOException {
+ return create(context.getLossTolerance().isLossTolerant());
+ }
+
+ /**
+ * Returns how many bytes may be written to one Content Claim before the
framework creates a new one.
+ * Implementations must return an in-memory value; I/O is not allowed. The
value may change at runtime.
+ */
+ default long getMaxAppendableClaimBytes() {
+ return DEFAULT_MAX_APPENDABLE_CLAIM_BYTES;
+ }
+
/**
* Increments the number of claimants for the given claim
*
diff --git
a/nifi-framework-api/src/main/java/org/apache/nifi/controller/repository/FlowFileRepository.java
b/nifi-framework-api/src/main/java/org/apache/nifi/controller/repository/FlowFileRepository.java
index 6521ba08b61..0520111f0ca 100644
---
a/nifi-framework-api/src/main/java/org/apache/nifi/controller/repository/FlowFileRepository.java
+++
b/nifi-framework-api/src/main/java/org/apache/nifi/controller/repository/FlowFileRepository.java
@@ -76,6 +76,14 @@ public interface FlowFileRepository extends Closeable {
*/
void updateRepository(Collection<RepositoryRecord> records) throws
IOException;
+ /**
+ * Updates the repository with the given records for the given context.
The default implementation delegates to
+ * {@link #updateRepository(Collection)}.
+ */
+ default void updateRepository(final Collection<RepositoryRecord> records,
final FlowFileUpdateContext context) throws IOException {
+ updateRepository(records);
+ }
+
/**
* Loads all FlowFiles found within the repository, establishes the content
* claims and their reference count
diff --git
a/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/controller/repository/metrics/PerformanceTracker.java
b/nifi-framework-api/src/main/java/org/apache/nifi/controller/repository/FlowFileUpdateContext.java
similarity index 60%
copy from
nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/controller/repository/metrics/PerformanceTracker.java
copy to
nifi-framework-api/src/main/java/org/apache/nifi/controller/repository/FlowFileUpdateContext.java
index 822baab41ff..f8a60526a88 100644
---
a/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/controller/repository/metrics/PerformanceTracker.java
+++
b/nifi-framework-api/src/main/java/org/apache/nifi/controller/repository/FlowFileUpdateContext.java
@@ -14,25 +14,20 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
+package org.apache.nifi.controller.repository;
-package org.apache.nifi.controller.repository.metrics;
-
-public interface PerformanceTracker {
- void beginContentRead();
-
- void endContentRead();
-
- long getContentReadNanos();
-
- void beginContentWrite();
-
- void endContentWrite();
-
- long getContentWriteNanos();
-
- void beginSessionCommit();
+/**
+ * Identifies the component whose session is writing {@link RepositoryRecord}s
to a {@link FlowFileRepository}.
+ */
+public interface FlowFileUpdateContext {
- void endSessionCommit();
+ /**
+ * @return the identifier of the component whose session is updating the
repository; never {@code null}
+ */
+ String getComponentIdentifier();
- long getSessionCommitNanos();
+ /**
+ * @return the owning Connector identifier, or {@code null} if the
component is not in a Connector-managed flow
+ */
+ String getConnectorIdentifier();
}
diff --git
a/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/controller/repository/metrics/PerformanceTracker.java
b/nifi-framework-api/src/main/java/org/apache/nifi/controller/repository/LossTolerance.java
similarity index 58%
copy from
nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/controller/repository/metrics/PerformanceTracker.java
copy to
nifi-framework-api/src/main/java/org/apache/nifi/controller/repository/LossTolerance.java
index 822baab41ff..7e1104d85a1 100644
---
a/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/controller/repository/metrics/PerformanceTracker.java
+++
b/nifi-framework-api/src/main/java/org/apache/nifi/controller/repository/LossTolerance.java
@@ -14,25 +14,23 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
+package org.apache.nifi.controller.repository;
-package org.apache.nifi.controller.repository.metrics;
-
-public interface PerformanceTracker {
- void beginContentRead();
-
- void endContentRead();
-
- long getContentReadNanos();
-
- void beginContentWrite();
-
- void endContentWrite();
-
- long getContentWriteNanos();
+/**
+ * Indicates whether data may be lost without significant consequence. A
repository is free to trade durability for
+ * performance when handling data that is marked as {@link #LOSS_TOLERANT},
for example by using more volatile storage.
+ */
+public enum LossTolerance {
+ LOSS_TOLERANT(true),
+ LOSS_INTOLERANT(false);
- void beginSessionCommit();
+ private final boolean lossTolerant;
- void endSessionCommit();
+ LossTolerance(final boolean lossTolerant) {
+ this.lossTolerant = lossTolerant;
+ }
- long getSessionCommitNanos();
+ public boolean isLossTolerant() {
+ return lossTolerant;
+ }
}
diff --git
a/nifi-framework-api/src/main/java/org/apache/nifi/controller/repository/StandardContentClaimCreationContext.java
b/nifi-framework-api/src/main/java/org/apache/nifi/controller/repository/StandardContentClaimCreationContext.java
new file mode 100644
index 00000000000..63c21cdc939
--- /dev/null
+++
b/nifi-framework-api/src/main/java/org/apache/nifi/controller/repository/StandardContentClaimCreationContext.java
@@ -0,0 +1,47 @@
+/*
+ * 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.controller.repository;
+
+import java.util.Objects;
+
+public class StandardContentClaimCreationContext implements
ContentClaimCreationContext {
+
+ private final String componentIdentifier;
+ private final String connectorIdentifier;
+ private final LossTolerance lossTolerance;
+
+ public StandardContentClaimCreationContext(final String
componentIdentifier, final String connectorIdentifier, final LossTolerance
lossTolerance) {
+ this.componentIdentifier = Objects.requireNonNull(componentIdentifier,
"Component Identifier is required");
+ this.connectorIdentifier = connectorIdentifier;
+ this.lossTolerance = Objects.requireNonNull(lossTolerance, "Loss
Tolerance is required");
+ }
+
+ @Override
+ public String getComponentIdentifier() {
+ return componentIdentifier;
+ }
+
+ @Override
+ public String getConnectorIdentifier() {
+ return connectorIdentifier;
+ }
+
+ @Override
+ public LossTolerance getLossTolerance() {
+ return lossTolerance;
+ }
+}
diff --git
a/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/controller/repository/io/ContentClaimOutputStream.java
b/nifi-framework-api/src/main/java/org/apache/nifi/controller/repository/StandardFlowFileUpdateContext.java
similarity index 52%
copy from
nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/controller/repository/io/ContentClaimOutputStream.java
copy to
nifi-framework-api/src/main/java/org/apache/nifi/controller/repository/StandardFlowFileUpdateContext.java
index 3c4d7785f00..f6787ef6ab0 100644
---
a/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/controller/repository/io/ContentClaimOutputStream.java
+++
b/nifi-framework-api/src/main/java/org/apache/nifi/controller/repository/StandardFlowFileUpdateContext.java
@@ -14,24 +14,27 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
+package org.apache.nifi.controller.repository;
-package org.apache.nifi.controller.repository.io;
+import java.util.Objects;
-import org.apache.nifi.controller.repository.claim.ContentClaim;
+public class StandardFlowFileUpdateContext implements FlowFileUpdateContext {
-import java.io.IOException;
-import java.io.OutputStream;
+ private final String componentIdentifier;
+ private final String connectorIdentifier;
-public abstract class ContentClaimOutputStream extends OutputStream {
+ public StandardFlowFileUpdateContext(final String componentIdentifier,
final String connectorIdentifier) {
+ this.componentIdentifier = Objects.requireNonNull(componentIdentifier,
"Component Identifier is required");
+ this.connectorIdentifier = connectorIdentifier;
+ }
- /**
- * Creates a new Content Claim that is backed by this OutputStream. This
allows the caller to
- * create a new Content Claim but ensure that they keep writing to the
same OutputStream, which can
- * significantly improve performance.
- *
- * @return a new ContentClaim
- * @throws IOException if unable to finalize the current ContentClaim or
create a new one
- */
- public abstract ContentClaim newContentClaim() throws IOException;
+ @Override
+ public String getComponentIdentifier() {
+ return componentIdentifier;
+ }
+ @Override
+ public String getConnectorIdentifier() {
+ return connectorIdentifier;
+ }
}
diff --git
a/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/controller/repository/io/ContentClaimInputStream.java
b/nifi-framework-api/src/main/java/org/apache/nifi/controller/repository/io/ContentClaimInputStream.java
similarity index 97%
rename from
nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/controller/repository/io/ContentClaimInputStream.java
rename to
nifi-framework-api/src/main/java/org/apache/nifi/controller/repository/io/ContentClaimInputStream.java
index bfb9f5d651f..3c129890b9d 100644
---
a/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/controller/repository/io/ContentClaimInputStream.java
+++
b/nifi-framework-api/src/main/java/org/apache/nifi/controller/repository/io/ContentClaimInputStream.java
@@ -20,7 +20,6 @@ import
org.apache.nifi.controller.repository.ContentRepository;
import org.apache.nifi.controller.repository.claim.ContentClaim;
import org.apache.nifi.controller.repository.metrics.PerformanceTracker;
import
org.apache.nifi.controller.repository.metrics.PerformanceTrackingInputStream;
-import org.apache.nifi.stream.io.StreamUtils;
import java.io.BufferedInputStream;
import java.io.IOException;
@@ -215,7 +214,7 @@ public class ContentClaimInputStream extends InputStream {
performanceTracker.beginContentRead();
try {
- StreamUtils.skip(delegate, markOffset - claimOffset);
+ delegate.skipNBytes(markOffset - claimOffset);
} finally {
performanceTracker.endContentRead();
}
@@ -243,7 +242,7 @@ public class ContentClaimInputStream extends InputStream {
performanceTracker.beginContentRead();
try {
delegate = new
PerformanceTrackingInputStream(contentRepository.read(contentClaim),
performanceTracker);
- StreamUtils.skip(delegate, claimOffset);
+ delegate.skipNBytes(claimOffset);
currentOffset = claimOffset;
} finally {
performanceTracker.endContentRead();
diff --git
a/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/controller/repository/io/ContentClaimOutputStream.java
b/nifi-framework-api/src/main/java/org/apache/nifi/controller/repository/io/ContentClaimOutputStream.java
similarity index 100%
rename from
nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/controller/repository/io/ContentClaimOutputStream.java
rename to
nifi-framework-api/src/main/java/org/apache/nifi/controller/repository/io/ContentClaimOutputStream.java
diff --git
a/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/controller/repository/metrics/PerformanceTracker.java
b/nifi-framework-api/src/main/java/org/apache/nifi/controller/repository/metrics/PerformanceTracker.java
similarity index 100%
rename from
nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/controller/repository/metrics/PerformanceTracker.java
rename to
nifi-framework-api/src/main/java/org/apache/nifi/controller/repository/metrics/PerformanceTracker.java
diff --git
a/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/controller/repository/metrics/PerformanceTrackingInputStream.java
b/nifi-framework-api/src/main/java/org/apache/nifi/controller/repository/metrics/PerformanceTrackingInputStream.java
similarity index 100%
rename from
nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/controller/repository/metrics/PerformanceTrackingInputStream.java
rename to
nifi-framework-api/src/main/java/org/apache/nifi/controller/repository/metrics/PerformanceTrackingInputStream.java
diff --git
a/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/controller/repository/AbstractRepositoryContext.java
b/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/controller/repository/AbstractRepositoryContext.java
index f92390352ae..8430a77f957 100644
---
a/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/controller/repository/AbstractRepositoryContext.java
+++
b/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/controller/repository/AbstractRepositoryContext.java
@@ -56,6 +56,8 @@ public abstract class AbstractRepositoryContext implements
RepositoryContext {
private final AtomicLong connectionIndex;
private final StateManager stateManager;
private final ComponentMetricContext componentMetricContext;
+ private final ContentClaimCreationContext contentClaimCreationContext;
+ private final FlowFileUpdateContext flowFileUpdateContext;
private final String componentNameCounterContext;
private final String componentTypeCounterContext;
@@ -89,6 +91,11 @@ public abstract class AbstractRepositoryContext implements
RepositoryContext {
this.componentNameCounterContext = connectable.getName() + " (" +
connectable.getIdentifier() + ")";
this.componentTypeCounterContext = "All " +
connectable.getComponentType() + "'s";
+
+ final String connectorIdentifier =
connectable.getProcessGroup().findOwningConnectorIdentifier().orElse(null);
+ final LossTolerance lossTolerance = connectable.isLossTolerant() ?
LossTolerance.LOSS_TOLERANT : LossTolerance.LOSS_INTOLERANT;
+ this.contentClaimCreationContext = new
StandardContentClaimCreationContext(connectable.getIdentifier(),
connectorIdentifier, lossTolerance);
+ this.flowFileUpdateContext = new
StandardFlowFileUpdateContext(connectable.getIdentifier(), connectorIdentifier);
}
@Override
@@ -198,11 +205,21 @@ public abstract class AbstractRepositoryContext
implements RepositoryContext {
return contentRepo;
}
+ @Override
+ public ContentClaimCreationContext getContentClaimCreationContext() {
+ return contentClaimCreationContext;
+ }
+
@Override
public FlowFileRepository getFlowFileRepository() {
return flowFileRepo;
}
+ @Override
+ public FlowFileUpdateContext getFlowFileUpdateContext() {
+ return flowFileUpdateContext;
+ }
+
@Override
public FlowFileEventRepository getFlowFileEventRepository() {
return flowFileEventRepo;
diff --git
a/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/controller/repository/RepositoryContext.java
b/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/controller/repository/RepositoryContext.java
index a26a39bfff9..6d5ce99b092 100644
---
a/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/controller/repository/RepositoryContext.java
+++
b/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/controller/repository/RepositoryContext.java
@@ -47,8 +47,12 @@ public interface RepositoryContext {
ContentRepository getContentRepository();
+ ContentClaimCreationContext getContentClaimCreationContext();
+
FlowFileRepository getFlowFileRepository();
+ FlowFileUpdateContext getFlowFileUpdateContext();
+
FlowFileEventRepository getFlowFileEventRepository();
ProvenanceEventRepository getProvenanceRepository();
diff --git
a/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/controller/repository/StandardProcessSession.java
b/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/controller/repository/StandardProcessSession.java
index 1eb60299dd0..0349ccd79cc 100644
---
a/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/controller/repository/StandardProcessSession.java
+++
b/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/controller/repository/StandardProcessSession.java
@@ -616,7 +616,7 @@ public class StandardProcessSession implements
ProcessSession, ProvenanceEventEn
try {
final Collection<StandardRepositoryRecord> repoRecords =
checkpoint.records.values();
if (!repoRecords.isEmpty()) {
-
context.getFlowFileRepository().updateRepository((Collection) repoRecords);
+
context.getFlowFileRepository().updateRepository((Collection) repoRecords,
context.getFlowFileUpdateContext());
context.getConnectable().getFlowFileActivity().updateLatestActivityTime();
}
} catch (final IOException ioe) {
@@ -1387,7 +1387,7 @@ public class StandardProcessSession implements
ProcessSession, ProvenanceEventEn
if (!abortedRecords.isEmpty()) {
try {
-
context.getFlowFileRepository().updateRepository(abortedRecords);
+
context.getFlowFileRepository().updateRepository(abortedRecords,
context.getFlowFileUpdateContext());
} catch (final IOException ioe) {
LOG.error("Unable to update FlowFile repository for aborted
records", ioe);
}
@@ -1401,7 +1401,7 @@ public class StandardProcessSession implements
ProcessSession, ProvenanceEventEn
if (!transientClaims.isEmpty()) {
final RepositoryRecord repoRecord = new
TransientClaimRepositoryRecord(transientClaims);
try {
-
context.getFlowFileRepository().updateRepository(Collections.singletonList(repoRecord));
+
context.getFlowFileRepository().updateRepository(Collections.singletonList(repoRecord),
context.getFlowFileUpdateContext());
} catch (final IOException ioe) {
LOG.error("Unable to update FlowFile repository to cleanup
transient claims", ioe);
}
@@ -2752,7 +2752,7 @@ public class StandardProcessSession implements
ProcessSession, ProvenanceEventEn
};
context.getProvenanceRepository().registerEvents(iterable);
- context.getFlowFileRepository().updateRepository(expiredRecords);
+ context.getFlowFileRepository().updateRepository(expiredRecords,
context.getFlowFileUpdateContext());
} catch (final IOException e) {
LOG.error("Failed to update FlowFile Repository to record expired
records", e);
}
@@ -3055,7 +3055,7 @@ public class StandardProcessSession implements
ProcessSession, ProvenanceEventEn
final ContentRepository contentRepo = context.getContentRepository();
final ContentClaim newClaim;
try {
- newClaim =
contentRepo.create(context.getConnectable().isLossTolerant());
+ newClaim =
contentRepo.create(context.getContentClaimCreationContext());
claimLog.debug("Creating ContentClaim {} for 'merge' for {}",
newClaim, destinationRecord.getCurrent());
} catch (final IOException e) {
throw new FlowFileAccessException("Unable to create ContentClaim
due to " + e, e);
@@ -3369,7 +3369,7 @@ public class StandardProcessSession implements
ProcessSession, ProvenanceEventEn
claimCache.flush(oldClaim);
try (final InputStream oldClaimIn = read(source)) {
- newClaim =
context.getContentRepository().create(context.getConnectable().isLossTolerant());
+ newClaim =
context.getContentRepository().create(context.getContentClaimCreationContext());
claimLog.debug("Creating ContentClaim {} for 'append' for
{}", newClaim, source);
final OutputStream rawOutStream =
context.getContentRepository().write(newClaim);
@@ -3679,7 +3679,7 @@ public class StandardProcessSession implements
ProcessSession, ProvenanceEventEn
final long claimOffset;
try {
- newClaim =
context.getContentRepository().create(context.getConnectable().isLossTolerant());
+ newClaim =
context.getContentRepository().create(context.getContentClaimCreationContext());
claimLog.debug("Creating ContentClaim {} for 'importFrom' for {}",
newClaim, destination);
} catch (final IOException e) {
throw new FlowFileAccessException("Unable to create ContentClaim
due to " + e, e);
@@ -3741,7 +3741,7 @@ public class StandardProcessSession implements
ProcessSession, ProvenanceEventEn
final long newSize;
try {
try {
- newClaim =
context.getContentRepository().create(context.getConnectable().isLossTolerant());
+ newClaim =
context.getContentRepository().create(context.getContentClaimCreationContext());
claimLog.debug("Creating ContentClaim {} for 'importFrom' for
{}", newClaim, destination);
newSize =
context.getContentRepository().importFrom(createTaskTerminationStream(source),
newClaim);
diff --git
a/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/groups/StandardProcessGroup.java
b/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/groups/StandardProcessGroup.java
index dec748330ef..f2b41766f27 100644
---
a/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/groups/StandardProcessGroup.java
+++
b/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/groups/StandardProcessGroup.java
@@ -178,6 +178,7 @@ public final class StandardProcessGroup implements
ProcessGroup {
private final AtomicReference<StandardVersionControlInformation>
versionControlInfo = new AtomicReference<>();
private static final SecureRandom randomGenerator = new SecureRandom();
private final String connectorId;
+ private volatile Optional<String> owningConnectorIdentifier;
private final ProcessScheduler scheduler;
private final ControllerServiceProvider controllerServiceProvider;
@@ -309,6 +310,8 @@ public final class StandardProcessGroup implements
ProcessGroup {
@Override
public void setParent(final ProcessGroup newParent) {
parent.set(newParent);
+ // Forget a value resolved before this group was attached to its
parent.
+ owningConnectorIdentifier = null;
// Inherit connector-supplied MDC attributes from the parent so
descendants of a connector's managed
// flow carry the same connector metadata (attributing their logs and
status metrics to the connector).
// Runs on every re-parent (including initial attach), so PGs added
later inherit automatically.
@@ -364,19 +367,22 @@ public final class StandardProcessGroup implements
ProcessGroup {
}
@Override
- public Optional<ConnectorNode> findOwningConnector() {
- ProcessGroup group = this;
- while (group != null) {
- final Optional<String> owningConnectorId =
group.getConnectorIdentifier();
- if (owningConnectorId.isPresent()) {
- final ConnectorNode connectorNode =
flowManager.getConnector(owningConnectorId.get());
- return Optional.ofNullable(connectorNode);
- }
-
- group = group.getParent();
+ public Optional<String> findOwningConnectorIdentifier() {
+ Optional<String> identifier = owningConnectorIdentifier;
+ // Check-then-modify, but safe. The parent cannot change from one
Connector to another; during initialization
+ // it may go from no Connector to some Connector. The only incorrect
cached value is null (unresolved), and
+ // the next call looks it up again.
+ if (identifier == null) {
+ identifier = ProcessGroup.super.findOwningConnectorIdentifier();
+ owningConnectorIdentifier = identifier;
}
- return Optional.empty();
+ return identifier;
+ }
+
+ @Override
+ public Optional<ConnectorNode> findOwningConnector() {
+ return findOwningConnectorIdentifier().map(flowManager::getConnector);
}
@Override
diff --git
a/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/test/java/org/apache/nifi/groups/StandardProcessGroupTest.java
b/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/test/java/org/apache/nifi/groups/StandardProcessGroupTest.java
index 73d41805d43..958ce530bef 100644
---
a/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/test/java/org/apache/nifi/groups/StandardProcessGroupTest.java
+++
b/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/test/java/org/apache/nifi/groups/StandardProcessGroupTest.java
@@ -62,6 +62,7 @@ import static org.mockito.ArgumentMatchers.anyString;
import static org.mockito.ArgumentMatchers.eq;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
@@ -403,6 +404,51 @@ class StandardProcessGroupTest {
assertEquals(expected, leaf.getLoggingAttributes());
}
+ @Test
+ void testFindOwningConnectorIdentifierWalksParentHierarchy() {
+ final StandardProcessGroup connectorGroup =
createStandardProcessGroup("connector-group", "connector-1");
+ final StandardProcessGroup child = createStandardProcessGroup("child");
+ final StandardProcessGroup grandchild =
createStandardProcessGroup("grandchild");
+
+ child.setParent(connectorGroup);
+ grandchild.setParent(child);
+
+ assertEquals(Optional.of("connector-1"),
connectorGroup.findOwningConnectorIdentifier());
+ assertEquals(Optional.of("connector-1"),
child.findOwningConnectorIdentifier());
+ assertEquals(Optional.of("connector-1"),
grandchild.findOwningConnectorIdentifier());
+ assertEquals(Optional.empty(),
processGroup.findOwningConnectorIdentifier());
+ }
+
+ @Test
+ void testFindOwningConnectorIdentifierResolvesHierarchyOnlyOnce() {
+
when(parentProcessGroup.getConnectorIdentifier()).thenReturn(Optional.of("connector-1"));
+ processGroup.setParent(parentProcessGroup);
+
+ assertEquals(Optional.of("connector-1"),
processGroup.findOwningConnectorIdentifier());
+ assertEquals(Optional.of("connector-1"),
processGroup.findOwningConnectorIdentifier());
+ assertEquals(Optional.of("connector-1"),
processGroup.findOwningConnectorIdentifier());
+ verify(parentProcessGroup, times(1)).getConnectorIdentifier();
+
+ final ProcessGroup unmanagedParent = mock(ProcessGroup.class);
+ final StandardProcessGroup unmanagedGroup =
createStandardProcessGroup("unmanaged-group");
+ unmanagedGroup.setParent(unmanagedParent);
+
+ assertEquals(Optional.empty(),
unmanagedGroup.findOwningConnectorIdentifier());
+ assertEquals(Optional.empty(),
unmanagedGroup.findOwningConnectorIdentifier());
+
+ verify(unmanagedParent, times(1)).getConnectorIdentifier();
+ }
+
+ @Test
+ void testFindOwningConnectorIdentifierResolvedAgainAfterParentAssigned() {
+ assertEquals(Optional.empty(),
processGroup.findOwningConnectorIdentifier());
+
+
when(parentProcessGroup.getConnectorIdentifier()).thenReturn(Optional.of("connector-1"));
+ processGroup.setParent(parentProcessGroup);
+
+ assertEquals(Optional.of("connector-1"),
processGroup.findOwningConnectorIdentifier());
+ }
+
@Test
void testDropAllFlowFilesCompletionWaitsForEveryConnection() {
when(flowManager.getFlowAnalyzer()).thenReturn(Optional.empty());
@@ -431,6 +477,10 @@ class StandardProcessGroupTest {
}
private StandardProcessGroup createStandardProcessGroup(final String id) {
+ return createStandardProcessGroup(id, null);
+ }
+
+ private StandardProcessGroup createStandardProcessGroup(final String id,
final String connectorId) {
return new StandardProcessGroup(
id,
controllerServiceProvider,
@@ -445,7 +495,7 @@ class StandardProcessGroupTest {
properties,
statelessGroupNodeFactory,
assetManager,
- null
+ connectorId
);
}
diff --git
a/nifi-framework-bundle/nifi-framework/nifi-framework-core-api/src/main/java/org/apache/nifi/groups/ProcessGroup.java
b/nifi-framework-bundle/nifi-framework/nifi-framework-core-api/src/main/java/org/apache/nifi/groups/ProcessGroup.java
index 67ae8990138..c96f40c2253 100644
---
a/nifi-framework-bundle/nifi-framework/nifi-framework-core-api/src/main/java/org/apache/nifi/groups/ProcessGroup.java
+++
b/nifi-framework-bundle/nifi-framework/nifi-framework-core-api/src/main/java/org/apache/nifi/groups/ProcessGroup.java
@@ -145,22 +145,29 @@ public interface ProcessGroup extends
ComponentAuthorizable, Positionable, Versi
Optional<String> getConnectorIdentifier();
/**
- * Returns the owning Connector for this Process Group, traversing the
Process Group hierarchy until a Process Group
- * is found that is associated with a Connector. If no Process Group in
the hierarchy is associated with a Connector,
- * an empty Optional is returned. This is useful for determining whether a
component is managed by a Connector.
- *
- * <p>The default implementation returns {@link Optional#empty()}.
Implementations that can resolve a
- * {@link ConnectorNode} from a connector identifier (typically via a
FlowManager) should override this method
- * and walk the parent chain using {@link #getConnectorIdentifier()} and
{@link #getParent()} to locate the
- * owning Connector.</p>
- *
- * @return an Optional containing the owning ConnectorNode, or empty if
this Process Group and all of its ancestors are
- * not managed by a Connector
+ * @return the Connector that owns this Process Group or an ancestor, or
empty if none
*/
default Optional<ConnectorNode> findOwningConnector() {
return Optional.empty();
}
+ /**
+ * @return the identifier of the Connector that owns this Process Group or
an ancestor, or empty if none
+ */
+ default Optional<String> findOwningConnectorIdentifier() {
+ ProcessGroup group = this;
+ while (group != null) {
+ final Optional<String> connectorIdentifier =
group.getConnectorIdentifier();
+ if (connectorIdentifier.isPresent()) {
+ return connectorIdentifier;
+ }
+
+ group = group.getParent();
+ }
+
+ return Optional.empty();
+ }
+
/**
* @return the user-set comments about this ProcessGroup, or
* <code>null</code> if no comments have been set
diff --git
a/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/FlowController.java
b/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/FlowController.java
index 8044d9abb66..fcf76a0ac9a 100644
---
a/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/FlowController.java
+++
b/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/FlowController.java
@@ -189,7 +189,6 @@ import org.apache.nifi.nar.PythonBundle;
import org.apache.nifi.parameter.ParameterContextManager;
import org.apache.nifi.parameter.ParameterProvider;
import org.apache.nifi.parameter.StandardParameterContextManager;
-import org.apache.nifi.processor.DataUnit;
import org.apache.nifi.processor.Processor;
import org.apache.nifi.processor.Relationship;
import org.apache.nifi.processor.StandardProcessContext;
@@ -601,9 +600,8 @@ public class FlowController implements
ReportingTaskProvider, FlowAnalysisRulePr
processScheduler = new
StandardProcessScheduler(timerDrivenEngineRef.get(), this,
stateManagerProvider, this.nifiProperties, lifecycleStateManager);
parameterContextManager = new StandardParameterContextManager();
- final long maxAppendableBytes = getMaxAppendableBytes();
repositoryContextFactory = new
RepositoryContextFactory(contentRepository, flowFileRepository,
flowFileEventRepository,
- counterRepositoryRef.get(), componentMetricReporter,
provenanceRepository, stateManagerProvider, maxAppendableBytes);
+ counterRepositoryRef.get(), componentMetricReporter,
provenanceRepository, stateManagerProvider);
assetManager = createAssetManager(
nifiProperties,
@@ -1352,7 +1350,6 @@ public class FlowController implements
ReportingTaskProvider, FlowAnalysisRulePr
flowFileRepository.updateMaxFlowFileIdentifier(maxIdFromSwapFiles
+ 1);
// Begin expiring FlowFiles that are old
- final long maxAppendableClaimBytes = getMaxAppendableBytes();
final RepositoryContextFactory contextFactory = new
RepositoryContextFactory(
contentRepository,
flowFileRepository,
@@ -1360,8 +1357,7 @@ public class FlowController implements
ReportingTaskProvider, FlowAnalysisRulePr
counterRepositoryRef.get(),
getComponentMetricReporter(),
provenanceRepository,
- stateManagerProvider,
- maxAppendableClaimBytes
+ stateManagerProvider
);
processScheduler.scheduleFrameworkTask(new ExpireFlowFiles(this,
contextFactory), "Expire FlowFiles", 30L, 30L, TimeUnit.SECONDS);
@@ -1406,12 +1402,6 @@ public class FlowController implements
ReportingTaskProvider, FlowAnalysisRulePr
}
}
- private long getMaxAppendableBytes() {
- final String maxAppendableClaimSize =
nifiProperties.getMaxAppendableClaimSize();
- final long maxAppendableClaimBytes =
DataUnit.parseDataSize(maxAppendableClaimSize, DataUnit.B).longValue();
- return maxAppendableClaimBytes;
- }
-
private void notifyComponentsConfigurationRestored() {
for (final ProcessorNode procNode :
flowManager.getRootGroup().findAllProcessors()) {
final Processor processor = procNode.getProcessor();
diff --git
a/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/repository/FileSystemRepository.java
b/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/repository/FileSystemRepository.java
index b8fba96fff8..ddcb048da27 100644
---
a/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/repository/FileSystemRepository.java
+++
b/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/repository/FileSystemRepository.java
@@ -639,6 +639,11 @@ public class FileSystemRepository implements
ContentRepository {
}
}
+ @Override
+ public long getMaxAppendableClaimBytes() {
+ return maxAppendableClaimLength;
+ }
+
@Override
public ContentClaim create(final boolean lossTolerant) throws IOException {
ResourceClaim resourceClaim;
diff --git
a/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/repository/NonPurgeableContentRepository.java
b/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/repository/NonPurgeableContentRepository.java
index eaef52a5aca..974be1cf7d7 100644
---
a/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/repository/NonPurgeableContentRepository.java
+++
b/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/repository/NonPurgeableContentRepository.java
@@ -68,6 +68,16 @@ public class NonPurgeableContentRepository implements
ContentRepository {
return delegate.create(lossTolerant);
}
+ @Override
+ public ContentClaim create(final ContentClaimCreationContext context)
throws IOException {
+ return delegate.create(context);
+ }
+
+ @Override
+ public long getMaxAppendableClaimBytes() {
+ return delegate.getMaxAppendableClaimBytes();
+ }
+
@Override
public int incrementClaimaintCount(final ContentClaim claim) {
return delegate.incrementClaimaintCount(claim);
diff --git
a/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/repository/StandardRepositoryContext.java
b/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/repository/StandardRepositoryContext.java
index 4b569269de2..0d72a7ebee2 100644
---
a/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/repository/StandardRepositoryContext.java
+++
b/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/repository/StandardRepositoryContext.java
@@ -28,8 +28,6 @@ import java.util.concurrent.atomic.AtomicLong;
public class StandardRepositoryContext extends AbstractRepositoryContext
implements RepositoryContext {
- private final long maxAppendableClaimBytes;
-
public StandardRepositoryContext(
final Connectable connectable,
final AtomicLong connectionIndex,
@@ -39,15 +37,13 @@ public class StandardRepositoryContext extends
AbstractRepositoryContext impleme
final CounterRepository counterRepository,
final ComponentMetricReporter componentMetricReporter,
final ProvenanceEventRepository provenanceRepository,
- final StateManager stateManager,
- final long maxAppendableClaimBytes
+ final StateManager stateManager
) {
super(connectable, connectionIndex, contentRepository,
flowFileRepository, flowFileEventRepository, counterRepository,
componentMetricReporter, provenanceRepository, stateManager);
- this.maxAppendableClaimBytes = maxAppendableClaimBytes;
}
@Override
public ContentClaimWriteCache createContentClaimWriteCache(final
PerformanceTracker performanceTracker) {
- return new StandardContentClaimWriteCache(getContentRepository(),
performanceTracker, maxAppendableClaimBytes, 8192);
+ return new StandardContentClaimWriteCache(getContentRepository(),
performanceTracker, getContentClaimCreationContext(), 8192);
}
}
diff --git
a/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/repository/claim/StandardContentClaimWriteCache.java
b/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/repository/claim/StandardContentClaimWriteCache.java
index 74639395563..7203d47874f 100644
---
a/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/repository/claim/StandardContentClaimWriteCache.java
+++
b/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/repository/claim/StandardContentClaimWriteCache.java
@@ -17,6 +17,7 @@
package org.apache.nifi.controller.repository.claim;
+import org.apache.nifi.controller.repository.ContentClaimCreationContext;
import org.apache.nifi.controller.repository.ContentRepository;
import org.apache.nifi.controller.repository.io.ContentClaimOutputStream;
import org.apache.nifi.controller.repository.metrics.PerformanceTracker;
@@ -35,13 +36,13 @@ public class StandardContentClaimWriteCache implements
ContentClaimWriteCache {
private final Map<ResourceClaim, MappedOutputStream> streamMap = new
ConcurrentHashMap<>();
private final Queue<ContentClaim> queue = new LinkedList<>();
private final PerformanceTracker performanceTracker;
- private final long maxAppendableClaimBytes;
+ private final ContentClaimCreationContext creationContext;
private final int bufferSize;
- public StandardContentClaimWriteCache(final ContentRepository contentRepo,
final PerformanceTracker performanceTracker, final long
maxAppendableClaimBytes, final int bufferSize) {
+ public StandardContentClaimWriteCache(final ContentRepository contentRepo,
final PerformanceTracker performanceTracker, final ContentClaimCreationContext
creationContext, final int bufferSize) {
this.contentRepo = contentRepo;
this.performanceTracker = performanceTracker;
- this.maxAppendableClaimBytes = maxAppendableClaimBytes;
+ this.creationContext = creationContext;
this.bufferSize = bufferSize;
}
@@ -70,7 +71,7 @@ public class StandardContentClaimWriteCache implements
ContentClaimWriteCache {
}
}
- final ContentClaim claim = contentRepo.create(false);
+ final ContentClaim claim = contentRepo.create(creationContext);
registerStream(claim);
return claim;
}
@@ -152,7 +153,7 @@ public class StandardContentClaimWriteCache implements
ContentClaimWriteCache {
}
// Add the claim back to the queue if it is still writable
- if ((scc.getOffset() + scc.getLength()) <
maxAppendableClaimBytes) {
+ if ((scc.getOffset() + scc.getLength()) <
contentRepo.getMaxAppendableClaimBytes()) {
queue.offer(claim);
}
}
diff --git
a/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/scheduling/RepositoryContextFactory.java
b/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/scheduling/RepositoryContextFactory.java
index 8aabdb676e9..9ee55397a31 100644
---
a/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/scheduling/RepositoryContextFactory.java
+++
b/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/scheduling/RepositoryContextFactory.java
@@ -40,7 +40,6 @@ public class RepositoryContextFactory {
private final ComponentMetricReporter componentMetricReporter;
private final ProvenanceRepository provenanceRepo;
private final StateManagerProvider stateManagerProvider;
- private final long maxAppendableClaimBytes;
public RepositoryContextFactory(
final ContentRepository contentRepository,
@@ -49,8 +48,7 @@ public class RepositoryContextFactory {
final CounterRepository counterRepository,
final ComponentMetricReporter componentMetricReporter,
final ProvenanceRepository provenanceRepository,
- final StateManagerProvider stateManagerProvider,
- final long maxAppendableClaimBytes
+ final StateManagerProvider stateManagerProvider
) {
this.contentRepo = contentRepository;
this.flowFileRepo = flowFileRepository;
@@ -59,7 +57,6 @@ public class RepositoryContextFactory {
this.componentMetricReporter = componentMetricReporter;
this.provenanceRepo = provenanceRepository;
this.stateManagerProvider = stateManagerProvider;
- this.maxAppendableClaimBytes = maxAppendableClaimBytes;
}
public RepositoryContext newProcessContext(final Connectable connectable,
final AtomicLong connectionIndex) {
@@ -76,8 +73,7 @@ public class RepositoryContextFactory {
counterRepo,
componentMetricReporter,
provenanceRepo,
- stateManager,
- maxAppendableClaimBytes
+ stateManager
);
}
diff --git
a/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/repository/NonPurgeableContentRepositoryTest.java
b/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/repository/NonPurgeableContentRepositoryTest.java
new file mode 100644
index 00000000000..2a33ccc972a
--- /dev/null
+++
b/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/repository/NonPurgeableContentRepositoryTest.java
@@ -0,0 +1,60 @@
+/*
+ * 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.controller.repository;
+
+import org.apache.nifi.controller.repository.claim.ContentClaim;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.mockito.Mock;
+import org.mockito.junit.jupiter.MockitoExtension;
+
+import java.io.IOException;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertSame;
+import static org.mockito.Mockito.when;
+
+@ExtendWith(MockitoExtension.class)
+class NonPurgeableContentRepositoryTest {
+ private static final long DELEGATE_MAX_APPENDABLE_CLAIM_BYTES = 1_048_576L;
+
+ private static final ContentClaimCreationContext CREATION_CONTEXT = new
StandardContentClaimCreationContext("component-1", "connector-1",
LossTolerance.LOSS_TOLERANT);
+
+ @Mock
+ private ContentRepository delegate;
+
+ @Mock
+ private ContentClaim contentClaim;
+
+ @Test
+ void testMaxAppendableClaimBytesTakenFromDelegate() {
+
when(delegate.getMaxAppendableClaimBytes()).thenReturn(DELEGATE_MAX_APPENDABLE_CLAIM_BYTES);
+
+ final ContentRepository repository = new
NonPurgeableContentRepository(delegate);
+
+ assertEquals(DELEGATE_MAX_APPENDABLE_CLAIM_BYTES,
repository.getMaxAppendableClaimBytes());
+ }
+
+ @Test
+ void testCreationContextForwardedToDelegate() throws IOException {
+ when(delegate.create(CREATION_CONTEXT)).thenReturn(contentClaim);
+
+ final ContentRepository repository = new
NonPurgeableContentRepository(delegate);
+
+ assertSame(contentClaim, repository.create(CREATION_CONTEXT));
+ }
+}
diff --git
a/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/repository/StandardProcessSessionIT.java
b/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/repository/StandardProcessSessionIT.java
index 582cd68d11f..6baab772b82 100644
---
a/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/repository/StandardProcessSessionIT.java
+++
b/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/repository/StandardProcessSessionIT.java
@@ -92,6 +92,7 @@ import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Objects;
+import java.util.Optional;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
@@ -128,6 +129,10 @@ import static org.mockito.Mockito.when;
public class StandardProcessSessionIT {
private static final Relationship FAKE_RELATIONSHIP = new
Relationship.Builder().name("FAKE").build();
+ private static final String CONNECTABLE_ID = "connectable-1";
+
+ private static final String CONNECTOR_ID = "connector-1";
+
private StandardProcessSession session;
private MockContentRepository contentRepo;
private FlowFileQueue flowFileQueue;
@@ -193,12 +198,13 @@ public class StandardProcessSessionIT {
final ProcessGroup procGroup = mock(ProcessGroup.class);
when(procGroup.getIdentifier()).thenReturn("proc-group-identifier-1");
when(procGroup.getLoggingAttributes()).thenReturn(Map.of());
+
when(procGroup.findOwningConnectorIdentifier()).thenReturn(Optional.of(CONNECTOR_ID));
connectable = mock(Connectable.class);
when(connectable.hasIncomingConnection()).thenReturn(true);
when(connectable.getIncomingConnections()).thenReturn(connList);
when(connectable.getProcessGroup()).thenReturn(procGroup);
- when(connectable.getIdentifier()).thenReturn("connectable-1");
+ when(connectable.getIdentifier()).thenReturn(CONNECTABLE_ID);
when(connectable.getConnectableType()).thenReturn(ConnectableType.INPUT_PORT);
when(connectable.getComponentType()).thenReturn("Unit Test Component");
when(connectable.getBackoffMechanism()).thenReturn(BackoffMechanism.PENALIZE_FLOWFILE);
@@ -227,7 +233,7 @@ public class StandardProcessSessionIT {
stateManager.setIgnoreAnnotations(true);
context = new StandardRepositoryContext(connectable, new
AtomicLong(0L), contentRepo, flowFileRepo, flowFileEventRepository,
- counterRepository, componentMetricReporter, provenanceRepo,
stateManager, 50_000L);
+ counterRepository, componentMetricReporter, provenanceRepo,
stateManager);
session = new StandardProcessSession(context, () -> false, new
NopPerformanceTracker());
}
@@ -2356,6 +2362,80 @@ public class StandardProcessSessionIT {
assertEquals(5, transientClaims.size());
}
+ @Test
+ public void
testContentClaimCreationContextIdentifiesComponentAndConnector() {
+ FlowFile flowFile = session.create();
+ flowFile = session.importFrom(new
ByteArrayInputStream("imported".getBytes(StandardCharsets.UTF_8)), flowFile);
+ flowFile = session.write(flowFile, out ->
out.write("written".getBytes(StandardCharsets.UTF_8)));
+
+ session.transfer(flowFile, new
Relationship.Builder().name("success").build());
+ session.commit();
+
+ // The first claim is created directly by the import; the second is
obtained from the session's write cache.
+ final List<ContentClaimCreationContext> creationContexts =
contentRepo.getCreationContexts();
+ assertEquals(2, creationContexts.size());
+
+ for (final ContentClaimCreationContext creationContext :
creationContexts) {
+ assertEquals(CONNECTABLE_ID,
creationContext.getComponentIdentifier());
+ assertEquals(CONNECTOR_ID,
creationContext.getConnectorIdentifier());
+ assertEquals(LossTolerance.LOSS_INTOLERANT,
creationContext.getLossTolerance());
+ }
+ }
+
+ @Test
+ public void testContentClaimCreationContextLossToleranceMatchesComponent()
{
+ when(connectable.isLossTolerant()).thenReturn(true);
+
+ final StandardRepositoryContext lossTolerantContext = new
StandardRepositoryContext(connectable, new AtomicLong(0L), contentRepo,
flowFileRepo,
+ flowFileEventRepository, counterRepository,
componentMetricReporter, provenanceRepo, stateManager);
+ final StandardProcessSession lossTolerantSession = new
StandardProcessSession(lossTolerantContext, () -> false, new
NopPerformanceTracker());
+
+ try {
+ FlowFile flowFile = lossTolerantSession.create();
+ flowFile = lossTolerantSession.importFrom(new
ByteArrayInputStream("imported".getBytes(StandardCharsets.UTF_8)), flowFile);
+ lossTolerantSession.write(flowFile, out ->
out.write("written".getBytes(StandardCharsets.UTF_8)));
+ } finally {
+ lossTolerantSession.rollback();
+ }
+
+ final List<ContentClaimCreationContext> creationContexts =
contentRepo.getCreationContexts();
+ assertEquals(2, creationContexts.size());
+
+ for (final ContentClaimCreationContext creationContext :
creationContexts) {
+ assertEquals(LossTolerance.LOSS_TOLERANT,
creationContext.getLossTolerance());
+ }
+ }
+
+ @Test
+ public void
testFlowFileUpdateContextIdentifiesComponentAndConnectorOnCommit() {
+ flowFileQueue.put(new MockFlowFileRecord(1L));
+
+ final FlowFile flowFile = session.get();
+ session.transfer(flowFile, new
Relationship.Builder().name("success").build());
+ session.commit();
+
+ final List<FlowFileUpdateContext> updateContexts =
flowFileRepo.getUpdateContexts();
+ assertEquals(1, updateContexts.size());
+ assertEquals(CONNECTABLE_ID,
updateContexts.getFirst().getComponentIdentifier());
+ assertEquals(CONNECTOR_ID,
updateContexts.getFirst().getConnectorIdentifier());
+ }
+
+ @Test
+ public void
testFlowFileUpdateContextIdentifiesComponentAndConnectorOnRollback() {
+ flowFileQueue.put(new MockFlowFileRecord(1L));
+
+ FlowFile flowFile = session.get();
+ flowFile = session.write(flowFile, out ->
out.write("content".getBytes(StandardCharsets.UTF_8)));
+ assertNotNull(flowFile);
+
+ session.rollback();
+
+ final List<FlowFileUpdateContext> updateContexts =
flowFileRepo.getUpdateContexts();
+ assertEquals(1, updateContexts.size());
+ assertEquals(CONNECTABLE_ID,
updateContexts.getFirst().getComponentIdentifier());
+ assertEquals(CONNECTOR_ID,
updateContexts.getFirst().getConnectorIdentifier());
+ }
+
@Test
public void testMultipleReadCounts() throws IOException {
final ContentClaim contentClaim = contentRepo.create("Hello
there".getBytes(StandardCharsets.UTF_8));
@@ -3107,8 +3187,7 @@ public class StandardProcessSessionIT {
counterRepository,
componentMetricReporter,
provenanceRepo,
- stateManager,
- 50_000L);
+ stateManager);
return new StandardProcessSession(context, () -> false, new
NopPerformanceTracker());
}
@@ -3118,6 +3197,7 @@ public class StandardProcessSessionIT {
private boolean failOnUpdate = false;
private final AtomicLong idGenerator = new AtomicLong(0L);
private final List<RepositoryRecord> updates = new ArrayList<>();
+ private final List<FlowFileUpdateContext> updateContexts = new
ArrayList<>();
private final ContentRepository contentRepo;
public MockFlowFileRepository(final ContentRepository contentRepo) {
@@ -3164,10 +3244,20 @@ public class StandardProcessSessionIT {
}
}
+ @Override
+ public void updateRepository(final Collection<RepositoryRecord>
records, final FlowFileUpdateContext context) throws IOException {
+ updateContexts.add(context);
+ FlowFileRepository.super.updateRepository(records, context);
+ }
+
public List<RepositoryRecord> getUpdates() {
return updates;
}
+ public List<FlowFileUpdateContext> getUpdateContexts() {
+ return updateContexts;
+ }
+
@Override
public long getStorageCapacity() {
return 0;
@@ -3220,6 +3310,7 @@ public class StandardProcessSessionIT {
private final AtomicLong idGenerator = new AtomicLong(0L);
private final AtomicLong claimsRemoved = new AtomicLong(0L);
+ private final List<ContentClaimCreationContext> creationContexts =
Collections.synchronizedList(new ArrayList<>());
private ResourceClaimManager claimManager;
private boolean disableRead = false;
@@ -3256,6 +3347,16 @@ public class StandardProcessSessionIT {
return contentClaim;
}
+ @Override
+ public ContentClaim create(final ContentClaimCreationContext context)
throws IOException {
+ creationContexts.add(context);
+ return ContentRepository.super.create(context);
+ }
+
+ public List<ContentClaimCreationContext> getCreationContexts() {
+ return creationContexts;
+ }
+
public ContentClaim create(byte[] content) throws IOException {
final ResourceClaim resourceClaim =
claimManager.newResourceClaim("container", "section",
String.valueOf(idGenerator.getAndIncrement()), false, false);
final StandardContentClaim contentClaim = new
StandardContentClaim(resourceClaim, 0L);
diff --git
a/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/repository/claim/TestStandardContentClaimWriteCache.java
b/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/repository/claim/TestStandardContentClaimWriteCache.java
index 343b325455e..7d2698dba99 100644
---
a/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/repository/claim/TestStandardContentClaimWriteCache.java
+++
b/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/repository/claim/TestStandardContentClaimWriteCache.java
@@ -17,22 +17,29 @@
package org.apache.nifi.controller.repository.claim;
+import org.apache.nifi.controller.repository.ContentClaimCreationContext;
import org.apache.nifi.controller.repository.FileSystemRepository;
+import org.apache.nifi.controller.repository.LossTolerance;
+import
org.apache.nifi.controller.repository.StandardContentClaimCreationContext;
import org.apache.nifi.controller.repository.StandardContentRepositoryContext;
-import org.apache.nifi.controller.repository.TestFileSystemRepository;
import org.apache.nifi.controller.repository.metrics.NopPerformanceTracker;
-import org.apache.nifi.controller.repository.util.DiskUtils;
import org.apache.nifi.events.EventReporter;
import org.apache.nifi.stream.io.StreamUtils;
import org.apache.nifi.util.NiFiProperties;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
-import java.io.File;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
import java.util.Random;
import static org.junit.jupiter.api.Assertions.assertArrayEquals;
@@ -41,31 +48,51 @@ import static
org.junit.jupiter.api.Assertions.assertNotEquals;
import static org.junit.jupiter.api.Assertions.assertNotNull;
public class TestStandardContentClaimWriteCache {
+ private static final ContentClaimCreationContext CREATION_CONTEXT = new
StandardContentClaimCreationContext("component-1", "connector-1",
LossTolerance.LOSS_INTOLERANT);
+
+ private static final int BUFFER_SIZE = 4;
+
+ private static final Path NIFI_PROPERTIES_FILE =
Paths.get("src/test/resources/conf/nifi.properties");
+
+ @TempDir
+ private Path tempDir;
+
+ private final List<FileSystemRepository> repositories = new ArrayList<>();
private FileSystemRepository repository = null;
private StandardResourceClaimManager claimManager = null;
- private final File rootFile = new
File("target/testContentClaimWriteCache");
@BeforeEach
public void setup() throws IOException {
- NiFiProperties nifiProperties =
NiFiProperties.createBasicNiFiProperties(TestFileSystemRepository.class.getResource("/conf/nifi.properties").getFile());
- if (rootFile.exists()) {
- DiskUtils.deleteRecursively(rootFile);
- }
- repository = new FileSystemRepository(nifiProperties);
claimManager = new StandardResourceClaimManager();
- repository.initialize(new
StandardContentRepositoryContext(claimManager, EventReporter.NO_OP));
- repository.purge();
+ repository = createRepository(Map.of());
}
@AfterEach
public void shutdown() throws IOException {
- repository.shutdown();
+ for (final FileSystemRepository createdRepository : repositories) {
+ createdRepository.shutdown();
+ }
+ }
+
+ private FileSystemRepository createRepository(final Map<String, String>
additionalProperties) throws IOException {
+ final Path repositoryDirectory = tempDir.resolve("content-repository-"
+ repositories.size());
+
+ final Map<String, String> properties = new
HashMap<>(additionalProperties);
+
properties.put(NiFiProperties.REPOSITORY_CONTENT_PREFIX.concat("default"),
repositoryDirectory.toString());
+
+ final NiFiProperties nifiProperties =
NiFiProperties.createBasicNiFiProperties(NIFI_PROPERTIES_FILE.toString(),
properties);
+ final FileSystemRepository createdRepository = new
FileSystemRepository(nifiProperties);
+ createdRepository.initialize(new
StandardContentRepositoryContext(claimManager, EventReporter.NO_OP));
+ createdRepository.purge();
+
+ repositories.add(createdRepository);
+ return createdRepository;
}
@Test
public void testFlushWriteCorrectData() throws IOException {
- final ContentClaimWriteCache cache = new
StandardContentClaimWriteCache(repository, new NopPerformanceTracker(),
50_000L, 4);
+ final ContentClaimWriteCache cache = new
StandardContentClaimWriteCache(repository, new NopPerformanceTracker(),
CREATION_CONTEXT, BUFFER_SIZE);
final ContentClaim claim1 = cache.getContentClaim();
assertNotNull(claim1);
@@ -78,10 +105,11 @@ public class TestStandardContentClaimWriteCache {
cache.flush();
assertEquals(13L, claim1.getLength());
- final InputStream in = repository.read(claim1);
- final byte[] buff = new byte[(int) claim1.getLength()];
- StreamUtils.fillBuffer(in, buff);
- assertArrayEquals("hellogood-bye".getBytes(), buff);
+ try (final InputStream in = repository.read(claim1)) {
+ final byte[] buff = new byte[(int) claim1.getLength()];
+ StreamUtils.fillBuffer(in, buff);
+ assertArrayEquals("hellogood-bye".getBytes(), buff);
+ }
final ContentClaim claim2 = cache.getContentClaim();
final OutputStream out2 = cache.write(claim2);
@@ -92,15 +120,18 @@ public class TestStandardContentClaimWriteCache {
cache.flush();
assertEquals(13L, claim2.getLength());
- final InputStream in2 = repository.read(claim2);
- final byte[] buff2 = new byte[(int) claim2.getLength()];
- StreamUtils.fillBuffer(in2, buff2);
- assertArrayEquals("good-dayhello".getBytes(), buff2);
+ try (final InputStream in = repository.read(claim2)) {
+ final byte[] buff = new byte[(int) claim2.getLength()];
+ StreamUtils.fillBuffer(in, buff);
+ assertArrayEquals("good-dayhello".getBytes(), buff);
+ }
+
+ cache.reset();
}
@Test
public void testWriteLargeRollsOverToNewFileOnNext() throws IOException {
- final ContentClaimWriteCache cache = new
StandardContentClaimWriteCache(repository, new NopPerformanceTracker(),
50_000L, 4);
+ final ContentClaimWriteCache cache = new
StandardContentClaimWriteCache(repository, new NopPerformanceTracker(),
CREATION_CONTEXT, BUFFER_SIZE);
final ContentClaim claim1 = cache.getContentClaim();
assertNotNull(claim1);
@@ -142,4 +173,29 @@ public class TestStandardContentClaimWriteCache {
assertEquals(1,
claimManager.getClaimantCount(claim4.getResourceClaim()));
}
+ @Test
+ public void testRolloverBoundaryDeterminedByContentRepository() throws
IOException {
+ final FileSystemRepository smallClaimRepository =
createRepository(Map.of(NiFiProperties.MAX_APPENDABLE_CLAIM_SIZE, "1 KB"));
+ assertEquals(1024L, smallClaimRepository.getMaxAppendableClaimBytes());
+
+ final ContentClaimWriteCache cache = new
StandardContentClaimWriteCache(smallClaimRepository, new
NopPerformanceTracker(), CREATION_CONTEXT, BUFFER_SIZE);
+ final byte[] content = new byte[600];
+
+ final ContentClaim claim1 = cache.getContentClaim();
+ try (final OutputStream out = cache.write(claim1)) {
+ out.write(content);
+ }
+
+ // 600 bytes have been written to the Resource Claim, which is below
the 1 KB limit, so it remains appendable.
+ final ContentClaim claim2 = cache.getContentClaim();
+ assertEquals(claim1.getResourceClaim(), claim2.getResourceClaim());
+
+ try (final OutputStream out = cache.write(claim2)) {
+ out.write(content);
+ }
+
+ // 1,200 bytes have now been written to the Resource Claim, which
exceeds the 1 KB limit.
+ final ContentClaim claim3 = cache.getContentClaim();
+ assertNotEquals(claim1.getResourceClaim(), claim3.getResourceClaim());
+ }
}
diff --git
a/nifi-stateless/nifi-stateless-bundle/nifi-stateless-engine/src/main/java/org/apache/nifi/stateless/repository/StatelessContentClaimWriteCache.java
b/nifi-stateless/nifi-stateless-bundle/nifi-stateless-engine/src/main/java/org/apache/nifi/stateless/repository/StatelessContentClaimWriteCache.java
index cd42e160d26..bd6c3431adc 100644
---
a/nifi-stateless/nifi-stateless-bundle/nifi-stateless-engine/src/main/java/org/apache/nifi/stateless/repository/StatelessContentClaimWriteCache.java
+++
b/nifi-stateless/nifi-stateless-bundle/nifi-stateless-engine/src/main/java/org/apache/nifi/stateless/repository/StatelessContentClaimWriteCache.java
@@ -17,6 +17,7 @@
package org.apache.nifi.stateless.repository;
+import org.apache.nifi.controller.repository.ContentClaimCreationContext;
import org.apache.nifi.controller.repository.ContentRepository;
import org.apache.nifi.controller.repository.claim.ContentClaim;
import org.apache.nifi.controller.repository.claim.ContentClaimWriteCache;
@@ -33,11 +34,13 @@ import java.util.List;
public class StatelessContentClaimWriteCache implements ContentClaimWriteCache
{
private final ContentRepository contentRepository;
private final PerformanceTracker performanceTracker;
+ private final ContentClaimCreationContext creationContext;
private final List<OutputStream> writtenTo = new ArrayList<>();
- public StatelessContentClaimWriteCache(final ContentRepository
contentRepository, final PerformanceTracker performanceTracker) {
+ public StatelessContentClaimWriteCache(final ContentRepository
contentRepository, final PerformanceTracker performanceTracker, final
ContentClaimCreationContext creationContext) {
this.contentRepository = contentRepository;
this.performanceTracker = performanceTracker;
+ this.creationContext = creationContext;
}
@Override
@@ -55,7 +58,7 @@ public class StatelessContentClaimWriteCache implements
ContentClaimWriteCache {
@Override
public ContentClaim getContentClaim() throws IOException {
- return contentRepository.create(false);
+ return contentRepository.create(creationContext);
}
@Override
diff --git
a/nifi-stateless/nifi-stateless-bundle/nifi-stateless-engine/src/main/java/org/apache/nifi/stateless/repository/StatelessRepositoryContext.java
b/nifi-stateless/nifi-stateless-bundle/nifi-stateless-engine/src/main/java/org/apache/nifi/stateless/repository/StatelessRepositoryContext.java
index f201c935357..065e3850273 100644
---
a/nifi-stateless/nifi-stateless-bundle/nifi-stateless-engine/src/main/java/org/apache/nifi/stateless/repository/StatelessRepositoryContext.java
+++
b/nifi-stateless/nifi-stateless-bundle/nifi-stateless-engine/src/main/java/org/apache/nifi/stateless/repository/StatelessRepositoryContext.java
@@ -52,6 +52,6 @@ public class StatelessRepositoryContext extends
AbstractRepositoryContext implem
@Override
public ContentClaimWriteCache createContentClaimWriteCache(final
PerformanceTracker performanceTracker) {
- return new StatelessContentClaimWriteCache(contentRepository,
performanceTracker);
+ return new StatelessContentClaimWriteCache(contentRepository,
performanceTracker, getContentClaimCreationContext());
}
}
