This is an automated email from the ASF dual-hosted git repository.
bbende 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 0574c82f024 NIFI-15885: Addressed issues that cause Offload to get
stuck: when a … (#11184)
0574c82f024 is described below
commit 0574c82f0247031a5af734328885f8e7530a7c53
Author: Mark Payne <[email protected]>
AuthorDate: Thu Apr 30 12:28:07 2026 -0400
NIFI-15885: Addressed issues that cause Offload to get stuck: when a …
(#11184)
* NIFI-15885: Addressed issues that cause Offload to get stuck: when a
processor extends AbstractSessionFactoryProcessor and stores a reference to a
created ProcessSession but not its factory, garbage collection could cause
framework to be unable to rollback sessions. When a processor is terminated,
ensure we rollback retained sessions even if all threads are completed. Fixed
related bug that caused background thread never to complete when waiting for
processor thread count to reach 1 [...]
* NIFI-15885: Fix to unit test
* NIFI-15885: Allow StandardProcessSession.migrate to unwrap delegating
Session wrappers so FactoryRetainingProcessSession (used to keep an
ActiveProcessSessionFactory reachable for offload/terminate) is recognized as a
StandardProcessSession target. Adds a framework-internal
DelegatingProcessSession contract implemented by the wrapper, and an IT
covering the migrate-to-wrapper path that reproduces the MergeRecord CI failure.
---
.../nifi/controller/StandardProcessorNode.java | 8 +
.../repository/DelegatingProcessSession.java | 39 +++
.../repository/StandardProcessSession.java | 16 +-
.../WeakHashMapProcessSessionFactory.java | 290 ++++++++++++++++++++-
.../controller/scheduling/LifecycleStateTest.java | 107 ++++++++
.../scheduling/StandardProcessScheduler.java | 17 +-
.../repository/StandardProcessSessionIT.java | 26 ++
.../scheduling/TestStandardProcessScheduler.java | 141 ++++++++++
.../nifi/tests/system/clustering/OffloadIT.java | 65 +++++
9 files changed, 701 insertions(+), 8 deletions(-)
diff --git
a/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/controller/StandardProcessorNode.java
b/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/controller/StandardProcessorNode.java
index 79248c56f9b..11c60f38115 100644
---
a/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/controller/StandardProcessorNode.java
+++
b/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/controller/StandardProcessorNode.java
@@ -1905,6 +1905,14 @@ public class StandardProcessorNode extends ProcessorNode
implements Connectable
}
}
}
+ } else if (lifecycleState.isTerminated()) {
+ // Termination was requested while the stop
sequence was waiting for active threads to drain.
+ // LifecycleState.terminate() reset the active
thread count to zero, so the count==1
+ // condition above will never be reached and
rescheduling would loop forever. Complete the
+ // stop action and exit. completeStopAction() is
idempotent if procNode.terminate() already
+ // invoked it.
+ LOG.debug("Stop sequence for {} aborted because
LifecycleState was terminated", this);
+ completeStopAction();
} else {
// Not all of the active threads have finished.
Try again in 100 milliseconds.
executor.schedule(this, 100,
TimeUnit.MILLISECONDS);
diff --git
a/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/controller/repository/DelegatingProcessSession.java
b/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/controller/repository/DelegatingProcessSession.java
new file mode 100644
index 00000000000..62e71f970d4
--- /dev/null
+++
b/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/controller/repository/DelegatingProcessSession.java
@@ -0,0 +1,39 @@
+/*
+ * 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.processor.ProcessSession;
+
+/**
+ * Framework-internal contract implemented by {@link ProcessSession} wrappers
that forward all
+ * operations to an underlying delegate Session. This allows framework code
that needs to operate
+ * on a concrete Session implementation (for example, {@code
StandardProcessSession.migrate}, which
+ * requires its target to be another {@code StandardProcessSession}) to look
through any number of
+ * wrapping layers and recover the underlying Session.
+ *
+ * Extensions must not implement this interface; it is intended for
framework-internal Session
+ * wrappers only.
+ */
+public interface DelegatingProcessSession extends ProcessSession {
+
+ /**
+ * @return the Session that this wrapper forwards operations to. May
itself be a
+ * {@link DelegatingProcessSession} when wrappers are nested.
+ */
+ ProcessSession getDelegate();
+}
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 5410ef7f55d..a97e47326cc 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
@@ -1501,11 +1501,23 @@ public class StandardProcessSession implements
ProcessSession, ProvenanceEventEn
throw new IllegalArgumentException("Must supply at least one
FlowFile to migrate");
}
- if (!(newOwner instanceof StandardProcessSession)) {
+ // Look through any framework-internal Session wrappers (such as the
one used to keep an
+ // ActiveProcessSessionFactory reachable for the offload/terminate
path) so the underlying
+ // StandardProcessSession can be located.
+ ProcessSession resolvedOwner = newOwner;
+ while (resolvedOwner instanceof DelegatingProcessSession delegating) {
+ resolvedOwner = delegating.getDelegate();
+ }
+
+ if (!(resolvedOwner instanceof StandardProcessSession standardOwner)) {
throw new IllegalArgumentException("Cannot migrate from a
StandardProcessSession to a " + newOwner.getClass());
}
- migrate((StandardProcessSession) newOwner, flowFiles);
+ if (standardOwner == this) {
+ throw new IllegalArgumentException("Cannot migrate FlowFiles from
a Process Session to itself");
+ }
+
+ migrate(standardOwner, flowFiles);
}
private synchronized void migrate(final StandardProcessSession newOwner,
Collection<FlowFile> flowFiles) {
diff --git
a/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/controller/repository/WeakHashMapProcessSessionFactory.java
b/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/controller/repository/WeakHashMapProcessSessionFactory.java
index 5fa7c82e6e3..9aadf1a9f0a 100644
---
a/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/controller/repository/WeakHashMapProcessSessionFactory.java
+++
b/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/controller/repository/WeakHashMapProcessSessionFactory.java
@@ -17,12 +17,32 @@
package org.apache.nifi.controller.repository;
+import org.apache.nifi.components.state.Scope;
+import org.apache.nifi.components.state.StateMap;
+import org.apache.nifi.controller.queue.QueueSize;
+import org.apache.nifi.flowfile.FlowFile;
+import org.apache.nifi.processor.FlowFileFilter;
import org.apache.nifi.processor.ProcessSession;
import org.apache.nifi.processor.ProcessSessionFactory;
+import org.apache.nifi.processor.Relationship;
import org.apache.nifi.processor.exception.TerminatedTaskException;
+import org.apache.nifi.processor.io.InputStreamCallback;
+import org.apache.nifi.processor.io.OutputStreamCallback;
+import org.apache.nifi.processor.io.StreamCallback;
+import org.apache.nifi.processor.metrics.CommitTiming;
+import org.apache.nifi.provenance.ProvenanceReporter;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.nio.file.Path;
+import java.util.Collection;
+import java.util.List;
import java.util.Map;
+import java.util.Set;
import java.util.WeakHashMap;
+import java.util.function.Consumer;
+import java.util.regex.Pattern;
public class WeakHashMapProcessSessionFactory implements
ActiveProcessSessionFactory {
private final ProcessSessionFactory delegate;
@@ -39,9 +59,16 @@ public class WeakHashMapProcessSessionFactory implements
ActiveProcessSessionFac
throw new TerminatedTaskException();
}
- final ProcessSession session = delegate.createSession();
- sessionMap.put(session, Boolean.TRUE);
- return session;
+ final ProcessSession delegateSession = delegate.createSession();
+
+ // Wrap the delegate Session so that the returned Session strongly
references this factory. The
+ // LifecycleState tracks ActiveProcessSessionFactory instances in a
WeakHashMap, so without a
+ // back-reference from the Session a Processor that retains the
Session but not the factory across
+ // onTrigger invocations would let the WeakHashMap entry clear,
leaving terminate() unable to roll
+ // back the orphaned Session and preventing offload from completing.
+ final ProcessSession wrapper = new
FactoryRetainingProcessSession(delegateSession, this);
+ sessionMap.put(wrapper, Boolean.TRUE);
+ return wrapper;
}
@Override
@@ -56,4 +83,261 @@ public class WeakHashMapProcessSessionFactory implements
ActiveProcessSessionFac
sessionMap.clear();
}
+
+ /**
+ * A delegating {@link ProcessSession} that strongly references the
factory that produced it. This
+ * back-reference exists solely to keep the factory reachable so that the
{@code LifecycleState}
+ * WeakHashMap entry tracking the factory is not cleared while the Session
is still in use. All
+ * operations are forwarded to the underlying delegate Session.
+ */
+ private static final class FactoryRetainingProcessSession implements
DelegatingProcessSession {
+ private final ProcessSession delegate;
+ @SuppressWarnings("unused") // Retained only to keep the factory
reachable; see class Javadoc.
+ private final WeakHashMapProcessSessionFactory factoryRetention;
+
+ private FactoryRetainingProcessSession(final ProcessSession delegate,
final WeakHashMapProcessSessionFactory factoryRetention) {
+ this.delegate = delegate;
+ this.factoryRetention = factoryRetention;
+ }
+
+ @Override
+ public ProcessSession getDelegate() {
+ return delegate;
+ }
+
+ @Override
+ public void commit() {
+ delegate.commit();
+ }
+
+ @Override
+ public void commitAsync() {
+ delegate.commitAsync();
+ }
+
+ @Override
+ public void commitAsync(final Runnable onSuccess, final
Consumer<Throwable> onFailure) {
+ delegate.commitAsync(onSuccess, onFailure);
+ }
+
+ @Override
+ public void rollback() {
+ delegate.rollback();
+ }
+
+ @Override
+ public void rollback(final boolean penalize) {
+ delegate.rollback(penalize);
+ }
+
+ @Override
+ public void migrate(final ProcessSession newOwner, final
Collection<FlowFile> flowFiles) {
+ delegate.migrate(newOwner, flowFiles);
+ }
+
+ @Override
+ public void migrate(final ProcessSession newOwner) {
+ delegate.migrate(newOwner);
+ }
+
+ @Override
+ public void adjustCounter(final String name, final long delta, final
boolean immediate) {
+ delegate.adjustCounter(name, delta, immediate);
+ }
+
+ @Override
+ public void recordGauge(final String name, final double value, final
CommitTiming commitTiming) {
+ delegate.recordGauge(name, value, commitTiming);
+ }
+
+ @Override
+ public FlowFile get() {
+ return delegate.get();
+ }
+
+ @Override
+ public List<FlowFile> get(final int maxResults) {
+ return delegate.get(maxResults);
+ }
+
+ @Override
+ public List<FlowFile> get(final FlowFileFilter filter) {
+ return delegate.get(filter);
+ }
+
+ @Override
+ public QueueSize getQueueSize() {
+ return delegate.getQueueSize();
+ }
+
+ @Override
+ public FlowFile create() {
+ return delegate.create();
+ }
+
+ @Override
+ public FlowFile create(final FlowFile parent) {
+ return delegate.create(parent);
+ }
+
+ @Override
+ public FlowFile create(final Collection<FlowFile> parents) {
+ return delegate.create(parents);
+ }
+
+ @Override
+ public FlowFile clone(final FlowFile example) {
+ return delegate.clone(example);
+ }
+
+ @Override
+ public FlowFile clone(final FlowFile example, final long offset, final
long size) {
+ return delegate.clone(example, offset, size);
+ }
+
+ @Override
+ public FlowFile penalize(final FlowFile flowFile) {
+ return delegate.penalize(flowFile);
+ }
+
+ @Override
+ public FlowFile putAttribute(final FlowFile flowFile, final String
key, final String value) {
+ return delegate.putAttribute(flowFile, key, value);
+ }
+
+ @Override
+ public FlowFile putAllAttributes(final FlowFile flowFile, final
Map<String, String> attributes) {
+ return delegate.putAllAttributes(flowFile, attributes);
+ }
+
+ @Override
+ public FlowFile removeAttribute(final FlowFile flowFile, final String
key) {
+ return delegate.removeAttribute(flowFile, key);
+ }
+
+ @Override
+ public FlowFile removeAllAttributes(final FlowFile flowFile, final
Set<String> keys) {
+ return delegate.removeAllAttributes(flowFile, keys);
+ }
+
+ @Override
+ public FlowFile removeAllAttributes(final FlowFile flowFile, final
Pattern keyPattern) {
+ return delegate.removeAllAttributes(flowFile, keyPattern);
+ }
+
+ @Override
+ public void transfer(final FlowFile flowFile, final Relationship
relationship) {
+ delegate.transfer(flowFile, relationship);
+ }
+
+ @Override
+ public void transfer(final FlowFile flowFile) {
+ delegate.transfer(flowFile);
+ }
+
+ @Override
+ public void transfer(final Collection<FlowFile> flowFiles) {
+ delegate.transfer(flowFiles);
+ }
+
+ @Override
+ public void transfer(final Collection<FlowFile> flowFiles, final
Relationship relationship) {
+ delegate.transfer(flowFiles, relationship);
+ }
+
+ @Override
+ public void remove(final FlowFile flowFile) {
+ delegate.remove(flowFile);
+ }
+
+ @Override
+ public void remove(final Collection<FlowFile> flowFiles) {
+ delegate.remove(flowFiles);
+ }
+
+ @Override
+ public void read(final FlowFile source, final InputStreamCallback
reader) {
+ delegate.read(source, reader);
+ }
+
+ @Override
+ public InputStream read(final FlowFile flowFile) {
+ return delegate.read(flowFile);
+ }
+
+ @Override
+ public FlowFile merge(final Collection<FlowFile> sources, final
FlowFile destination) {
+ return delegate.merge(sources, destination);
+ }
+
+ @Override
+ public FlowFile merge(final Collection<FlowFile> sources, final
FlowFile destination, final byte[] header, final byte[] footer, final byte[]
demarcator) {
+ return delegate.merge(sources, destination, header, footer,
demarcator);
+ }
+
+ @Override
+ public FlowFile write(final FlowFile source, final
OutputStreamCallback writer) {
+ return delegate.write(source, writer);
+ }
+
+ @Override
+ public FlowFile write(final FlowFile source, final StreamCallback
writer) {
+ return delegate.write(source, writer);
+ }
+
+ @Override
+ public OutputStream write(final FlowFile source) {
+ return delegate.write(source);
+ }
+
+ @Override
+ public FlowFile append(final FlowFile source, final
OutputStreamCallback writer) {
+ return delegate.append(source, writer);
+ }
+
+ @Override
+ public FlowFile importFrom(final Path source, final boolean
keepSourceFile, final FlowFile destination) {
+ return delegate.importFrom(source, keepSourceFile, destination);
+ }
+
+ @Override
+ public FlowFile importFrom(final InputStream source, final FlowFile
destination) {
+ return delegate.importFrom(source, destination);
+ }
+
+ @Override
+ public void exportTo(final FlowFile flowFile, final Path destination,
final boolean append) {
+ delegate.exportTo(flowFile, destination, append);
+ }
+
+ @Override
+ public void exportTo(final FlowFile flowFile, final OutputStream
destination) {
+ delegate.exportTo(flowFile, destination);
+ }
+
+ @Override
+ public ProvenanceReporter getProvenanceReporter() {
+ return delegate.getProvenanceReporter();
+ }
+
+ @Override
+ public void setState(final Map<String, String> state, final Scope
scope) throws IOException {
+ delegate.setState(state, scope);
+ }
+
+ @Override
+ public StateMap getState(final Scope scope) throws IOException {
+ return delegate.getState(scope);
+ }
+
+ @Override
+ public boolean replaceState(final StateMap oldValue, final Map<String,
String> newValue, final Scope scope) throws IOException {
+ return delegate.replaceState(oldValue, newValue, scope);
+ }
+
+ @Override
+ public void clearState(final Scope scope) throws IOException {
+ delegate.clearState(scope);
+ }
+ }
}
diff --git
a/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/test/java/org/apache/nifi/controller/scheduling/LifecycleStateTest.java
b/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/test/java/org/apache/nifi/controller/scheduling/LifecycleStateTest.java
new file mode 100644
index 00000000000..8398907cfa1
--- /dev/null
+++
b/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/test/java/org/apache/nifi/controller/scheduling/LifecycleStateTest.java
@@ -0,0 +1,107 @@
+/*
+ * 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.scheduling;
+
+import org.apache.nifi.controller.repository.ActiveProcessSessionFactory;
+import org.apache.nifi.controller.repository.WeakHashMapProcessSessionFactory;
+import org.apache.nifi.processor.ProcessSession;
+import org.apache.nifi.processor.ProcessSessionFactory;
+import org.junit.jupiter.api.Test;
+
+import java.lang.ref.Reference;
+import java.lang.ref.WeakReference;
+
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+class LifecycleStateTest {
+
+ private static final long GARBAGE_COLLECTION_TIMEOUT_MILLIS = 5_000L;
+
+ /**
+ * Verifies that a Session created by an ActiveProcessSessionFactory is
rolled back when the
+ * LifecycleState is terminated, even when the only strong reference to
the factory has been released.
+ *
+ * This represents the scenario in which a Processor that extends
AbstractSessionFactoryProcessor
+ * (rather than AbstractProcessor) creates a Session and stashes it in a
member field for use across
+ * subsequent onTrigger invocations. ConnectableTask creates a fresh
ActiveProcessSessionFactory for
+ * each invocation and registers it with the LifecycleState via
incrementActiveThreadCount, but it
+ * does not retain a strong reference to that factory once the invocation
returns. The Processor
+ * retains the Session, not the factory.
+ *
+ * If the factory is allowed to become unreachable while the Session is
still in use, terminate()
+ * must still be able to reach the factory in order to roll back the
Session. Otherwise, a stop
+ * followed by a terminate (such as the sequence performed during node
offload and decommission)
+ * leaves the FlowFiles unacknowledged on their source queues, preventing
the queue counts from
+ * dropping to zero and preventing offload from completing.
+ */
+ @Test
+ void testTerminateRollsBackSessionRetainedAfterFactoryReferenceReleased()
throws InterruptedException {
+ final ProcessSession retainedSession = mock(ProcessSession.class);
+ final ProcessSessionFactory delegate =
mock(ProcessSessionFactory.class);
+ when(delegate.createSession()).thenReturn(retainedSession);
+
+ final LifecycleState lifecycleState = new
LifecycleState("component-id");
+
+ final FactoryRegistration registration =
registerAndReleaseFactory(lifecycleState, delegate);
+
+ encourageGarbageCollection(registration.factoryReference());
+
+ lifecycleState.terminate();
+
+ // The session returned to a Processor is a wrapper that delegates to
retainedSession; verify the
+ // delegate mock observes the rollback rather than the wrapper itself.
+ verify(retainedSession).rollback();
+
+ // Keep the registration reachable through verification so that the
wrapper Session it holds (which
+ // strongly references the factory) prevents the LifecycleState
WeakHashMap entry from clearing
+ // before terminate() runs.
+ Reference.reachabilityFence(registration);
+ }
+
+ /**
+ * Performs the registration in a separate stack frame so that no strong
references to the factory
+ * exist on the caller's stack outside of what {@code
WeakHashMapProcessSessionFactory.createSession()}
+ * arranges to retain via the returned Session wrapper.
+ */
+ private FactoryRegistration registerAndReleaseFactory(final LifecycleState
lifecycleState, final ProcessSessionFactory delegate) {
+ final ActiveProcessSessionFactory factory = new
WeakHashMapProcessSessionFactory(delegate);
+ lifecycleState.incrementActiveThreadCount(factory);
+ final ProcessSession sessionWrapper = factory.createSession();
+ return new FactoryRegistration(sessionWrapper, new
WeakReference<>(factory));
+ }
+
+ /**
+ * Applies garbage collection pressure to encourage clearing of the
WeakReference. This surfaces
+ * the orphaned-session scenario when no other reference path keeps the
factory reachable. A
+ * correct implementation that retains the factory through the Session
will simply iterate to
+ * the timeout without the WeakReference being cleared, and the subsequent
terminate() call will
+ * still roll back the Session.
+ */
+ private static void encourageGarbageCollection(final WeakReference<?>
reference) throws InterruptedException {
+ final long deadline = System.currentTimeMillis() +
GARBAGE_COLLECTION_TIMEOUT_MILLIS;
+ while (reference.get() != null && System.currentTimeMillis() <
deadline) {
+ System.gc();
+ Thread.sleep(50L);
+ }
+ }
+
+ private record FactoryRegistration(ProcessSession sessionWrapper,
WeakReference<ActiveProcessSessionFactory> factoryReference) {
+ }
+}
diff --git
a/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/scheduling/StandardProcessScheduler.java
b/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/scheduling/StandardProcessScheduler.java
index 6c5645b28aa..dce3a4cde7b 100644
---
a/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/scheduling/StandardProcessScheduler.java
+++
b/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/scheduling/StandardProcessScheduler.java
@@ -611,14 +611,25 @@ public final class StandardProcessScheduler implements
ProcessScheduler {
}
final LifecycleState state = getLifecycleState(procNode, false, false);
- if (state.getActiveThreadCount() == 0) {
- LOG.debug("Will not terminate {} because it has no active
threads", procNode);
+
+ // Capture the active thread count before invoking state.terminate(),
since that call resets the
+ // count to zero. The captured count drives whether we still need to
interrupt registered threads
+ // and reload the Processor instance below.
+ final int activeThreadsBeforeTermination =
state.getActiveThreadCount();
+
+ // Always terminate the LifecycleState so that any retained
ActiveProcessSessionFactory instances
+ // have terminateActiveSessions() invoked, rolling back Sessions that
the Processor stashed in member
+ // state and never released. This must run even when no Processor
thread is currently in flight;
+ // otherwise an offload can hang indefinitely on FlowFiles
unacknowledged by orphaned Sessions.
+ state.terminate();
+
+ if (activeThreadsBeforeTermination == 0) {
+ LOG.debug("LifecycleState terminated for {}; no active threads to
interrupt", procNode);
return;
}
LOG.debug("Terminating {}", procNode);
- state.terminate();
final int tasksTerminated = procNode.terminate();
getSchedulingAgent(procNode).incrementMaxThreadCount(tasksTerminated);
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 b74ed7e51b0..314e3295ec5 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
@@ -2202,6 +2202,32 @@ public class StandardProcessSessionIT {
session.commit();
}
+ @Test
+ public void testMigrateToDelegatingSessionWrapper() {
+ FlowFile flowFile = session.create();
+ flowFile = session.write(flowFile, out ->
out.write("contents".getBytes(StandardCharsets.UTF_8)));
+
+ final StandardProcessSession newSession = new
StandardProcessSession(context, () -> false, new NopPerformanceTracker());
+
+ // Simulate the framework path that wraps a Session through
WeakHashMapProcessSessionFactory: the
+ // Processor receives the wrapper, and other framework code that holds
the underlying StandardProcessSession
+ // must still be able to migrate FlowFiles to it.
+ final WeakHashMapProcessSessionFactory wrappingFactory = new
WeakHashMapProcessSessionFactory(() -> newSession);
+ final ProcessSession wrapperSession = wrappingFactory.createSession();
+
+ assertTrue(session.isFlowFileKnown(flowFile));
+ assertFalse(newSession.isFlowFileKnown(flowFile));
+
+ session.migrate(wrapperSession, Collections.singleton(flowFile));
+
+ assertFalse(session.isFlowFileKnown(flowFile));
+ assertTrue(newSession.isFlowFileKnown(flowFile));
+
+ newSession.remove(flowFile);
+ newSession.commit();
+ session.commit();
+ }
+
@Test
public void testMigrateAfterTransferToAutoTerminatedRelationship() {
final long start = System.currentTimeMillis();
diff --git
a/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/scheduling/TestStandardProcessScheduler.java
b/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/scheduling/TestStandardProcessScheduler.java
index abadd3b0477..fad7b11af00 100644
---
a/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/scheduling/TestStandardProcessScheduler.java
+++
b/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/scheduling/TestStandardProcessScheduler.java
@@ -37,6 +37,7 @@ import org.apache.nifi.controller.ProcessScheduler;
import org.apache.nifi.controller.ProcessorNode;
import org.apache.nifi.controller.ReloadComponent;
import org.apache.nifi.controller.ReportingTaskNode;
+import org.apache.nifi.controller.ScheduledState;
import org.apache.nifi.controller.StandardProcessorNode;
import org.apache.nifi.controller.TerminationAwareLogger;
import org.apache.nifi.controller.ValidationContextFactory;
@@ -44,6 +45,8 @@ import org.apache.nifi.controller.flow.FlowManager;
import org.apache.nifi.controller.kerberos.KerberosConfig;
import
org.apache.nifi.controller.reporting.StandardReportingInitializationContext;
import org.apache.nifi.controller.reporting.StandardReportingTaskNode;
+import org.apache.nifi.controller.repository.ActiveProcessSessionFactory;
+import org.apache.nifi.controller.repository.WeakHashMapProcessSessionFactory;
import
org.apache.nifi.controller.scheduling.processors.FailOnScheduledProcessor;
import org.apache.nifi.controller.service.ControllerServiceNode;
import org.apache.nifi.controller.service.ControllerServiceProvider;
@@ -61,6 +64,7 @@ import org.apache.nifi.nar.SystemBundle;
import org.apache.nifi.processor.AbstractProcessor;
import org.apache.nifi.processor.ProcessContext;
import org.apache.nifi.processor.ProcessSession;
+import org.apache.nifi.processor.ProcessSessionFactory;
import org.apache.nifi.processor.Processor;
import org.apache.nifi.processor.StandardProcessorInitializationContext;
import org.apache.nifi.processor.StandardValidationContextFactory;
@@ -83,6 +87,8 @@ import org.mockito.Mockito;
import org.mockito.stubbing.Answer;
import java.io.File;
+import java.lang.ref.Reference;
+import java.lang.reflect.Field;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
@@ -101,6 +107,7 @@ import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicReference;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
@@ -711,4 +718,138 @@ public class TestStandardProcessScheduler {
return new StandardProcessScheduler(new FlowEngine(1, "Unit Test",
true), Mockito.mock(FlowController.class),
stateMgrProvider, nifiProperties, new
StandardLifecycleStateManager());
}
+
+ /**
+ * Verifies that {@link
StandardProcessScheduler#terminateProcessor(ProcessorNode)} causes any
+ * {@link ActiveProcessSessionFactory} retained on the {@link
LifecycleState} to have its active
+ * sessions rolled back, even when no Processor thread is currently in
flight.
+ *
+ * Reproduces the scenario in which a Processor that extends {@code
AbstractSessionFactoryProcessor}
+ * has cleanly returned from {@code onTrigger} but stashed the Session in
a member field; the Session
+ * remains unacknowledged on its incoming queue and offload of the node
hangs unless terminate causes
+ * a rollback through the LifecycleState's retained factories.
+ */
+ @Test
+ @Timeout(30)
+ public void
testTerminateProcessorRollsBackRetainedSessionWhenNoActiveThreads() throws
Exception {
+ final TerminationTestHarness harness = createTerminationTestHarness();
+ final ProcessorNode procNode = createSimpleProcessorNode(harness);
+
+ final LifecycleState lifecycleState =
harness.lifecycleStateManager().getOrRegisterLifecycleState(procNode.getIdentifier(),
false, false);
+
+ final ProcessSession retainedSession =
Mockito.mock(ProcessSession.class);
+ final ProcessSessionFactory delegateFactory =
Mockito.mock(ProcessSessionFactory.class);
+ when(delegateFactory.createSession()).thenReturn(retainedSession);
+
+ final WeakHashMapProcessSessionFactory retainedFactory = new
WeakHashMapProcessSessionFactory(delegateFactory);
+ lifecycleState.incrementActiveThreadCount(retainedFactory);
+ final ProcessSession sessionWrapper = retainedFactory.createSession();
+ lifecycleState.decrementActiveThreadCount();
+
+ assertEquals(0, lifecycleState.getActiveThreadCount());
+ assertEquals(ScheduledState.STOPPED, procNode.getScheduledState());
+
+ harness.scheduler().terminateProcessor(procNode);
+
+ Mockito.verify(retainedSession).rollback();
+ // Keep the Session wrapper reachable through verification so that the
factory's WeakHashMap
+ // entry tracking it cannot be cleared by the GC before
terminateActiveSessions() iterates it.
+ Reference.reachabilityFence(sessionWrapper);
+ Reference.reachabilityFence(retainedFactory);
+
+ harness.scheduler().shutdown();
+ }
+
+ /**
+ * Verifies that the stop background poll loop in {@code
StandardProcessorNode.stop()} exits cleanly
+ * once {@link LifecycleState#terminate()} has been invoked, instead of
rescheduling itself every
+ * 100ms forever in the component lifecycle thread pool.
+ *
+ * Without the fix, {@code LifecycleState.terminate()} resets the active
thread count to zero, which
+ * the poll loop interprets as "still waiting for threads to drain" (it is
comparing against 1, which
+ * represents the stop background thread itself), so it keeps rescheduling
and leaks one polling task
+ * per terminated processor.
+ */
+ @Test
+ @Timeout(30)
+ public void testStopBackgroundPollLoopExitsAfterLifecycleStateTerminated()
throws Exception {
+ final TerminationTestHarness harness = createTerminationTestHarness();
+ final ProcessorNode procNode = createSimpleProcessorNode(harness);
+
+ final LifecycleState lifecycleState =
harness.lifecycleStateManager().getOrRegisterLifecycleState(procNode.getIdentifier(),
false, false);
+ lifecycleState.setScheduled(true);
+ // Represents an in-flight onTrigger thread that is wedged and will
not return on its own.
+ lifecycleState.incrementActiveThreadCount(null);
+
+ // The stop sequence requires the Processor to be in RUNNING;
reflectively force it there since this
+ // test does not run a real scheduling agent.
+ forceScheduledState(procNode, ScheduledState.RUNNING);
+
+ harness.scheduler().stopProcessor(procNode,
ProcessorStopLifecycleMethods.TRIGGER_ONSTOPPED);
+
+ // Allow the first poll iteration to run and reschedule itself at
+100ms.
+ Thread.sleep(50L);
+
+ harness.scheduler().terminateProcessor(procNode);
+
+ // Wait long enough for the previously rescheduled poll iteration to
fire after termination, then
+ // assert that the polling task is not continuing to reschedule itself
in the executor queue.
+ Thread.sleep(500L);
+
+ final long deadline = System.currentTimeMillis() + 500L;
+ while (System.currentTimeMillis() < deadline) {
+ final int queueSize =
harness.componentLifeCyclePool().getQueue().size();
+ assertFalse(queueSize > 0, "Stop polling task continued to
reschedule after LifecycleState termination; queue size = " + queueSize);
+ Thread.sleep(20L);
+ }
+
+ harness.scheduler().shutdown();
+ }
+
+ private TerminationTestHarness createTerminationTestHarness() {
+ final FlowController flowController =
Mockito.mock(FlowController.class);
+
when(flowController.getExtensionManager()).thenReturn(extensionManager);
+
when(flowController.getReloadComponent()).thenReturn(Mockito.mock(ReloadComponent.class));
+
when(flowController.getControllerServiceProvider()).thenReturn(serviceProvider);
+
+ final LifecycleStateManager lifecycleStateManager = new
StandardLifecycleStateManager();
+ final FlowEngine componentLifeCyclePool = new FlowEngine(2,
"Termination Test", true);
+
+ final StandardProcessScheduler localScheduler = new
StandardProcessScheduler(componentLifeCyclePool, flowController,
+ stateMgrProvider, nifiProperties, lifecycleStateManager);
+ localScheduler.setSchedulingAgent(SchedulingStrategy.TIMER_DRIVEN,
Mockito.mock(SchedulingAgent.class));
+
+ return new TerminationTestHarness(localScheduler,
lifecycleStateManager, componentLifeCyclePool);
+ }
+
+ private ProcessorNode createSimpleProcessorNode(final
TerminationTestHarness harness) {
+ final String uuid = UUID.randomUUID().toString();
+ final Processor processor = new NoOpProcessor();
+ processor.initialize(new StandardProcessorInitializationContext(uuid,
null, null, null, KerberosConfig.NOT_CONFIGURED));
+
+ final TerminationAwareLogger logger =
Mockito.mock(TerminationAwareLogger.class);
+ final LoggableComponent<Processor> loggableComponent = new
LoggableComponent<>(processor, systemBundle.getBundleDetails().getCoordinate(),
logger);
+ final ProcessorNode procNode = new
StandardProcessorNode(loggableComponent, uuid,
+ new StandardValidationContextFactory(serviceProvider),
harness.scheduler(), serviceProvider, Mockito.mock(ReloadComponent.class),
+ Mockito.mock(VerifiableComponentFactory.class), extensionManager,
new SynchronousValidationTrigger());
+ rootGroup.addProcessor(procNode);
+ return procNode;
+ }
+
+ private static void forceScheduledState(final ProcessorNode procNode,
final ScheduledState targetState) throws Exception {
+ final Field scheduledStateField =
ProcessorNode.class.getDeclaredField("scheduledState");
+ scheduledStateField.setAccessible(true);
+ @SuppressWarnings("unchecked")
+ final AtomicReference<ScheduledState> scheduledStateRef =
(AtomicReference<ScheduledState>) scheduledStateField.get(procNode);
+ scheduledStateRef.set(targetState);
+ }
+
+ public static class NoOpProcessor extends AbstractProcessor {
+ @Override
+ public void onTrigger(final ProcessContext context, final
ProcessSession session) {
+ }
+ }
+
+ private record TerminationTestHarness(StandardProcessScheduler scheduler,
LifecycleStateManager lifecycleStateManager, FlowEngine componentLifeCyclePool)
{
+ }
}
diff --git
a/nifi-system-tests/nifi-system-test-suite/src/test/java/org/apache/nifi/tests/system/clustering/OffloadIT.java
b/nifi-system-tests/nifi-system-test-suite/src/test/java/org/apache/nifi/tests/system/clustering/OffloadIT.java
index 6ea8e2917f2..99334a3a3e3 100644
---
a/nifi-system-tests/nifi-system-test-suite/src/test/java/org/apache/nifi/tests/system/clustering/OffloadIT.java
+++
b/nifi-system-tests/nifi-system-test-suite/src/test/java/org/apache/nifi/tests/system/clustering/OffloadIT.java
@@ -31,6 +31,7 @@ import org.slf4j.LoggerFactory;
import java.io.IOException;
import java.util.Collections;
+import java.util.Map;
import java.util.concurrent.TimeUnit;
public class OffloadIT extends NiFiSystemIT {
@@ -84,6 +85,70 @@ public class OffloadIT extends NiFiSystemIT {
waitForAllNodesConnected();
}
+ /**
+ * Verifies that a node can complete offload after a Processor that
retains Sessions across onTrigger
+ * invocations is stopped and terminated. The HoldInput Processor extends
AbstractSessionFactoryProcessor
+ * and stashes its ProcessSession (not the ProcessSessionFactory) in a
member field. With the configured
+ * Hold Time of one hour, HoldInput never voluntarily releases the
FlowFiles it has accepted, so the
+ * only way for those FlowFiles to be acknowledged on the source queue is
through the framework's stop
+ * and terminate path rolling back the stashed Session.
+ */
+ @Test
+ @Timeout(120)
+ public void
testOffloadCompletesAfterStoppingAndTerminatingProcessorThatRetainsSessions()
throws NiFiClientException, IOException, InterruptedException {
+ ProcessorEntity generate =
getClientUtil().createProcessor("GenerateFlowFile");
+ ProcessorEntity hold = getClientUtil().createProcessor("HoldInput");
+ final ConnectionEntity generateToHold =
getClientUtil().createConnection(generate, hold, "success");
+
+ getClientUtil().setAutoTerminatedRelationships(hold, "success");
+
+ generate = getClientUtil().updateProcessorProperties(generate,
Map.of("File Size", "1 KB"));
+ final ProcessorConfigDTO generateConfig =
generate.getComponent().getConfig();
+ generateConfig.setSchedulingPeriod("0 sec");
+ getClientUtil().updateProcessorConfig(generate, generateConfig);
+
+ // Configure HoldInput with a Hold Time that exceeds the test timeout
so it pulls FlowFiles
+ // into stashed Sessions and never voluntarily releases them.
+ hold = getClientUtil().updateProcessorProperties(hold, Map.of("Hold
Time", "1 hour"));
+
+ getClientUtil().startProcessor(generate);
+ getClientUtil().startProcessor(hold);
+
+ // Confirm GenerateFlowFile has produced FlowFiles into the
connection. HoldInput pulls
+ // 10,000 FlowFiles per onTrigger, so the queue may be drained back to
zero quickly, but
+ // first observing it as non-empty proves both Processors are running
and that HoldInput
+ // is being scheduled with FlowFiles available.
+ waitForQueueNotEmpty(generateToHold.getId());
+
+ // Allow time on each node for HoldInput to perform at least one
onTrigger invocation that pulls
+ // FlowFiles in and stashes the Session.
+ Thread.sleep(2_000L);
+
+ getClientUtil().stopProcessor(generate);
+ getClientUtil().stopProcessor(hold);
+
+ // With HoldInput stopped, terminate it so the framework's terminate
path runs while the previously
+ // created Sessions are still pinning FlowFiles as unacknowledged on
the source queue.
+ getNifiClient().getProcessorClient().terminateProcessor(hold.getId());
+
+ final NodeDTO nodeTwoDto = getNodeDtoByApiPort(5672);
+ disconnectNode(nodeTwoDto);
+ waitForNodeStatus(nodeTwoDto, "DISCONNECTED");
+
+ final String nodeId = nodeTwoDto.getNodeId();
+ getClientUtil().offloadNode(nodeId);
+
+ // Offload waits until the queued FlowFile count on every queue drops
to zero. If the stop and
+ // terminate sequence above failed to roll back the stashed Sessions,
the unacknowledged FlowFiles
+ // on the source queue prevent the count from reaching zero and this
wait will not complete before
+ // the test timeout.
+ waitForNodeStatus(nodeTwoDto, "OFFLOADED");
+
+ getClientUtil().connectNode(nodeId);
+ waitForAllNodesConnected();
+ destroyFlow();
+ }
+
private void disconnectNode(final NodeDTO nodeDto) throws
NiFiClientException, IOException, InterruptedException {
getClientUtil().disconnectNode(nodeDto.getNodeId());
