[Bug]: NPE on Java SDK harness startup on Dataflow

[Abacn]

What happened?

Seen this several times:

org.apache.beam.vendor.grpc.v1p69p0.io.grpc.StatusRuntimeException: CANCELLED: Failed to read message.
	at org.apache.beam.vendor.grpc.v1p69p0.io.grpc.Status.asRuntimeException(Status.java:532)
	at org.apache.beam.vendor.grpc.v1p69p0.io.grpc.stub.ClientCalls$StreamObserverToCallListenerAdapter.onClose(ClientCalls.java:481)
	at org.apache.beam.vendor.grpc.v1p69p0.io.grpc.internal.ClientCallImpl.closeObserver(ClientCallImpl.java:564)
	at org.apache.beam.vendor.grpc.v1p69p0.io.grpc.internal.ClientCallImpl.access$100(ClientCallImpl.java:72)
	at org.apache.beam.vendor.grpc.v1p69p0.io.grpc.internal.ClientCallImpl$ClientStreamListenerImpl$1StreamClosed.runInternal(ClientCallImpl.java:729)
	at org.apache.beam.vendor.grpc.v1p69p0.io.grpc.internal.ClientCallImpl$ClientStreamListenerImpl$1StreamClosed.runInContext(ClientCallImpl.java:710)
	at org.apache.beam.vendor.grpc.v1p69p0.io.grpc.internal.ContextRunnable.run(ContextRunnable.java:37)
	at org.apache.beam.vendor.grpc.v1p69p0.io.grpc.internal.SerializingExecutor.run(SerializingExecutor.java:133)
	at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128)
	at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628)
	at java.base/java.lang.Thread.run(Thread.java:829)
Caused by: java.lang.NullPointerException
	at org.apache.beam.fn.harness.status.BeamFnStatusClient.getMemoryUsage(BeamFnStatusClient.java:164)
	at org.apache.beam.fn.harness.status.BeamFnStatusClient$InboundObserver.onNext(BeamFnStatusClient.java:283)
	at org.apache.beam.fn.harness.status.BeamFnStatusClient$InboundObserver.onNext(BeamFnStatusClient.java:279)
	at org.apache.beam.vendor.grpc.v1p69p0.io.grpc.stub.ClientCalls$StreamObserverToCallListenerAdapter.onMessage(ClientCalls.java:468)
	at org.apache.beam.vendor.grpc.v1p69p0.io.grpc.internal.ClientCallImpl$ClientStreamListenerImpl$1MessagesAvailable.runInternal(ClientCallImpl.java:657)
	at org.apache.beam.vendor.grpc.v1p69p0.io.grpc.internal.ClientCallImpl$ClientStreamListenerImpl$1MessagesAvailable.runInContext(ClientCallImpl.java:644)
	... 5 more

It points to here:

beam/sdks/java/harness/src/main/java/org/apache/beam/fn/harness/status/BeamFnStatusClient.java

Line 164 in ab19118

memory.add(memoryMonitor.describeMemory());

it appears memoryMonitor is null. However, it has been setup in constructor. Not sure how this is possible

The error is transient. Happens for some seconds at pipeline start up on worker but the pipeline can evantually run

Issue Priority

Priority: 2 (default / most bugs should be filed as P2)

Issue Components

• Component: Python SDK
• Component: Java SDK
• Component: Go SDK
• Component: Typescript SDK
• Component: IO connector
• Component: Beam YAML
• Component: Beam examples
• Component: Beam playground
• Component: Beam katas
• Component: Website
• Component: Infrastructure
• Component: Spark Runner
• Component: Flink Runner
• Component: Samza Runner
• Component: Twister2 Runner
• Component: Hazelcast Jet Runner
• Component: Google Cloud Dataflow Runner

[ppawel]

I see the same in all our production pipelines (mix of batch and streaming) with both Beam 2.62.0 and Beam 2.65.0.

Environment: Google Cloud Dataflow, Java SDK (Java 17)

[ppawel]

Also I see a related exception, looks like an exception in exception handler related to the first one?

java.lang.NullPointerException: Cannot invoke "java.util.concurrent.CompletableFuture.completeExceptionally(java.lang.Throwable)" because the return value of "org.apache.beam.fn.harness.status.BeamFnStatusClient.access$300(org.apache.beam.fn.harness.status.BeamFnStatusClient)" is null
	at org.apache.beam.fn.harness.status.BeamFnStatusClient$InboundObserver.onError(BeamFnStatusClient.java:300)
	at org.apache.beam.vendor.grpc.v1p69p0.io.grpc.stub.ClientCalls$StreamObserverToCallListenerAdapter.onClose(ClientCalls.java:481)
	at org.apache.beam.vendor.grpc.v1p69p0.io.grpc.internal.ClientCallImpl.closeObserver(ClientCallImpl.java:564)
	at org.apache.beam.vendor.grpc.v1p69p0.io.grpc.internal.ClientCallImpl.access$100(ClientCallImpl.java:72)
	at org.apache.beam.vendor.grpc.v1p69p0.io.grpc.internal.ClientCallImpl$ClientStreamListenerImpl$1StreamClosed.runInternal(ClientCallImpl.java:729)
	at org.apache.beam.vendor.grpc.v1p69p0.io.grpc.internal.ClientCallImpl$ClientStreamListenerImpl$1StreamClosed.runInContext(ClientCallImpl.java:710)
	at org.apache.beam.vendor.grpc.v1p69p0.io.grpc.internal.ContextRunnable.run(ContextRunnable.java:37)
	at org.apache.beam.vendor.grpc.v1p69p0.io.grpc.internal.SerializingExecutor.run(SerializingExecutor.java:133)
	at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1136)
	at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:635)
	at java.base/java.lang.Thread.run(Thread.java:840)

[scwhittle]

This issue is that the rpc is created before the memorymonitor is initialized. Since the rpc could respond before the memorymonitor is intiailized, this exception can occur. This seems like it would be rare since the rpc response would have to be very fast or MemoryMonitor constructor taking a long time as the memory monitor is otherwise intiialized right after starting the rpc.

Sending #35252 to fix.

[ppawel]

I think this is also causing restarting workers every ~1h in streaming jobs. Symptom is:

Status service is unhealthy: UNAVAILABLE: SDK harness sdk-0-0 has not responded to a status request in 1h5m25.731689364s which is longer than the threshold of 1h 

Shortly after, autoscaler says it raises number of workers to 1 so I guess it thinks it lost the unhealthy worker... this causes buildup of the job backlog because startup of a new worker takes 1-2 minutes...

@scwhittle Is it planned to release a minor update to Beam 2.65 including your patch?

[Abacn]

If I read correctly #35252 fixes the NPE in issue description, which happens at SDK harness startup, while #35126 (comment) reported by @ppawel is a separate issue

[ppawel]

I think it's related because the worker shutdown/restart only happens inside pipelines/jobs that have this original exception in them (the NPE). This NPE also appears in the logs immediately after new worker is started, then after ~1 hour, new worker is replacing the "unhealthy" one and then the NPE happens etc.

So I assume that this NPE prevents the runtime from checking healthiness properly and that's why it kills the worker.

[Abacn]

Thanks for explanation, yeah if so it's a good news (that we now fixed two NPE symptoms)