fix: NPE in parsing error status during HttpJson server streaming

Author: baeminbo

gax-java/gax-httpjson/src/main/java/com/google/api/gax/httpjson/HttpJsonCallableFactory.java

@@ -206,7 +206,8 @@ ServerStreamingCallable<RequestT, ResponseT> createServerStreamingCallable(
           ClientContext clientContext) {
 
     ServerStreamingCallable<RequestT, ResponseT> callable =
-        new HttpJsonDirectServerStreamingCallable<>(httpJsoncallSettings.getMethodDescriptor());
+        new HttpJsonDirectServerStreamingCallable<>(
+            httpJsoncallSettings.getMethodDescriptor(), httpJsoncallSettings.getTypeRegistry());
 
     if (httpJsoncallSettings.getParamsExtractor() != null) {
       callable =

gax-java/gax-httpjson/src/main/java/com/google/api/gax/httpjson/HttpJsonClientCallImpl.java

@@ -135,6 +135,7 @@ final class HttpJsonClientCallImpl<RequestT, ResponseT>
       HttpTransport httpTransport,
       Executor executor,
       ScheduledExecutorService deadlineCancellationExecutor) {
+    Preconditions.checkNotNull(callOptions.getTypeRegistry());
     this.methodDescriptor = methodDescriptor;
     this.endpoint = endpoint;
     this.callOptions = callOptions;

gax-java/gax-httpjson/src/main/java/com/google/api/gax/httpjson/HttpJsonDirectServerStreamingCallable.java

@@ -34,6 +34,7 @@
 import com.google.api.gax.rpc.ServerStreamingCallable;
 import com.google.api.gax.rpc.StreamController;
 import com.google.common.base.Preconditions;
+import com.google.protobuf.TypeRegistry;
 
 /**
  * {@code HttpJsonDirectServerStreamingCallable} creates server-streaming REST calls.
@@ -49,17 +50,24 @@ class HttpJsonDirectServerStreamingCallable<RequestT, ResponseT>
     extends ServerStreamingCallable<RequestT, ResponseT> {
 
   private final ApiMethodDescriptor<RequestT, ResponseT> descriptor;
+  private final TypeRegistry typeRegistry;
 
-  HttpJsonDirectServerStreamingCallable(ApiMethodDescriptor<RequestT, ResponseT> descriptor) {
+  HttpJsonDirectServerStreamingCallable(
+      ApiMethodDescriptor<RequestT, ResponseT> descriptor, TypeRegistry typeRegistry) {
     this.descriptor = descriptor;
+    this.typeRegistry = typeRegistry;
   }
 
   @Override
   public void call(
-      RequestT request, ResponseObserver<ResponseT> responseObserver, ApiCallContext context) {
-
+      RequestT request, ResponseObserver<ResponseT> responseObserver, ApiCallContext inputContext) {
     Preconditions.checkNotNull(request);
     Preconditions.checkNotNull(responseObserver);
+    HttpJsonCallContext context = HttpJsonCallContext.createDefault().nullToSelf(inputContext);
+
+    context =
+        context.withCallOptions(
+            context.getCallOptions().toBuilder().setTypeRegistry(typeRegistry).build());
 
     HttpJsonClientCall<RequestT, ResponseT> call = HttpJsonClientCalls.newCall(descriptor, context);
     HttpJsonDirectStreamController<RequestT, ResponseT> controller =