[
https://issues.apache.org/jira/browse/HDDS-17278?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Andrey Yarovoy updated HDDS-17278:
----------------------------------
Description:
*Problem*
With `ozone.fs.datastream.enabled=true`, a JVM that has written a file through
Ratis data streaming doesn't exit
when its main thread returns, unless it explicitly closes its
`FileSystem`/`OzoneClient` first. The Ratis streaming
client's shared Netty worker group runs a {*}{{*}}non-daemon{{*}}{*} thread,
and that thread keeps the JVM alive.
Most Hadoop code doesn't close cached `FileSystem` instances. It relies on
Hadoop's `ClientFinalizer` shutdown hook
to close them at exit. That hook runs only once the JVM starts shutting down,
which the non-daemon thread prevents.
So the client is never closed, and its reference to the shared worker group is
never released. Nothing breaks the
cycle except an external kill.
The normal (non-streaming) write path is not affected: its gRPC Netty threads
(`grpc-default-worker-ELG-*`) are
daemon threads.
*Impact*
Every MapReduce map task that writes with data streaming hangs after it
completes. The AppMaster waits
`mapreduce.task.exit.timeout` (default 60 s), takes a thread dump and kills the
container. The container's slot
stays occupied for that whole time, so {*}{{*}}every extra wave of map tasks
starts 60 s late{{*}}{*}.
Measured with TestDFSIO on a 12-datanode cluster (Ozone 2.2.0, Kerberos, gRPC
TLS, RATIS/THREE, FSO bucket over
`ofs://`), streaming against the normal write path, same day and same cluster:
|TestDFSIO -write, job time|normal|data streaming|
|—|—|—|
|250 files (1 wave)|31 s|31 s|
|500 files (2 waves)|39–70 s|103–133 s (+63 to +72 s)|
|1000 files (3 waves)|51–131 s|186–240 s (+109 to +134 s)|
The penalty is about 60 s per extra wave whatever the file size (128 MB to 1
GB). Across 20 write jobs, total job
time went from 15.2 to 27.8 min, although each task wrote 27% faster with
streaming. Other short-lived clients that
stream and don't close their client hang the same way: CLI tools, Spark or Hive
executors, custom applications.
*Evidence*
YARN logs of a 500-file write (`application_1792563382394_0007`): all 359
first-wave map attempts, one per map
slot, follow the same sequence:
```
23:30:54,052 TaskAttemptImpl: attempt_..._m_000250_0 TaskAttempt Transitioned
from RUNNING to SUCCESS_FINISHING_CONTAINER
23:31:58,011 AbstractLivelinessMonitor: Expired:attempt_..._m_000250_0 Timed
out after 60 secs [TaskAttemptFinishingMonitor]
```
All 359 thread dumps, written to the containers' stdout before the kill, show
the main thread already returned
(`DestroyJavaVM` waiting) and exactly one non-daemon application thread:
```
"NettyClientStreamRpc-workerGroup--thread1" #69 prio=5 os_prio=0 ... runnable
java.lang.Thread.State: RUNNABLE
at org.apache.ratis.thirdparty.io.netty.channel.epoll.Native.epollWait(Native
Method)
at
org.apache.ratis.thirdparty.io.netty.channel.epoll.EpollEventLoop.epollWaitNoTimerChange(EpollEventLoop.java:308)
at
org.apache.ratis.thirdparty.io.netty.channel.epoll.EpollEventLoop.run(EpollEventLoop.java:365)
at
org.apache.ratis.thirdparty.io.netty.util.concurrent.SingleThreadEventExecutor$4.run(SingleThreadEventExecutor.java:1007)
at java.lang.Thread.run(Thread.java:840)
```
Client log of the same task: `raft.netty.dataStream.client.worker-group.share =
true (default)`,
`worker-group.size = 160 (default)`, `Create EpollEventLoopGroup for
NettyClientStreamRpc-workerGroup`.
*Code path (Ratis 3.3.1)*
- `NettyClientStreamRpc$WorkerGroupGetter.SHARED`: with
`raft.netty.dataStream.client.worker-group.share=true` (the
default), one `EventLoopGroup` per JVM, held by a `ReferenceCountedObject` and
shut down (`shutdownGracefully`)
only when the last `NettyClientStreamRpc` is closed.
- `NettyUtils.newEventLoopGroup(name, size, useEpoll)` →
`ConcurrentUtils.newThreadFactory(name)`: creates plain
`Thread`s without `setDaemon(true)`, so each event-loop thread inherits the
daemon flag of the thread that starts
it. Netty starts it lazily from the writer thread (here `main`), so it is
non-daemon.
- Ozone: `XceiverClientRatis` creates the `RaftClient` with its
`DataStreamApi`. The `NettyClientStreamRpc` is
closed only by `XceiverClientManager.close()`, reached from
`FileSystem.close()` / `OzoneClient.close()`.
*How to reproduce*
```java
public static void main(String[] args) throws Exception {
Configurationconf= newOzoneConfiguration();
conf.setBoolean("ozone.fs.datastream.enabled", true);
FileSystemfs= FileSystem.get(URI.create("ofs://<om>/"), conf);
try (FSDataOutputStreamout= fs.create(newPath("/vol/ratis3bucket/f")))
{ out.write(newbyte[8 << 20]); // > ozone.fs.datastream.auto.threshold (4MB),
so it streams }
// no fs.close(): the JVM never exits
}
```
Datanodes need `hdds.container.ratis.datastream.enabled=true`. Adding
`fs.close()` before returning makes it exit,
and so does setting `ozone.fs.datastream.enabled=false`.
*Proposed fix*
1. {*}{{*}}Ratis (root cause, companion issue):{{*}}{*} create the DataStream
client's worker-group threads as daemon threads,
as gRPC does for its client event loops. A client-side I/O pool shouldn't keep
a JVM alive.
2. {*}{{*}}Ozone, until a fixed Ratis is available:{{*}}{*} if Ozone can supply
the thread factory or event loop group used by the
streaming client, use daemon threads. If it can't, document the issue for
`ozone.fs.datastream.enabled`.
3. {*}{{*}}Test:{{*}}{*} a test that runs the reproducer above in a child JVM
and asserts that it exits within a few seconds.
Workaround: lower `mapreduce.task.exit.timeout` (eg. 5000) and
`mapreduce.task.exit.timeout.check-interval-ms` (eg.
1000) for MapReduce jobs, or close the `FileSystem` explicitly in applications.
was:
*Problem*
With `ozone.fs.datastream.enabled=true`, a JVM that has written a file through
Ratis data streaming doesn't exit
when its main thread returns, unless it explicitly closes its
`FileSystem`/`OzoneClient` first. The Ratis streaming
client's shared Netty worker group runs a *{*}non-daemon{*}* thread, and that
thread keeps the JVM alive.
Most Hadoop code doesn't close cached `FileSystem` instances. It relies on
Hadoop's `ClientFinalizer` shutdown hook
to close them at exit. That hook runs only once the JVM starts shutting down,
which the non-daemon thread prevents.
So the client is never closed, and its reference to the shared worker group is
never released. Nothing breaks the
cycle except an external kill.
The normal (non-streaming) write path is not affected: its gRPC Netty threads
(`grpc-default-worker-ELG-*`) are
daemon threads.
*Impact*
Every MapReduce map task that writes with data streaming hangs after it
completes. The AppMaster waits
`mapreduce.task.exit.timeout` (default 60 s), takes a thread dump and kills the
container. The container's slot
stays occupied for that whole time, so *{*}every extra wave of map tasks starts
60 s late{*}*.
Measured with TestDFSIO on a 12-datanode cluster (Ozone 2.2.0, Kerberos, gRPC
TLS, RATIS/THREE, FSO bucket over
`ofs://`), streaming against the normal write path, same day and same cluster:
|TestDFSIO -write, job time|normal|data streaming|
|—|—|—|
|250 files (1 wave)|31 s|31 s|
|500 files (2 waves)|39–70 s|103–133 s (+63 to +72 s)|
|1000 files (3 waves)|51–131 s|186–240 s (+109 to +134 s)|
The penalty is about 60 s per extra wave whatever the file size (128 MB to 1
GB). Across 20 write jobs, total job
time went from 15.2 to 27.8 min, although each task wrote 27% faster with
streaming. Other short-lived clients that
stream and don't close their client hang the same way: CLI tools, Spark or Hive
executors, custom applications.
*Evidence*
YARN logs of a 500-file write (`application_1792563382394_0007`): all 359
first-wave map attempts, one per map
slot, follow the same sequence:
```
23:30:54,052 TaskAttemptImpl: attempt_..._m_000250_0 TaskAttempt Transitioned
from RUNNING to SUCCESS_FINISHING_CONTAINER
23:31:58,011 AbstractLivelinessMonitor: Expired:attempt_..._m_000250_0 Timed
out after 60 secs [TaskAttemptFinishingMonitor]
```
All 359 thread dumps, written to the containers' stdout before the kill, show
the main thread already returned
(`DestroyJavaVM` waiting) and exactly one non-daemon application thread:
```
"NettyClientStreamRpc-workerGroup--thread1" #69 prio=5 os_prio=0 ... runnable
java.lang.Thread.State: RUNNABLE
at org.apache.ratis.thirdparty.io.netty.channel.epoll.Native.epollWait(Native
Method)
at
org.apache.ratis.thirdparty.io.netty.channel.epoll.EpollEventLoop.epollWaitNoTimerChange(EpollEventLoop.java:308)
at
org.apache.ratis.thirdparty.io.netty.channel.epoll.EpollEventLoop.run(EpollEventLoop.java:365)
at
org.apache.ratis.thirdparty.io.netty.util.concurrent.SingleThreadEventExecutor$4.run(SingleThreadEventExecutor.java:1007)
at java.lang.Thread.run(Thread.java:840)
```
Client log of the same task: `raft.netty.dataStream.client.worker-group.share =
true (default)`,
`worker-group.size = 160 (default)`, `Create EpollEventLoopGroup for
NettyClientStreamRpc-workerGroup`.
*Code path (Ratis 3.3.1)*
- `NettyClientStreamRpc$WorkerGroupGetter.SHARED`: with
`raft.netty.dataStream.client.worker-group.share=true` (the
default), one `EventLoopGroup` per JVM, held by a `ReferenceCountedObject` and
shut down (`shutdownGracefully`)
only when the last `NettyClientStreamRpc` is closed.
- `NettyUtils.newEventLoopGroup(name, size, useEpoll)` →
`ConcurrentUtils.newThreadFactory(name)`: creates plain
`Thread`s without `setDaemon(true)`, so each event-loop thread inherits the
daemon flag of the thread that starts
it. Netty starts it lazily from the writer thread (here `main`), so it is
non-daemon.
- Ozone: `XceiverClientRatis` creates the `RaftClient` with its
`DataStreamApi`. The `NettyClientStreamRpc` is
closed only by `XceiverClientManager.close()`, reached from
`FileSystem.close()` / `OzoneClient.close()`.
*How to reproduce*
```java
public static void main(String[] args) throws Exception {
Configurationconf= newOzoneConfiguration();
conf.setBoolean("ozone.fs.datastream.enabled", true);
FileSystemfs= FileSystem.get(URI.create("ofs://<om>/"), conf);
try (FSDataOutputStreamout= fs.create(newPath("/vol/ratis3bucket/f")))
{ out.write(newbyte[8 << 20]); // > ozone.fs.datastream.auto.threshold (4MB),
so it streams }
// no fs.close(): the JVM never exits
}
```
Datanodes need `hdds.container.ratis.datastream.enabled=true`. Adding
`fs.close()` before returning makes it exit,
and so does setting `ozone.fs.datastream.enabled=false`.
#
##
###
#### Proposed fix
1. *{*}Ratis (root cause, companion issue):{*}* create the DataStream client's
worker-group threads as daemon threads,
as gRPC does for its client event loops. A client-side I/O pool shouldn't keep
a JVM alive.
2. *{*}Ozone, until a fixed Ratis is available:{*}* if Ozone can supply the
thread factory or event loop group used by the
streaming client, use daemon threads. If it can't, document the issue for
`ozone.fs.datastream.enabled`.
3. *{*}Test:{*}* a test that runs the reproducer above in a child JVM and
asserts that it exits within a few seconds.
Workaround: lower `mapreduce.task.exit.timeout` (eg. 5000) and
`mapreduce.task.exit.timeout.check-interval-ms` (eg.
1000) for MapReduce jobs, or close the `FileSystem` explicitly in applications.
> Processes that write with Ratis data streaming don't exit until they are
> killed
> -------------------------------------------------------------------------------
>
> Key: HDDS-17278
> URL: https://issues.apache.org/jira/browse/HDDS-17278
> Project: Apache Ozone
> Issue Type: Bug
> Components: Ozone Client
> Reporter: Andrey Yarovoy
> Assignee: Andrey Yarovoy
> Priority: Major
>
> *Problem*
> With `ozone.fs.datastream.enabled=true`, a JVM that has written a file
> through Ratis data streaming doesn't exit
> when its main thread returns, unless it explicitly closes its
> `FileSystem`/`OzoneClient` first. The Ratis streaming
> client's shared Netty worker group runs a {*}{{*}}non-daemon{{*}}{*} thread,
> and that thread keeps the JVM alive.
> Most Hadoop code doesn't close cached `FileSystem` instances. It relies on
> Hadoop's `ClientFinalizer` shutdown hook
> to close them at exit. That hook runs only once the JVM starts shutting down,
> which the non-daemon thread prevents.
> So the client is never closed, and its reference to the shared worker group
> is never released. Nothing breaks the
> cycle except an external kill.
> The normal (non-streaming) write path is not affected: its gRPC Netty threads
> (`grpc-default-worker-ELG-*`) are
> daemon threads.
> *Impact*
> Every MapReduce map task that writes with data streaming hangs after it
> completes. The AppMaster waits
> `mapreduce.task.exit.timeout` (default 60 s), takes a thread dump and kills
> the container. The container's slot
> stays occupied for that whole time, so {*}{{*}}every extra wave of map tasks
> starts 60 s late{{*}}{*}.
> Measured with TestDFSIO on a 12-datanode cluster (Ozone 2.2.0, Kerberos, gRPC
> TLS, RATIS/THREE, FSO bucket over
> `ofs://`), streaming against the normal write path, same day and same cluster:
> |TestDFSIO -write, job time|normal|data streaming|
> |—|—|—|
> |250 files (1 wave)|31 s|31 s|
> |500 files (2 waves)|39–70 s|103–133 s (+63 to +72 s)|
> |1000 files (3 waves)|51–131 s|186–240 s (+109 to +134 s)|
> The penalty is about 60 s per extra wave whatever the file size (128 MB to 1
> GB). Across 20 write jobs, total job
> time went from 15.2 to 27.8 min, although each task wrote 27% faster with
> streaming. Other short-lived clients that
> stream and don't close their client hang the same way: CLI tools, Spark or
> Hive executors, custom applications.
> *Evidence*
> YARN logs of a 500-file write (`application_1792563382394_0007`): all 359
> first-wave map attempts, one per map
> slot, follow the same sequence:
> ```
> 23:30:54,052 TaskAttemptImpl: attempt_..._m_000250_0 TaskAttempt Transitioned
> from RUNNING to SUCCESS_FINISHING_CONTAINER
> 23:31:58,011 AbstractLivelinessMonitor: Expired:attempt_..._m_000250_0 Timed
> out after 60 secs [TaskAttemptFinishingMonitor]
> ```
> All 359 thread dumps, written to the containers' stdout before the kill, show
> the main thread already returned
> (`DestroyJavaVM` waiting) and exactly one non-daemon application thread:
> ```
> "NettyClientStreamRpc-workerGroup--thread1" #69 prio=5 os_prio=0 ... runnable
> java.lang.Thread.State: RUNNABLE
> at org.apache.ratis.thirdparty.io.netty.channel.epoll.Native.epollWait(Native
> Method)
> at
> org.apache.ratis.thirdparty.io.netty.channel.epoll.EpollEventLoop.epollWaitNoTimerChange(EpollEventLoop.java:308)
> at
> org.apache.ratis.thirdparty.io.netty.channel.epoll.EpollEventLoop.run(EpollEventLoop.java:365)
> at
> org.apache.ratis.thirdparty.io.netty.util.concurrent.SingleThreadEventExecutor$4.run(SingleThreadEventExecutor.java:1007)
> at java.lang.Thread.run(Thread.java:840)
> ```
> Client log of the same task: `raft.netty.dataStream.client.worker-group.share
> = true (default)`,
> `worker-group.size = 160 (default)`, `Create EpollEventLoopGroup for
> NettyClientStreamRpc-workerGroup`.
> *Code path (Ratis 3.3.1)*
> - `NettyClientStreamRpc$WorkerGroupGetter.SHARED`: with
> `raft.netty.dataStream.client.worker-group.share=true` (the
> default), one `EventLoopGroup` per JVM, held by a `ReferenceCountedObject`
> and shut down (`shutdownGracefully`)
> only when the last `NettyClientStreamRpc` is closed.
> - `NettyUtils.newEventLoopGroup(name, size, useEpoll)` →
> `ConcurrentUtils.newThreadFactory(name)`: creates plain
> `Thread`s without `setDaemon(true)`, so each event-loop thread inherits the
> daemon flag of the thread that starts
> it. Netty starts it lazily from the writer thread (here `main`), so it is
> non-daemon.
> - Ozone: `XceiverClientRatis` creates the `RaftClient` with its
> `DataStreamApi`. The `NettyClientStreamRpc` is
> closed only by `XceiverClientManager.close()`, reached from
> `FileSystem.close()` / `OzoneClient.close()`.
> *How to reproduce*
> ```java
> public static void main(String[] args) throws Exception {
> Configurationconf= newOzoneConfiguration();
> conf.setBoolean("ozone.fs.datastream.enabled", true);
> FileSystemfs= FileSystem.get(URI.create("ofs://<om>/"), conf);
> try (FSDataOutputStreamout= fs.create(newPath("/vol/ratis3bucket/f")))
> { out.write(newbyte[8 << 20]); // > ozone.fs.datastream.auto.threshold (4MB),
> so it streams }
> // no fs.close(): the JVM never exits
> }
> ```
> Datanodes need `hdds.container.ratis.datastream.enabled=true`. Adding
> `fs.close()` before returning makes it exit,
> and so does setting `ozone.fs.datastream.enabled=false`.
> *Proposed fix*
> 1. {*}{{*}}Ratis (root cause, companion issue):{{*}}{*} create the DataStream
> client's worker-group threads as daemon threads,
> as gRPC does for its client event loops. A client-side I/O pool shouldn't
> keep a JVM alive.
> 2. {*}{{*}}Ozone, until a fixed Ratis is available:{{*}}{*} if Ozone can
> supply the thread factory or event loop group used by the
> streaming client, use daemon threads. If it can't, document the issue for
> `ozone.fs.datastream.enabled`.
> 3. {*}{{*}}Test:{{*}}{*} a test that runs the reproducer above in a child JVM
> and asserts that it exits within a few seconds.
> Workaround: lower `mapreduce.task.exit.timeout` (eg. 5000) and
> `mapreduce.task.exit.timeout.check-interval-ms` (eg.
> 1000) for MapReduce jobs, or close the `FileSystem` explicitly in
> applications.
--
This message was sent by Atlassian Jira
(v10.3.26#10030026)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
