[
https://issues.apache.org/jira/browse/NIFI-16312?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Joe Witt updated NIFI-16312:
----------------------------
Description:
ExecuteProcess and ExecuteStreamCommand wait for a child process on the
processor
thread with no upper bound, and neither processor has a way to model "the
command
did not complete." The second problem makes the first one unsafe to fix.
h3. 1. No timeout: a hung command holds a processor thread indefinitely
ExecuteStreamCommand waits on the child with an unbounded waitFor():
{code:java}
try {
exitCode = process.waitFor();
} catch (InterruptedException e) {
logger.warn("Command Execution Process was interrupted", e);
}
{code}
ExecuteProcess, when Batch Duration is not set, waits on the reader task with an
unbounded Future.get(). The existing code comment acknowledges the gap:
{code:java}
// we are not creating batches; wait until process terminates.
// NB!!! Maybe get(long timeout, TimeUnit unit) should
// be used to avoid waiting forever.
try {
longRunningProcess.get();
} catch (final InterruptedException ignored) {
} catch (final ExecutionException ee) {
getLogger().error("Process execution failed", ee.getCause());
}
{code}
The reader task has an isScheduled() bail-out, but it is inside the read loop
and
is only evaluated after read() returns data. A command that produces no output
and
never exits blocks in read() forever and never reaches the check.
Stopping the processor does not help: both scheduling agents unschedule with
future.cancel(false), explicitly documented as "stop scheduling to run but do
not
interrupt currently running tasks." Recovery therefore requires an operator to
notice the stuck thread and use Terminate. NIFI-4877 reported this as Critical
with
the repro being a long sleep followed by Stop.
h3. 2. Non-completion is reported as success
ExecuteStreamCommand tracks the result in a plain int field that defaults to 0:
{code:java}
int exitCode;
{code}
If the wait does not produce an exit code, the field keeps its default and the
processor takes the success path:
{code:java}
final Relationship outputFlowFileRelationship = putToAttribute ?
ORIGINAL_RELATIONSHIP
: (exitCode != 0) ? NONZERO_STATUS_RELATIONSHIP :
OUTPUT_STREAM_RELATIONSHIP;
{code}
{code:java}
attributes.put("execution.status", Integer.toString(exitCode));
{code}
The FlowFile is routed to "output stream" with execution.status=0 and possibly
truncated stdout. Today the only path that reaches this is thread interruption,
which in practice comes from Terminate; because Terminate marks the task
terminated
before interrupting, the session subsequently throws TerminatedTaskException and
rolls back, so the incorrect result is usually discarded by the framework.
Adding a timeout removes that safety net. Process.waitFor(timeout, unit) returns
false on expiry rather than throwing, so a timed-out command leaves exitCode at
its
default on a path where the task is not terminated and the session commits
normally.
A timeout implementation that does not first make non-completion explicit would
commit a truncated FlowFile labelled execution.status=0.
h3. Proposed change
* Add an optional "Command Timeout" property to both processors, defaulting to
no
timeout so existing flows are unchanged.
* On expiry, terminate the child with destroy() followed by destroyForcibly()
after
a short grace period, and reap it so no process is leaked.
* Represent the wait outcome explicitly rather than as an int that defaults to
0,
distinguishing "exited with code N", "timed out", and "interrupted". This
mirrors
the approach taken in NIFI-15718.
* ExecuteStreamCommand: route timed-out runs to the existing "nonzero status"
relationship and record the reason in execution.error, plus an attribute
indicating the command was terminated on timeout. No new relationship.
* ExecuteProcess: follow the convention already used there for read failures,
i.e.
remove the partial FlowFile and log an error. No new relationship.
* On interruption, restore the thread's interrupt status and fail the invocation
rather than continuing as if the command had completed.
h3. Out of scope
* The stderr pipe deadlock in NIFI-5024 is already avoided in
ExecuteStreamCommand,
which redirects stderr to a temporary file rather than a pipe.
* No change to the meaning of execution.status for commands that do exit
normally.
h3. Testing
Timeout behaviour is deterministic and unit-testable: run a command that
outlives a
short configured timeout and assert the routing, the attributes, and that the
child
process is no longer alive. The command should be JVM-launched rather than a
shell
builtin so the tests are not skipped on Windows.
was:
ExecuteProcess and ExecuteStreamCommand wait for a child process on the
processor thread. Both catch InterruptedException and do not restore the
interrupt status. After that, they can continue as if the command completed
normally.
ExecuteProcess (unbounded wait and batch sleep):
try {
longRunningProcess.get();
} catch (final InterruptedException ignored) {
}
try {
TimeUnit.NANOSECONDS.sleep(batchNanos);
} catch (final InterruptedException ignored) {
}
If the wait is interrupted, the catch is empty. The processor then proceeds: if
stdout was written, the FlowFile is transferred to success. The interrupt flag
is cleared, so the framework cannot see that the worker was cancelled.
ExecuteStreamCommand (both waitFor paths):
int exitCode; // defaults to 0
try {
exitCode = process.waitFor();
} catch (InterruptedException e) {
logger.warn("Command Execution Process was interrupted", e);
}
exitCode is an instance field that defaults to 0. If waitFor() is interrupted,
it is never assigned, so the command is treated as exit 0. Routing then uses
that status (output vs nonzero-status / execution.status attribute).
This ticket is interrupt handling only:
- Call Thread.currentThread().interrupt() in every InterruptedException catch
in these two processors.
- Do not treat an interrupted wait as a successful exit (ExecuteStreamCommand
must not keep exitCode 0; ExecuteProcess must not route partial output to
success as if the process finished).
- Surefire tests using local commands (no Docker, no cloud).
Out of scope: a Command Timeout / destroyForcibly property. ExecuteProcess
already has an in-code note that Future.get(timeout) should be used to avoid
waiting forever. That is a separate JIRA after this one, so timeout handling
does not also have to invent interrupt policy.
Expected:
- An interrupted wait restores the interrupt flag.
- The session does not report a successful completed command.
- Existing success/nonzero-exit tests still pass.
Actual:
- InterruptedException is ignored or logged, the flag is cleared, and
ExecuteStreamCommand can attribute execution.status=0.
> Add command timeout to ExecuteProcess and ExecuteStreamCommand and fail when
> the command does not complete
> ----------------------------------------------------------------------------------------------------------
>
> Key: NIFI-16312
> URL: https://issues.apache.org/jira/browse/NIFI-16312
> Project: Apache NiFi
> Issue Type: Bug
> Reporter: Joe Witt
> Assignee: Joe Witt
> Priority: Major
>
> ExecuteProcess and ExecuteStreamCommand wait for a child process on the
> processor
> thread with no upper bound, and neither processor has a way to model "the
> command
> did not complete." The second problem makes the first one unsafe to fix.
> h3. 1. No timeout: a hung command holds a processor thread indefinitely
> ExecuteStreamCommand waits on the child with an unbounded waitFor():
> {code:java}
> try {
> exitCode = process.waitFor();
> } catch (InterruptedException e) {
> logger.warn("Command Execution Process was interrupted", e);
> }
> {code}
> ExecuteProcess, when Batch Duration is not set, waits on the reader task with
> an
> unbounded Future.get(). The existing code comment acknowledges the gap:
> {code:java}
> // we are not creating batches; wait until process terminates.
> // NB!!! Maybe get(long timeout, TimeUnit unit) should
> // be used to avoid waiting forever.
> try {
> longRunningProcess.get();
> } catch (final InterruptedException ignored) {
> } catch (final ExecutionException ee) {
> getLogger().error("Process execution failed", ee.getCause());
> }
> {code}
> The reader task has an isScheduled() bail-out, but it is inside the read loop
> and
> is only evaluated after read() returns data. A command that produces no
> output and
> never exits blocks in read() forever and never reaches the check.
> Stopping the processor does not help: both scheduling agents unschedule with
> future.cancel(false), explicitly documented as "stop scheduling to run but do
> not
> interrupt currently running tasks." Recovery therefore requires an operator to
> notice the stuck thread and use Terminate. NIFI-4877 reported this as
> Critical with
> the repro being a long sleep followed by Stop.
> h3. 2. Non-completion is reported as success
> ExecuteStreamCommand tracks the result in a plain int field that defaults to
> 0:
> {code:java}
> int exitCode;
> {code}
> If the wait does not produce an exit code, the field keeps its default and the
> processor takes the success path:
> {code:java}
> final Relationship outputFlowFileRelationship = putToAttribute ?
> ORIGINAL_RELATIONSHIP
> : (exitCode != 0) ? NONZERO_STATUS_RELATIONSHIP :
> OUTPUT_STREAM_RELATIONSHIP;
> {code}
> {code:java}
> attributes.put("execution.status", Integer.toString(exitCode));
> {code}
> The FlowFile is routed to "output stream" with execution.status=0 and possibly
> truncated stdout. Today the only path that reaches this is thread
> interruption,
> which in practice comes from Terminate; because Terminate marks the task
> terminated
> before interrupting, the session subsequently throws TerminatedTaskException
> and
> rolls back, so the incorrect result is usually discarded by the framework.
> Adding a timeout removes that safety net. Process.waitFor(timeout, unit)
> returns
> false on expiry rather than throwing, so a timed-out command leaves exitCode
> at its
> default on a path where the task is not terminated and the session commits
> normally.
> A timeout implementation that does not first make non-completion explicit
> would
> commit a truncated FlowFile labelled execution.status=0.
> h3. Proposed change
> * Add an optional "Command Timeout" property to both processors, defaulting
> to no
> timeout so existing flows are unchanged.
> * On expiry, terminate the child with destroy() followed by destroyForcibly()
> after
> a short grace period, and reap it so no process is leaked.
> * Represent the wait outcome explicitly rather than as an int that defaults
> to 0,
> distinguishing "exited with code N", "timed out", and "interrupted". This
> mirrors
> the approach taken in NIFI-15718.
> * ExecuteStreamCommand: route timed-out runs to the existing "nonzero status"
> relationship and record the reason in execution.error, plus an attribute
> indicating the command was terminated on timeout. No new relationship.
> * ExecuteProcess: follow the convention already used there for read failures,
> i.e.
> remove the partial FlowFile and log an error. No new relationship.
> * On interruption, restore the thread's interrupt status and fail the
> invocation
> rather than continuing as if the command had completed.
> h3. Out of scope
> * The stderr pipe deadlock in NIFI-5024 is already avoided in
> ExecuteStreamCommand,
> which redirects stderr to a temporary file rather than a pipe.
> * No change to the meaning of execution.status for commands that do exit
> normally.
> h3. Testing
> Timeout behaviour is deterministic and unit-testable: run a command that
> outlives a
> short configured timeout and assert the routing, the attributes, and that the
> child
> process is no longer alive. The command should be JVM-launched rather than a
> shell
> builtin so the tests are not skipped on Windows.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
