[
https://issues.apache.org/jira/browse/SPARK-59907?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Jasraj Dhanoa updated SPARK-59907:
----------------------------------
Description:
When StreamingShuffleWriter.write() fails partway through, the in-flight
off-heap buffer can be left unreleased: either it is referenced only by a local
variable, or the send completion callback that releases it never runs. In both
cases task completion cleanup does not release it. By default these buffers
have no cleaner, so GC does not reclaim them either.
This only leaves a lasting leak when the failure is a non-fatal exception: a
fatal error such as OutOfDirectMemoryError makes the executor exit, which frees
its memory anyway. With UnsafeRowSerializer these paths are not expected to
throw a non-fatal exception in normal operation, so this is a defensive fix. If
one did occur, the buffer would stay allocated until the executor restarts.
Failure scenarios:
# The serializer throws while writing a record.
# Closing the serialization stream fails before a buffer is sent.
# Allocating or encoding the outgoing network frame fails before it is handed
to the client, so the send completion callback never runs.
Expected behavior: when write() fails, StreamingShuffleWriter should release
the in-flight buffer and return its memory budget before rethrowing the
exception.
Reproduction (tested on 5.0.0-SNAPSHOT, master at commit 44929778453): run a
streaming shuffle write with a serializer that throws on a given record
(scenario 1) or on stream close (scenario 2). After the task fails, the
captured buffer still has refCnt 1 and one buffer's worth of memory budget is
not returned. For scenario 3, send a DataMessage whose data buffer throws while
being encoded: the send completion callback is never invoked, so the buffer and
its memory budget are never returned. With the fix, the buffer is released and
the budget is fully returned in all three cases. Unit tests reproducing all
three scenarios are included in the PR.
was:
When StreamingShuffleWriter.write() fails partway through, the in-flight
off-heap buffer can be left unreleased: either it is referenced only by a local
variable, or the send completion callback that releases it never runs. In both
cases task completion cleanup does not release it. Because these buffers are
not reclaimed by GC, the executor loses that off-heap memory until it restarts.
Failure scenarios:
# The serializer throws while writing a record.
# Closing the serialization stream fails before a buffer is sent.
# Allocating or encoding the outgoing network frame fails (e.g.
OutOfDirectMemoryError) before it is handed to the client, so the send
completion callback never runs.
Expected behavior: when write() fails, StreamingShuffleWriter should release
the in-flight buffer and return its memory budget before rethrowing the
exception.
Reproduction (tested on 5.0.0-SNAPSHOT, master at commit 44929778453): run a
streaming shuffle write with a serializer that throws on a given record
(scenario 1) or on stream close (scenario 2). After the task fails, the
captured buffer still has refCnt 1 and one buffer's worth of memory budget is
not returned. For scenario 3, send a DataMessage whose data buffer throws while
being encoded: the send completion callback is never invoked, so the buffer and
its memory budget are never returned. With the fix, the buffer is released and
the budget is fully returned in all three cases. Unit tests reproducing all
three scenarios will be included in the PR.
> StreamingShuffleWriter leaks in-flight off-heap buffer when write fails
> -----------------------------------------------------------------------
>
> Key: SPARK-59907
> URL: https://issues.apache.org/jira/browse/SPARK-59907
> Project: Spark
> Issue Type: Bug
> Components: Structured Streaming
> Affects Versions: 5.0.0
> Reporter: Jasraj Dhanoa
> Priority: Major
> Labels: pull-request-available
>
> When StreamingShuffleWriter.write() fails partway through, the in-flight
> off-heap buffer can be left unreleased: either it is referenced only by a
> local variable, or the send completion callback that releases it never runs.
> In both cases task completion cleanup does not release it. By default these
> buffers have no cleaner, so GC does not reclaim them either.
> This only leaves a lasting leak when the failure is a non-fatal exception: a
> fatal error such as OutOfDirectMemoryError makes the executor exit, which
> frees its memory anyway. With UnsafeRowSerializer these paths are not
> expected to throw a non-fatal exception in normal operation, so this is a
> defensive fix. If one did occur, the buffer would stay allocated until the
> executor restarts.
> Failure scenarios:
> # The serializer throws while writing a record.
> # Closing the serialization stream fails before a buffer is sent.
> # Allocating or encoding the outgoing network frame fails before it is handed
> to the client, so the send completion callback never runs.
> Expected behavior: when write() fails, StreamingShuffleWriter should release
> the in-flight buffer and return its memory budget before rethrowing the
> exception.
> Reproduction (tested on 5.0.0-SNAPSHOT, master at commit 44929778453): run a
> streaming shuffle write with a serializer that throws on a given record
> (scenario 1) or on stream close (scenario 2). After the task fails, the
> captured buffer still has refCnt 1 and one buffer's worth of memory budget is
> not returned. For scenario 3, send a DataMessage whose data buffer throws
> while being encoded: the send completion callback is never invoked, so the
> buffer and its memory budget are never returned. With the fix, the buffer is
> released and the budget is fully returned in all three cases. Unit tests
> reproducing all three scenarios are included in the PR.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
