[
https://issues.apache.org/jira/browse/NIFI-16544?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Bob Paulin updated NIFI-16544:
------------------------------
Description:
When testing drain functionality on a Connector component I observed a hang on
starting the non-source processors. This was observed during the
prepareForUpdate
```
{
"tid": "853",
"time": "2026-11-04T02:08:47.025510714Z",
"virtual": true,
"name": "",
"state": "WAITING",
"parkBlocker":
{ "object":
"java.util.concurrent.CompletableFuture$Signaller@5506df48" }
,
"stack": [
"java.base\/java.lang.VirtualThread.park(VirtualThread.java:745)",
"java.base\/java.lang.System$1.parkVirtualThread(System.java:2285)",
"java.base\/java.util.concurrent.locks.LockSupport.park(LockSupport.java:221)",
"java.base\/java.util.concurrent.CompletableFuture$Signaller.block(CompletableFuture.java:1885)",
"java.base\/java.util.concurrent.ForkJoinPool.unmanagedBlock(ForkJoinPool.java:4364)",
"java.base\/java.util.concurrent.ForkJoinPool.managedBlock(ForkJoinPool.java:4310)",
"java.base\/java.util.concurrent.CompletableFuture.waitingGet(CompletableFuture.java:1919)",
"java.base\/java.util.concurrent.CompletableFuture.get(CompletableFuture.java:2093)",
"org.apache.nifi.components.connector.AbstractConnector.startNonSourceComponents(AbstractConnector.java:294)",
"org.apache.nifi.components.connector.AbstractConnector.lambda$drainFlowFiles$3(AbstractConnector.java:207)",
"java.base\/java.lang.VirtualThread.run(VirtualThread.java:460)"
]
},
```
The hang started showing up because I'm looking to wait on that future to do
some processing in prepareForUpdate
Exception
```
org.apache.nifi.components.connector.FlowUpdateException: Timed out waiting for
all FlowFiles to drain from the flow
at
com.somepackage.runtime.connectors.myconnector.MyConnector.awaitUrlDrain(MyConnector.java:241)
at
com.somepackage.runtime.connectors.myconnector.MyConnector.prepareForUpdate(MyConnector.java:173)
at
org.apache.nifi.components.connector.StandardConnectorNode.prepareForUpdate(StandardConnectorNode.java:309)
at
org.apache.nifi.components.connector.StandardConnectorRepository.updateConnector(StandardConnectorRepository.java:671)
at
org.apache.nifi.components.connector.StandardConnectorRepository.lambda$applyUpdate$0(StandardConnectorRepository.java:661)
at org.apache.nifi.engine.FlowEngine.lambda$wrap$1(FlowEngine.java:105)
at
java.base/java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:545)
at java.base/java.util.concurrent.FutureTask.run(FutureTask.java:328)
at
java.base/java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:309)
at
java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1090)
at
java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:614)
at java.base/java.lang.Thread.run(Thread.java:1474)
Caused by: java.util.concurrent.TimeoutException
at
java.base/java.util.concurrent.CompletableFuture.timedGet(CompletableFuture.java:1981)
at
java.base/java.util.concurrent.CompletableFuture.get(CompletableFuture.java:2116)
at
com.somepackage.runtime.connectors.myconnector.MyConnector.awaitUrlDrain(MyConnector.java:232)
... 11 common frames omitted
```
When a Processor or Stateless Group was not STOPPED, the start request
logged a warning and returned without invoking the SchedulingAgentCallback.
The Future returned to the caller was never completed, so any caller
blocking on it, such as a Connector draining FlowFiles, hung indefinitely.
Add SchedulingAgentCallback.onStartDeclined(ScheduledState) with a no-op
default. StandardProcessorNode and StandardStatelessGroupNode now invoke
it whenever they decline a start. The scheduler completes the Future for
an already running or starting component, and fails it for a disabled
processor or a stopping Stateless Group.
was:
When testing drain functionality on a Connector component I observed a hang on
starting the non-source processors. This was observed during the drain
function as well as during a drain called during prepareForUpdate
```
{
"tid": "853",
"time": "2026-11-04T02:08:47.025510714Z",
"virtual": true,
"name": "",
"state": "WAITING",
"parkBlocker": {
"object":
"java.util.concurrent.CompletableFuture$Signaller@5506df48"
},
"stack": [
"java.base\/java.lang.VirtualThread.park(VirtualThread.java:745)",
"java.base\/java.lang.System$1.parkVirtualThread(System.java:2285)",
"java.base\/java.util.concurrent.locks.LockSupport.park(LockSupport.java:221)",
"java.base\/java.util.concurrent.CompletableFuture$Signaller.block(CompletableFuture.java:1885)",
"java.base\/java.util.concurrent.ForkJoinPool.unmanagedBlock(ForkJoinPool.java:4364)",
"java.base\/java.util.concurrent.ForkJoinPool.managedBlock(ForkJoinPool.java:4310)",
"java.base\/java.util.concurrent.CompletableFuture.waitingGet(CompletableFuture.java:1919)",
"java.base\/java.util.concurrent.CompletableFuture.get(CompletableFuture.java:2093)",
"org.apache.nifi.components.connector.AbstractConnector.startNonSourceComponents(AbstractConnector.java:294)",
"org.apache.nifi.components.connector.AbstractConnector.lambda$drainFlowFiles$3(AbstractConnector.java:207)",
"java.base\/java.lang.VirtualThread.run(VirtualThread.java:460)"
]
},
```
When a Processor or Stateless Group was not STOPPED, the start request
logged a warning and returned without invoking the SchedulingAgentCallback.
The Future returned to the caller was never completed, so any caller
blocking on it, such as a Connector draining FlowFiles, hung indefinitely.
Add SchedulingAgentCallback.onStartDeclined(ScheduledState) with a no-op
default. StandardProcessorNode and StandardStatelessGroupNode now invoke
it whenever they decline a start. The scheduler completes the Future for
an already running or starting component, and fails it for a disabled
processor or a stopping Stateless Group.
> Complete start Futures when Processor or Stateless Group declines start
> -----------------------------------------------------------------------
>
> Key: NIFI-16544
> URL: https://issues.apache.org/jira/browse/NIFI-16544
> Project: Apache NiFi
> Issue Type: Bug
> Reporter: Bob Paulin
> Priority: Major
> Time Spent: 10m
> Remaining Estimate: 0h
>
> When testing drain functionality on a Connector component I observed a hang
> on starting the non-source processors. This was observed during the
> prepareForUpdate
> ```
> {
> "tid": "853",
> "time": "2026-11-04T02:08:47.025510714Z",
> "virtual": true,
> "name": "",
> "state": "WAITING",
> "parkBlocker":
> { "object":
> "java.util.concurrent.CompletableFuture$Signaller@5506df48" }
> ,
> "stack": [
>
> "java.base\/java.lang.VirtualThread.park(VirtualThread.java:745)",
>
> "java.base\/java.lang.System$1.parkVirtualThread(System.java:2285)",
>
> "java.base\/java.util.concurrent.locks.LockSupport.park(LockSupport.java:221)",
>
> "java.base\/java.util.concurrent.CompletableFuture$Signaller.block(CompletableFuture.java:1885)",
>
> "java.base\/java.util.concurrent.ForkJoinPool.unmanagedBlock(ForkJoinPool.java:4364)",
>
> "java.base\/java.util.concurrent.ForkJoinPool.managedBlock(ForkJoinPool.java:4310)",
>
> "java.base\/java.util.concurrent.CompletableFuture.waitingGet(CompletableFuture.java:1919)",
>
> "java.base\/java.util.concurrent.CompletableFuture.get(CompletableFuture.java:2093)",
>
> "org.apache.nifi.components.connector.AbstractConnector.startNonSourceComponents(AbstractConnector.java:294)",
>
> "org.apache.nifi.components.connector.AbstractConnector.lambda$drainFlowFiles$3(AbstractConnector.java:207)",
> "java.base\/java.lang.VirtualThread.run(VirtualThread.java:460)"
> ]
> },
> ```
> The hang started showing up because I'm looking to wait on that future to do
> some processing in prepareForUpdate
> Exception
> ```
> org.apache.nifi.components.connector.FlowUpdateException: Timed out waiting
> for all FlowFiles to drain from the flow
>
> at
> com.somepackage.runtime.connectors.myconnector.MyConnector.awaitUrlDrain(MyConnector.java:241)
>
> at
> com.somepackage.runtime.connectors.myconnector.MyConnector.prepareForUpdate(MyConnector.java:173)
>
> at
> org.apache.nifi.components.connector.StandardConnectorNode.prepareForUpdate(StandardConnectorNode.java:309)
>
> at
> org.apache.nifi.components.connector.StandardConnectorRepository.updateConnector(StandardConnectorRepository.java:671)
>
> at
> org.apache.nifi.components.connector.StandardConnectorRepository.lambda$applyUpdate$0(StandardConnectorRepository.java:661)
>
> at org.apache.nifi.engine.FlowEngine.lambda$wrap$1(FlowEngine.java:105)
>
> at
> java.base/java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:545)
>
> at java.base/java.util.concurrent.FutureTask.run(FutureTask.java:328)
>
> at
> java.base/java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:309)
>
> at
> java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1090)
>
> at
> java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:614)
>
> at java.base/java.lang.Thread.run(Thread.java:1474)
>
> Caused by: java.util.concurrent.TimeoutException
>
> at
> java.base/java.util.concurrent.CompletableFuture.timedGet(CompletableFuture.java:1981)
>
> at
> java.base/java.util.concurrent.CompletableFuture.get(CompletableFuture.java:2116)
>
> at
> com.somepackage.runtime.connectors.myconnector.MyConnector.awaitUrlDrain(MyConnector.java:232)
>
> ... 11 common frames omitted
> ```
> When a Processor or Stateless Group was not STOPPED, the start request
> logged a warning and returned without invoking the SchedulingAgentCallback.
> The Future returned to the caller was never completed, so any caller
> blocking on it, such as a Connector draining FlowFiles, hung indefinitely.
>
> Add SchedulingAgentCallback.onStartDeclined(ScheduledState) with a no-op
> default. StandardProcessorNode and StandardStatelessGroupNode now invoke
> it whenever they decline a start. The scheduler completes the Future for
> an already running or starting component, and fails it for a disabled
> processor or a stopping Stateless Group.
--
This message was sent by Atlassian Jira
(v10.3.26#10030026)
