Repository: salesforce/reactive-grpc
Pull request #75 (2018-04-21 08:33:28)

Downstream terminal event must invoke upstream cancelation 
On the client side.


Patch:
diff --git a/common/reactive-grpc-common/src/main/java/com/salesforce/reactivegrpc/common/CancellableStreamObserver.java b/common/reactive-grpc-common/src/main/java/com/salesforce/reactivegrpc/common/CancellableStreamObserver.java
index 95a547a..53704e6 100644
--- a/common/reactive-grpc-common/src/main/java/com/salesforce/reactivegrpc/common/CancellableStreamObserver.java
+++ b/common/reactive-grpc-common/src/main/java/com/salesforce/reactivegrpc/common/CancellableStreamObserver.java
@@ -7,17 +7,14 @@
 
 package com.salesforce.reactivegrpc.common;
 
-import io.grpc.Status;
-import io.grpc.StatusException;
-import io.grpc.StatusRuntimeException;
 import io.grpc.stub.ClientCallStreamObserver;
 import io.grpc.stub.ClientResponseObserver;
 
 /**
- * CancellableStreamObserver wraps a {@link io.grpc.stub.StreamObserver} and invokes an onCanceledHandler if
- * {@link io.grpc.stub.StreamObserver#onError(Throwable)} is invoked with a {@link StatusException} or
- * {@link StatusRuntimeException} of type {@code Status.Code.CANCELLED}. This class is used to hook gRPC server
- * cancellation events.
+ * {@code CancellableStreamObserver} is used by {@code ClientCalls} and wraps a {@link io.grpc.stub.StreamObserver},
+ * invoking an {@code onCanceledHandler} when a terminal event occurs ({@code onComplete()} or {@code onError()}.
+ * This is needed to ensure that upstream reactive streams are cancelled when downstream reactive streams are notified
+ * of a terminal event.
  *
  * @param <TRequest>
  * @param <TResponse>
@@ -38,21 +35,17 @@ public class CancellableStreamObserver<TRequest, TResponse> implements ClientRes
 
     @Override
     public void onError(Throwable t) {
-        if (t instanceof StatusException) {
-            if (((StatusException) t).getStatus().getCode() == Status.Code.CANCELLED) {
-                onCanceledHandler.run();
-            }
-        }
-        if (t instanceof StatusRuntimeException) {
-            if (((StatusRuntimeException) t).getStatus().getCode() == Status.Code.CANCELLED) {
-                onCanceledHandler.run();
-            }
-        }
+        // Signal the upstream that it shouldn't produce any more messages
+        onCanceledHandler.run();
+        // Signal the downstream of the error
         delegate.onError(t);
     }
 
     @Override
     public void onCompleted() {
+        // Signal the upstream that it shouldn't produce any more messages
+        onCanceledHandler.run();
+        // Signal the downstream completion
         delegate.onCompleted();
     }
 
diff --git a/common/reactive-grpc-common/src/main/java/com/salesforce/reactivegrpc/common/ReactiveStreamObserverPublisher.java b/common/reactive-grpc-common/src/main/java/com/salesforce/reactivegrpc/common/ReactiveStreamObserverPublisher.java
index e461ced..2f3f0a6 100644
--- a/common/reactive-grpc-common/src/main/java/com/salesforce/reactivegrpc/common/ReactiveStreamObserverPublisher.java
+++ b/common/reactive-grpc-common/src/main/java/com/salesforce/reactivegrpc/common/ReactiveStreamObserverPublisher.java
@@ -40,6 +40,7 @@ public class ReactiveStreamObserverPublisher<T> implements Publisher<T>, StreamO
     private CallStreamObserver callStreamObserver;
     private Subscriber<? super T> subscriber;
     private volatile boolean isCanceled;
+    private volatile boolean abandonDelayedCancel;
 
     // A gRPC server can sometimes send messages before subscribe() has been called and the consumer may not have
     // finished setting up the consumer pipeline. Use a countdown latch to prevent messages from processing before
@@ -89,11 +90,26 @@ public class ReactiveStreamObserverPublisher<T> implements Publisher<T>, StreamO
                     return;
                 }
 
-                isCanceled = true;
                 if (callStreamObserver instanceof ClientCallStreamObserver) {
+                    isCanceled = true;
                     ((ClientCallStreamObserver) callStreamObserver).cancel("Client canceled request", null);
-                } else {
-                    callStreamObserver.onError(Status.CANCELLED.withDescription("Server canceled request").asRuntimeException());
+                } else { // ServerCallStreamObserver
+                    new Thread() {
+                        private final int WAIT_FOR_ERROR_DELAY_MILLS = 100;
+
+                        @Override
+                        public void run() {
+                            try {
+                                Thread.sleep(WAIT_FOR_ERROR_DELAY_MILLS);
+                                if (!abandonDelayedCancel) {
+                                    isCanceled = true;
+                                    callStreamObserver.onError(Status.CANCELLED.withDescription("Server canceled request").asRuntimeException());
+                                }
+                            } catch (Throwable ex) {
+
+                            }
+                        }
+                    }.start();
                 }
             }
         });
@@ -119,7 +135,6 @@ public class ReactiveStreamObserverPublisher<T> implements Publisher<T>, StreamO
         } catch (InterruptedException e) {
 
         }
-
         subscriber.onError(Preconditions.checkNotNull(t));
         // Release the subscriber, we don't need a reference to it anymore
         subscriber = null;
@@ -140,4 +155,8 @@ public class ReactiveStreamObserverPublisher<T> implements Publisher<T>, StreamO
     public boolean isCanceled() {
         return isCanceled;
     }
+
+    public void abortPendingCancel() {
+        abandonDelayedCancel = true;
+    }
 }
diff --git a/pom.xml b/pom.xml
index 4d8e7ba..85fb4fa 100644
--- a/pom.xml
+++ b/pom.xml
@@ -56,7 +56,6 @@
         <grpc.version>1.9.0</grpc.version>
         <protoc.version>3.5.1</protoc.version> <!-- Same version as grpc-proto -->
         <grpc.contrib.version>0.8.0</grpc.contrib.version>
-
         <rxjava.version>2.1.10</rxjava.version>
         <reactor.version>3.1.5.RELEASE</reactor.version>
 
@@ -145,6 +144,12 @@
                 <version>${grpc.version}</version>
             </dependency>
             <!-- Test dependencies -->
+            <dependency>
+                <groupId>com.salesforce.servicelibs</groupId>
+                <artifactId>grpc-testing-contrib</artifactId>
+                <version>${grpc.contrib.version}</version>
+                <scope>test</scope>
+            </dependency>
             <dependency>
                 <groupId>junit</groupId>
                 <artifactId>junit</artifactId>
diff --git a/reactor/reactor-grpc-stub/src/main/java/com/salesforce/reactorgrpc/stub/ServerCalls.java b/reactor/reactor-grpc-stub/src/main/java/com/salesforce/reactorgrpc/stub/ServerCalls.java
index d6c7c16..2803320 100644
--- a/reactor/reactor-grpc-stub/src/main/java/com/salesforce/reactorgrpc/stub/ServerCalls.java
+++ b/reactor/reactor-grpc-stub/src/main/java/com/salesforce/reactorgrpc/stub/ServerCalls.java
@@ -101,6 +101,7 @@ public final class ServerCalls {
                 throwable -> {
                     // Don't try to respond if the server has already canceled the request
                     if (!streamObserverPublisher.isCanceled()) {
+                        streamObserverPublisher.abortPendingCancel();
                         responseObserver.onError(prepareError(throwable));
                     }
                 }
@@ -137,6 +138,7 @@ public final class ServerCalls {
                 },
                 throwable -> {
                     if (!streamObserverPublisher.isCanceled()) {
+                        streamObserverPublisher.abortPendingCancel();
                         subscriber.onError(throwable);
                     }
                 },
diff --git a/rx-java/rxgrpc-stub/src/main/java/com/salesforce/rxgrpc/stub/ServerCalls.java b/rx-java/rxgrpc-stub/src/main/java/com/salesforce/rxgrpc/stub/ServerCalls.java
index a28293a..61e5a8d 100644
--- a/rx-java/rxgrpc-stub/src/main/java/com/salesforce/rxgrpc/stub/ServerCalls.java
+++ b/rx-java/rxgrpc-stub/src/main/java/com/salesforce/rxgrpc/stub/ServerCalls.java
@@ -118,6 +118,7 @@ public final class ServerCalls {
                         public void accept(Throwable throwable) {
                             // Don't try to respond if the server has already canceled the request
                             if (!streamObserverPublisher.isCanceled()) {
+                                streamObserverPublisher.abortPendingCancel();
                                 responseObserver.onError(prepareError(throwable));
                             }
                         }
@@ -161,6 +162,7 @@ public final class ServerCalls {
                         @Override
                         public void accept(Throwable throwable) {
                             if (!streamObserverPublisher.isCanceled()) {
+                                streamObserverPublisher.abortPendingCancel();
                                 subscriber.onError(throwable);
                             }
                         }


Test patch:
diff --git a/common/reactive-grpc-common/src/test/java/com/salesforce/reactivegrpc/common/CancellableStreamObserverTest.java b/common/reactive-grpc-common/src/test/java/com/salesforce/reactivegrpc/common/CancellableStreamObserverTest.java
index 243e91c..acab40a 100644
--- a/common/reactive-grpc-common/src/test/java/com/salesforce/reactivegrpc/common/CancellableStreamObserverTest.java
+++ b/common/reactive-grpc-common/src/test/java/com/salesforce/reactivegrpc/common/CancellableStreamObserverTest.java
@@ -52,21 +52,4 @@ public class CancellableStreamObserverTest {
 
         assertThat(called.get()).isTrue();
     }
-
-    @Test
-    public void otherExceptionDoesNotTriggersHandler() {
-        ClientResponseObserver<Object, Object> delegate = mock(ClientResponseObserver.class);
-        final AtomicBoolean called = new AtomicBoolean(false);
-
-        CancellableStreamObserver<Object, Object> observer = new CancellableStreamObserver<Object, Object>(delegate, new Runnable() {
-            @Override
-            public void run() {
-                called.set(true);
-            }
-        });
-
-        observer.onError(Status.INTERNAL.asRuntimeException());
-
-        assertThat(called.get()).isFalse();
-    }
 }
diff --git a/reactor/reactor-grpc-test/pom.xml b/reactor/reactor-grpc-test/pom.xml
index eb22e58..535af1d 100644
--- a/reactor/reactor-grpc-test/pom.xml
+++ b/reactor/reactor-grpc-test/pom.xml
@@ -87,6 +87,11 @@
             <artifactId>grpc-contrib</artifactId>
             <scope>test</scope>
         </dependency>
+        <dependency>
+            <groupId>com.salesforce.servicelibs</groupId>
+            <artifactId>grpc-testing-contrib</artifactId>
+            <scope>test</scope>
+        </dependency>
     </dependencies>
 
     <build>
diff --git a/reactor/reactor-grpc-test/src/test/java/com/salesforce/reactorgrpc/CancellationPropagationIntegrationTest.java b/reactor/reactor-grpc-test/src/test/java/com/salesforce/reactorgrpc/CancellationPropagationIntegrationTest.java
index eefff27..6702436 100644
--- a/reactor/reactor-grpc-test/src/test/java/com/salesforce/reactorgrpc/CancellationPropagationIntegrationTest.java
+++ b/reactor/reactor-grpc-test/src/test/java/com/salesforce/reactorgrpc/CancellationPropagationIntegrationTest.java
@@ -27,6 +27,7 @@ import java.util.concurrent.atomic.AtomicInteger;
 import java.util.stream.IntStream;
 
 import static org.assertj.core.api.Assertions.assertThat;
+import static org.awaitility.Awaitility.await;
 
 @SuppressWarnings({"unchecked", "Duplicates"})
 public class CancellationPropagationIntegrationTest {
@@ -190,7 +191,10 @@ public class CancellationPropagationIntegrationTest {
                 .doOnError(throwable -> System.out.println(throwable.getMessage()));
 
         StepVerifier.create(observer)
-                .verifyError(StatusRuntimeException.class);
+                .expectNext(protoNum(9))
+                .verifyComplete();
+
+        await().atMost(org.awaitility.Duration.FIVE_HUNDRED_MILLISECONDS).untilTrue(requestWasCanceled);
 
         assertThat(requestWasCanceled.get()).isTrue();
         assertThat(requestDidProduce.get()).isTrue();
@@ -224,7 +228,11 @@ public class CancellationPropagationIntegrationTest {
                 .doOnError(throwable -> System.out.println(throwable.getMessage()));
 
         StepVerifier.create(observer)
-                .verifyError(StatusRuntimeException.class);
+                .expectNext(protoNum(-1))
+                .verifyComplete();
+
+        await().atMost(org.awaitility.Duration.FIVE_HUNDRED_MILLISECONDS).untilTrue(requestWasCanceled);
+
         assertThat(requestWasCanceled.get()).isTrue();
         assertThat(requestDidProduce.get()).isTrue();
     }
@@ -257,7 +265,11 @@ public class CancellationPropagationIntegrationTest {
                 .doOnError(throwable -> System.out.println(throwable.getMessage()));
 
         StepVerifier.create(observer)
-                .verifyError(StatusRuntimeException.class);
+                .expectNext(protoNum(9))
+                .verifyComplete();
+
+        await().atMost(org.awaitility.Duration.FIVE_HUNDRED_MILLISECONDS).untilTrue(requestWasCanceled);
+
         assertThat(requestWasCanceled.get()).isTrue();
         assertThat(requestDidProduce.get()).isTrue();
     }
@@ -290,7 +302,11 @@ public class CancellationPropagationIntegrationTest {
                 .doOnError(throwable -> System.out.println(throwable.getMessage()));
 
         StepVerifier.create(observer)
-                .verifyError(StatusRuntimeException.class);
+                .expectNext(protoNum(-1))
+                .verifyComplete();
+
+        await().atMost(org.awaitility.Duration.FIVE_HUNDRED_MILLISECONDS).untilTrue(requestWasCanceled);
+
         assertThat(requestWasCanceled.get()).isTrue();
         assertThat(requestDidProduce.get()).isTrue();
     }
diff --git a/reactor/reactor-grpc-test/src/test/java/com/salesforce/reactorgrpc/ServerErrorUpstreamCancellationIntegrationTest.java b/reactor/reactor-grpc-test/src/test/java/com/salesforce/reactorgrpc/ServerErrorUpstreamCancellationIntegrationTest.java
new file mode 100644
index 0000000..922189b
--- /dev/null
+++ b/reactor/reactor-grpc-test/src/test/java/com/salesforce/reactorgrpc/ServerErrorUpstreamCancellationIntegrationTest.java
@@ -0,0 +1,87 @@
+package com.salesforce.reactorgrpc;
+
+import com.google.protobuf.Empty;
+import com.salesforce.grpc.testing.contrib.NettyGrpcServerRule;
+import com.salesforce.servicelibs.NumberProto;
+import com.salesforce.servicelibs.ReactorNumbersGrpc;
+import io.grpc.Status;
+import io.grpc.StatusRuntimeException;
+import org.junit.Rule;
+import org.junit.Test;
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+import reactor.test.StepVerifier;
+
+import java.util.Arrays;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+public class ServerErrorUpstreamCancellationIntegrationTest {
+    @Rule
+    public NettyGrpcServerRule serverRule = new NettyGrpcServerRule();
+
+    private static class ExplodeAfterFiveService extends ReactorNumbersGrpc.NumbersImplBase {
+        @Override
+        public Flux<NumberProto.Number> twoWayPressure(Flux<NumberProto.Number> request) {
+            return request.map(x -> kaboom());
+        }
+
+        @Override
+        public Mono<NumberProto.Number> requestPressure(Flux<NumberProto.Number> request) {
+            return request.map(x -> kaboom()).single();
+        }
+
+        @Override
+        public Flux<NumberProto.Number> responsePressure(Mono<Empty> request) {
+            return request.map(x -> kaboom()).flux();
+        }
+
+        private NumberProto.Number kaboom() {
+            throw Status.FAILED_PRECONDITION.asRuntimeException();
+        }
+    }
+
+    @Test
+    public void serverErrorSignalsUpstreamCancellationManyToOne() {
+        serverRule.getServiceRegistry().addService(new ExplodeAfterFiveService());
+        ReactorNumbersGrpc.ReactorNumbersStub stub = ReactorNumbersGrpc.newReactorStub(serverRule.getChannel());
+
+        AtomicBoolean upstreamCancel = new AtomicBoolean(false);
+
+        Mono<NumberProto.Number> observer = Flux.range(0, Integer.MAX_VALUE)
+                .map(this::protoNum)
+                .doOnCancel(() -> upstreamCancel.set(true))
+                .as(stub::requestPressure)
+                .doOnError(System.out::println)
+                .doOnSuccess(i -> System.out.println(i.getNumber(0)));
+
+        StepVerifier.create(observer)
+                .verifyError(StatusRuntimeException.class);
+
+        assertThat(upstreamCancel.get()).isTrue();
+    }
+
+    @Test
+    public void serverErrorSignalsUpstreamCancellationBidi() {
+        serverRule.getServiceRegistry().addService(new ExplodeAfterFiveService());
+        ReactorNumbersGrpc.ReactorNumbersStub stub = ReactorNumbersGrpc.newReactorStub(serverRule.getChannel());
+
+        AtomicBoolean upstreamCancel = new AtomicBoolean(false);
+
+        Flux<NumberProto.Number> subscriber = Flux.range(0, Integer.MAX_VALUE)
+                .map(this::protoNum)
+                .doOnCancel(() -> upstreamCancel.set(true))
+                .compose(stub::twoWayPressure)
+                .doOnNext(i -> System.out.println(i.getNumber(0)));
+
+        StepVerifier.create(subscriber)
+                .verifyError(StatusRuntimeException.class);
+        assertThat(upstreamCancel.get()).isTrue();
+    }
+
+    private NumberProto.Number protoNum(int i) {
+        Integer[] ints = {i};
+        return NumberProto.Number.newBuilder().addAllNumber(Arrays.asList(ints)).build();
+    }
+}
diff --git a/reactor/reactor-grpc-test/src/test/java/com/salesforce/reactorgrpc/UnexpectedServerErrorIntegrationTest.java b/reactor/reactor-grpc-test/src/test/java/com/salesforce/reactorgrpc/UnexpectedServerErrorIntegrationTest.java
index 11c5c95..5ece123 100644
--- a/reactor/reactor-grpc-test/src/test/java/com/salesforce/reactorgrpc/UnexpectedServerErrorIntegrationTest.java
+++ b/reactor/reactor-grpc-test/src/test/java/com/salesforce/reactorgrpc/UnexpectedServerErrorIntegrationTest.java
@@ -18,6 +18,9 @@ import reactor.test.StepVerifier;
 
 import java.time.Duration;
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import static org.assertj.core.api.Assertions.assertThat;
 
 @SuppressWarnings("unchecked")
 public class UnexpectedServerErrorIntegrationTest {
@@ -100,7 +103,7 @@ public class UnexpectedServerErrorIntegrationTest {
         Mono<HelloResponse> resp = stub.sayHelloReqStream(req);
 
         StepVerifier.create(resp)
-                .verifyErrorMatches(t -> t instanceof StatusRuntimeException && ((StatusRuntimeException)t).getStatus().getCode() == Status.Code.CANCELLED);
+                .verifyErrorMatches(t -> t instanceof StatusRuntimeException && ((StatusRuntimeException)t).getStatus().getCode() == Status.Code.INTERNAL);
     }
 
     @Test
@@ -110,7 +113,7 @@ public class UnexpectedServerErrorIntegrationTest {
         Flux<HelloResponse> resp = stub.sayHelloBothStream(req);
 
         StepVerifier.create(resp)
-                .verifyErrorMatches(t -> t instanceof StatusRuntimeException && ((StatusRuntimeException)t).getStatus().getCode() == Status.Code.CANCELLED);
+                .verifyErrorMatches(t -> t instanceof StatusRuntimeException && ((StatusRuntimeException)t).getStatus().getCode() == Status.Code.INTERNAL);
     }
 
 }
diff --git a/rx-java/rxgrpc-test/pom.xml b/rx-java/rxgrpc-test/pom.xml
index 82935c2..7ab0bd4 100644
--- a/rx-java/rxgrpc-test/pom.xml
+++ b/rx-java/rxgrpc-test/pom.xml
@@ -81,6 +81,11 @@
             <artifactId>grpc-contrib</artifactId>
             <scope>test</scope>
         </dependency>
+        <dependency>
+            <groupId>com.salesforce.servicelibs</groupId>
+            <artifactId>grpc-testing-contrib</artifactId>
+            <scope>test</scope>
+        </dependency>
     </dependencies>
 
     <build>
diff --git a/rx-java/rxgrpc-test/src/test/java/com/salesforce/rxgrpc/CancellationPropagationIntegrationTest.java b/rx-java/rxgrpc-test/src/test/java/com/salesforce/rxgrpc/CancellationPropagationIntegrationTest.java
index abfc6d6..13b85f6 100644
--- a/rx-java/rxgrpc-test/src/test/java/com/salesforce/rxgrpc/CancellationPropagationIntegrationTest.java
+++ b/rx-java/rxgrpc-test/src/test/java/com/salesforce/rxgrpc/CancellationPropagationIntegrationTest.java
@@ -16,6 +16,7 @@ import io.reactivex.Single;
 import io.reactivex.disposables.Disposable;
 import io.reactivex.observers.TestObserver;
 import io.reactivex.subscribers.TestSubscriber;
+import org.awaitility.Duration;
 import org.junit.AfterClass;
 import org.junit.Before;
 import org.junit.BeforeClass;
@@ -28,6 +29,7 @@ import java.util.concurrent.atomic.AtomicInteger;
 import java.util.stream.IntStream;
 
 import static org.assertj.core.api.Assertions.assertThat;
+import static org.awaitility.Awaitility.await;
 
 @SuppressWarnings({"unchecked", "Duplicates"})
 public class CancellationPropagationIntegrationTest {
@@ -193,9 +195,11 @@ public class CancellationPropagationIntegrationTest {
                 .test();
 
         observer.awaitTerminalEvent(3, TimeUnit.SECONDS);
-
-        observer.assertError(StatusRuntimeException.class);
+        observer.assertComplete();
         observer.assertTerminated();
+
+        await().atMost(Duration.FIVE_HUNDRED_MILLISECONDS).untilTrue(requestWasCanceled);
+
         assertThat(requestWasCanceled.get()).isTrue();
         assertThat(requestDidProduce.get()).isTrue();
     }
@@ -229,8 +233,11 @@ public class CancellationPropagationIntegrationTest {
                 .test();
 
         observer.awaitTerminalEvent();
-        observer.assertError(StatusRuntimeException.class);
+        observer.assertComplete();
         observer.assertTerminated();
+
+        await().atMost(Duration.FIVE_HUNDRED_MILLISECONDS).untilTrue(requestWasCanceled);
+
         assertThat(requestWasCanceled.get()).isTrue();
         assertThat(requestDidProduce.get()).isTrue();
     }
diff --git a/rx-java/rxgrpc-test/src/test/java/com/salesforce/rxgrpc/ServerErrorUpstreamCancellationIntegrationTest.java b/rx-java/rxgrpc-test/src/test/java/com/salesforce/rxgrpc/ServerErrorUpstreamCancellationIntegrationTest.java
new file mode 100644
index 0000000..b5a4c0a
--- /dev/null
+++ b/rx-java/rxgrpc-test/src/test/java/com/salesforce/rxgrpc/ServerErrorUpstreamCancellationIntegrationTest.java
@@ -0,0 +1,93 @@
+package com.salesforce.rxgrpc;
+
+import com.google.protobuf.Empty;
+import com.salesforce.grpc.testing.contrib.NettyGrpcServerRule;
+import com.salesforce.servicelibs.NumberProto;
+import com.salesforce.servicelibs.RxNumbersGrpc;
+import io.grpc.Status;
+import io.grpc.StatusRuntimeException;
+import io.reactivex.Flowable;
+import io.reactivex.Single;
+import io.reactivex.observers.TestObserver;
+import io.reactivex.subscribers.TestSubscriber;
+import org.junit.Rule;
+import org.junit.Test;
+
+import java.util.Arrays;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicReference;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+public class ServerErrorUpstreamCancellationIntegrationTest {
+    @Rule
+    public NettyGrpcServerRule serverRule = new NettyGrpcServerRule();
+
+    private static class ExplodeAfterFiveService extends RxNumbersGrpc.NumbersImplBase {
+        @Override
+        public Flowable<NumberProto.Number> twoWayPressure(Flowable<NumberProto.Number> request) {
+            return request.map(x -> kaboom());
+        }
+
+        @Override
+        public Single<NumberProto.Number> requestPressure(Flowable<NumberProto.Number> request) {
+            return request.map(x -> kaboom()).firstOrError();
+        }
+
+        @Override
+        public Flowable<NumberProto.Number> responsePressure(Single<Empty> request) {
+            return request.map(x -> kaboom()).toFlowable();
+        }
+
+        private NumberProto.Number kaboom() {
+            throw Status.FAILED_PRECONDITION.asRuntimeException();
+        }
+    }
+
+    @Test
+    public void serverErrorSignalsUpstreamCancellationManyToOne() {
+        serverRule.getServiceRegistry().addService(new ExplodeAfterFiveService());
+        RxNumbersGrpc.RxNumbersStub stub = RxNumbersGrpc.newRxStub(serverRule.getChannel());
+
+        AtomicBoolean upstreamCancel = new AtomicBoolean(false);
+        AtomicReference<Throwable> throwable = new AtomicReference<>();
+
+        TestObserver<NumberProto.Number> observer = Flowable.range(0, Integer.MAX_VALUE)
+                .map(this::protoNum)
+                .doOnCancel(() -> upstreamCancel.set(true))
+                .as(stub::requestPressure)
+                .doOnError(throwable::set)
+                .doOnSuccess(i -> System.out.println(i.getNumber(0)))
+                .test();
+
+        observer.awaitTerminalEvent(3, TimeUnit.SECONDS);
+        observer.assertError(StatusRuntimeException.class);
+        assertThat(upstreamCancel.get()).isTrue();
+        assertThat(((StatusRuntimeException) throwable.get()).getStatus()).isEqualTo(Status.FAILED_PRECONDITION);
+    }
+
+    @Test
+    public void serverErrorSignalsUpstreamCancellationBidi() {
+        serverRule.getServiceRegistry().addService(new ExplodeAfterFiveService());
+        RxNumbersGrpc.RxNumbersStub stub = RxNumbersGrpc.newRxStub(serverRule.getChannel());
+
+        AtomicBoolean upstreamCancel = new AtomicBoolean(false);
+
+        TestSubscriber<NumberProto.Number> subscriber = Flowable.range(0, Integer.MAX_VALUE)
+                .map(this::protoNum)
+                .doOnCancel(() -> upstreamCancel.set(true))
+                .compose(stub::twoWayPressure)
+                .doOnNext(i -> System.out.println(i.getNumber(0)))
+                .test();
+
+        subscriber.awaitTerminalEvent(3, TimeUnit.SECONDS);
+        subscriber.assertError(StatusRuntimeException.class);
+        assertThat(upstreamCancel.get()).isTrue();
+    }
+
+    private NumberProto.Number protoNum(int i) {
+        Integer[] ints = {i};
+        return NumberProto.Number.newBuilder().addAllNumber(Arrays.asList(ints)).build();
+    }
+}
diff --git a/rx-java/rxgrpc-test/src/test/java/com/salesforce/rxgrpc/UnexpectedServerErrorIntegrationTest.java b/rx-java/rxgrpc-test/src/test/java/com/salesforce/rxgrpc/UnexpectedServerErrorIntegrationTest.java
index a108e3f..2a4ea78 100644
--- a/rx-java/rxgrpc-test/src/test/java/com/salesforce/rxgrpc/UnexpectedServerErrorIntegrationTest.java
+++ b/rx-java/rxgrpc-test/src/test/java/com/salesforce/rxgrpc/UnexpectedServerErrorIntegrationTest.java
@@ -101,7 +101,7 @@ public class UnexpectedServerErrorIntegrationTest {
         test.awaitTerminalEvent(3, TimeUnit.SECONDS);
         test.assertError(t -> t instanceof StatusRuntimeException);
         // Flowable requests get canceled when unexpected errors happen
-        test.assertError(t -> ((StatusRuntimeException)t).getStatus().getCode() == Status.Code.CANCELLED);
+        test.assertError(t -> ((StatusRuntimeException)t).getStatus().getCode() == Status.Code.INTERNAL);
     }
 
     @Test
@@ -113,6 +113,6 @@ public class UnexpectedServerErrorIntegrationTest {
 
         test.awaitTerminalEvent(3, TimeUnit.SECONDS);
         test.assertError(t -> t instanceof StatusRuntimeException);
-        test.assertError(t -> ((StatusRuntimeException)t).getStatus().getCode() == Status.Code.CANCELLED);
+        test.assertError(t -> ((StatusRuntimeException)t).getStatus().getCode() == Status.Code.INTERNAL);
     }
 }