gRPC OOM Error

[Bjorn De Bakker <de...@gmail.com>]

Hey all

I'm trying to run an Apache Beam pipeline on Flink (hosted on GKE).  I'm
using a custom container image that was built to run trained TensorFlow
models on a series of images, but the base Docker image is
apache/beam_python3.7_sdk:2.30.0, with a few libraries installed on top of
that.

I'm using the Apache Flink operator
<https://github.com/spotify/flink-on-k8s-operator> which is currently being
maintained by Spotify.  Because the pods are processing a lot of data, we
had to significantly increase memory usage of the TaskManagers.  However,
internally the code is throwing an OOM exception at one point, linked to a
gRPC call that is taking too long to complete:

*Output channel stalled for 1023s, outbound thread CHAIN MapPartition
(MapPartition at [1]PerformInference) -> FlatMap (FlatMap at
ExtractOutput[0]) -> Map (Key Extractor) -> GroupCombine (GroupCombine at
GroupCombine:
PerformInferenceAndCombineResults_dep_049/GroupPredictionsByImage) -> Map
(Key Extractor) (1/1). See: https://issues.apache.org/jira/browse/BEAM-4280
<https://issues.apache.org/jira/browse/BEAM-4280> for the history for this
issue.*








*Feb 18, 2022 11:51:05 AM
org.apache.beam.vendor.grpc.v1p36p0.io.grpc.netty.NettyServerTransport
notifyTerminatedINFO: Transport
failedorg.apache.beam.vendor.grpc.v1p36p0.io.netty.util.internal.OutOfDirectMemoryError:
failed to allocate 2097152 byte(s) of direct memory (used: 1205862679, max:
1207959552) at
org.apache.beam.vendor.grpc.v1p36p0.io.netty.util.internal.PlatformDependent.incrementMemoryCounter(PlatformDependent.java:754)
at
org.apache.beam.vendor.grpc.v1p36p0.io.netty.util.internal.PlatformDependent.allocateDirectNoCleaner(PlatformDependent.java:709)
at
org.apache.beam.vendor.grpc.v1p36p0.io.netty.buffer.PoolArena$DirectArena.allocateDirect(PoolArena.java:645)
at
org.apache.beam.vendor.grpc.v1p36p0.io.netty.buffer.PoolArena$DirectArena.newChunk(PoolArena.java:621)
at
org.apache.beam.vendor.grpc.v1p36p0.io.netty.buffer.PoolArena.allocateNormal(PoolArena.java:204)
at
org.apache.beam.vendor.grpc.v1p36p0.io.netty.buffer.PoolArena.tcacheAllocateNormal(PoolArena.java:188)*

I've tried to assign more memory to either the TaskManager or JobManager in
Flink, but the error stays the same and the numbers don't change either,
which tells me that this is not an issue with Flink itself, but with an
underlying process.

I've done some googling, but I couldn't find more information.  As this
isn't anything we touch, but an underlying library, I'm not sure what we
can change to solve the issue.

Any idea what the problem could be?

Thanks.

Kr,

Bjorn

Re: gRPC OOM Error

[Luke Cwik <lc...@google.com>]

There is a known issue[1] that was discovered a few days ago with the
DirectStreamObserver where it is incorrectly not respecting the gRPC
channel's ready status. Only every Nth message checks whether the channel
is ready which means N-1 don't and cause the OOM. You could swap and use
the experiment "beam_fn_api_buffered_stream" which will give you a
different implementation which shouldn't experience this issue. You can
also control the buffer size with the experiment
"beam_fn_api_buffered_stream_buffer_size=<BUFFER_SIZE>" with the default
BUFFER_SIZE=64 if unspecified. If this works for you, please update
BEAM-14184 with your results.

If you would like to try to fix it, the check here is wrong
https://github.com/apache/beam/blob/d62b5d6c43847a3b5ecb1321144cf10c6931bd98/sdks/java/fn-execution/src/main/java/org/apache/beam/sdk/fn/stream/DirectStreamObserver.java#L63

There was an unsuccessful fix attempt with
https://github.com/apache/beam/pull/17122

1: https://issues.apache.org/jira/browse/BEAM-14184

On Thu, Mar 17, 2022 at 8:46 AM Bjorn De Bakker <de...@gmail.com>
wrote:

> Hey all
>
> I'm trying to run an Apache Beam pipeline on Flink (hosted on GKE).  I'm
> using a custom container image that was built to run trained TensorFlow
> models on a series of images, but the base Docker image is
> apache/beam_python3.7_sdk:2.30.0, with a few libraries installed on top of
> that.
>
> I'm using the Apache Flink operator
> <https://github.com/spotify/flink-on-k8s-operator> which is currently
> being maintained by Spotify.  Because the pods are processing a lot of
> data, we had to significantly increase memory usage of the TaskManagers.
> However, internally the code is throwing an OOM exception at one point,
> linked to a gRPC call that is taking too long to complete:
>
> *Output channel stalled for 1023s, outbound thread CHAIN MapPartition
> (MapPartition at [1]PerformInference) -> FlatMap (FlatMap at
> ExtractOutput[0]) -> Map (Key Extractor) -> GroupCombine (GroupCombine at
> GroupCombine:
> PerformInferenceAndCombineResults_dep_049/GroupPredictionsByImage) -> Map
> (Key Extractor) (1/1). See: https://issues.apache.org/jira/browse/BEAM-4280
> <https://issues.apache.org/jira/browse/BEAM-4280> for the history for this
> issue.*
>
>
>
>
>
>
>
>
> *Feb 18, 2022 11:51:05 AM
> org.apache.beam.vendor.grpc.v1p36p0.io.grpc.netty.NettyServerTransport
> notifyTerminatedINFO: Transport
> failedorg.apache.beam.vendor.grpc.v1p36p0.io.netty.util.internal.OutOfDirectMemoryError:
> failed to allocate 2097152 byte(s) of direct memory (used: 1205862679, max:
> 1207959552) at
> org.apache.beam.vendor.grpc.v1p36p0.io.netty.util.internal.PlatformDependent.incrementMemoryCounter(PlatformDependent.java:754)
> at
> org.apache.beam.vendor.grpc.v1p36p0.io.netty.util.internal.PlatformDependent.allocateDirectNoCleaner(PlatformDependent.java:709)
> at
> org.apache.beam.vendor.grpc.v1p36p0.io.netty.buffer.PoolArena$DirectArena.allocateDirect(PoolArena.java:645)
> at
> org.apache.beam.vendor.grpc.v1p36p0.io.netty.buffer.PoolArena$DirectArena.newChunk(PoolArena.java:621)
> at
> org.apache.beam.vendor.grpc.v1p36p0.io.netty.buffer.PoolArena.allocateNormal(PoolArena.java:204)
> at
> org.apache.beam.vendor.grpc.v1p36p0.io.netty.buffer.PoolArena.tcacheAllocateNormal(PoolArena.java:188)*
>
> I've tried to assign more memory to either the TaskManager or JobManager
> in Flink, but the error stays the same and the numbers don't change either,
> which tells me that this is not an issue with Flink itself, but with an
> underlying process.
>
> I've done some googling, but I couldn't find more information.  As this
> isn't anything we touch, but an underlying library, I'm not sure what we
> can change to solve the issue.
>
> Any idea what the problem could be?
>
> Thanks.
>
> Kr,
>
> Bjorn
>