You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@beam.apache.org by sc...@apache.org on 2025/02/13 10:00:35 UTC
(beam) branch master updated: [Dataflow Streaming] do not spam logs when onNext called on closed StreamObserver (#33977)
This is an automated email from the ASF dual-hosted git repository.
scwhittle pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/beam.git
The following commit(s) were added to refs/heads/master by this push:
new 1b29e5f2f95 [Dataflow Streaming] do not spam logs when onNext called on closed StreamObserver (#33977)
1b29e5f2f95 is described below
commit 1b29e5f2f956d3065c7db2a1d29bc40d2c144e5e
Author: martin trieu <ma...@google.com>
AuthorDate: Thu Feb 13 02:00:28 2025 -0800
[Dataflow Streaming] do not spam logs when onNext called on closed StreamObserver (#33977)
---
.../worker/windmill/client/ResettableThrowingStreamObserver.java | 9 +++------
1 file changed, 3 insertions(+), 6 deletions(-)
diff --git a/runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/windmill/client/ResettableThrowingStreamObserver.java b/runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/windmill/client/ResettableThrowingStreamObserver.java
index 6b1eb127462..83320a00f22 100644
--- a/runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/windmill/client/ResettableThrowingStreamObserver.java
+++ b/runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/windmill/client/ResettableThrowingStreamObserver.java
@@ -122,12 +122,9 @@ final class ResettableThrowingStreamObserver<T> {
try {
delegate.onError(cancellationException);
} catch (IllegalStateException onErrorException) {
- // If the delegate above was already terminated via onError or onComplete from another
- // thread.
- logger.warn(
- "StreamObserver was already cancelled {} due to error.",
- onErrorException,
- cancellationException);
+ // The delegate above was already terminated via onError or onComplete.
+ // Fallthrough since this is possibly due to queued onNext() calls that are being made from
+ // previously blocked threads.
} catch (RuntimeException onErrorException) {
logger.warn(
"Encountered unexpected error {} when cancelling due to error.",