Is it safe to interact with StreamObserver asynchronously, i.e. with Java 8+ CompletableFutures?

I'm looking at the Simple RPC example from grpc.io's basic tutorial:

@Override
public void getFeature(Point request, StreamObserver<Feature> responseObserver) {
  responseObserver.onNext(checkFeature(request));
  responseObserver.onCompleted();
}

...

private Feature checkFeature(Point location) {
  for (Feature feature : features) {
    if (feature.getLocation().getLatitude() == location.getLatitude()
        && feature.getLocation().getLongitude() == location.getLongitude()) {
      return feature;
    }
  }

  // No feature was found, return an unnamed feature.
  return Feature.newBuilder().setName("").setLocation(location).build();
}

Are there any caveats to interacting with the StreamObserver from other threads? For example, say checkFeature() asynchronously hits another service, returning a CompletableFuture:

@Override
public void getFeature(Point request, StreamObserver<Feature> responseObserver) {
  checkFeature(request).
      thenAccept(feature -> responseObserver.onNext(feature));
  responseObserver.onCompleted();
}

Of course the above wouldn't work because the first thread would execute onCompleted() before the feature is returned. So let's fix that:

@Override
public void getFeature(Point request, StreamObserver<Feature> responseObserver) {
  checkFeature(request).
      thenAccept(feature -> {
        responseObserver.onNext(feature);
        responseObserver.onCompleted();
      });
}

I think this should work, but I'm new to Java so I wonder what ramifications there are. For example,

• Will Context.current() be consistent?
• Will anything cause the StreamObserver to destruct or close prematurely besides onNext() for a unary calls and onError()?
• Is there a better practice?

It would be great if someone could also step me through how they reasoned. I tried looking up actual implementations of StreamObserver but I wasn't sure what to look for.

> Will Context.current() be consistent?

Context.current() is using ThreadLocal. if you are accessing it on a different thread, it won't be consistent. You can propagate context between threads. You may find this post useful.

> Will anything cause the StreamObserver to destruct or close prematurely besides onNext() for a unary calls and onError()?

Yes, Normal flow of StreamObserver ends with onError or onCompleted.

As StreamObserver javadoc states, "Since individual StreamObservers are not thread-safe, if multiple threads will be writing to a StreamObserver concurrently, the application must synchronize calls". If you are calling StreamObserver concurrently, you need to synchronize the calls. In other words, if you know for sure it won't be called concurrently even if you are using multiple threads, it should be fine.

If accessing the same StreamObserver on multiple threads, I would try to synchronize it unless the performance is critical since it is error prone. At least, it deserves a nice comment.

Using thenAccept() to call onNext() and onCompleted() is fine, because the observer is not called concurrently from multiple threads.

The "broken" example that called onCompleted() separately was broken also because it could have called the observer from multiple threads without any form of synchronization. StreamObservers may not be called from multiple threads simultaneously.

Using thenAccept() isn't quite right though, as it doesn't handle the case where the future fails. So you need to receive the Throwable as well, which can be done with whenComplete():

@Override
public void getFeature(Point request, StreamObserver<Feature> responseObserver) {
  checkFeature(request).
      whenComplete((feature, t) -> {
        if (t != null) {
          responseObserver.onError(t);
        } else {
          responseObserver.onNext(feature);
          responseObserver.onCompleted();
        }
      });
}

The Context could easily be "wrong" when processing that lambda. Typically we'd look for "architectural" solutions to make sure the context is propagated, like wrapping all application thread pools in Context.currentContextExecutor() when creating them, so individual call sites don't need to be concerned with propagation. I'm not familiar enough with CompletableFuture to provide strategy for it.