gRPC Trailers-Only responses must be a single HEADERS frame (#3152)

### Motivation

When a blocking gRPC endpoint with streaming response encounters an exception, it attempts to send `Trailers-Only` response (grpc-status is returned in headers rather than trailers). However, the HTTP/2 layer sends two frames instead of one: `HEADERS with endStream=false` and `DATA with endStream=true`. Official grpc-java implementation doesn't like that, it wants a single `HEADERS` frame with `endStream=true` for `Trailers-Only` responses.

### Changes

I added a filter and a context key to signal when we have a case where `Trailers-Only` response is sent. The filter will aggregate the response stream, observe there are no trailers, drop them, and will only return a single `HEADERS` frame with `endStream=true`.
This behavior can be tested by `ProtocolCompatibilityTest.unimplementedServiceError`, we unskipped tests that were failing before this fix.

### Results

Before Change:

```
OUTBOUND HEADERS: streamId=3 headers=GrpcHttp2OutboundHeaders[:authority: 127.0.0.1:62451, :path: /grpc.netty.Compat/clientStreamingCall, :method: POST, :scheme: http, content-type: application/grpc, te: trailers, user-agent: grpc-java-netty/1.64.1, grpc-accept-encoding: gzip] streamDependency=0 weight=16 exclusive=false padding=0 endStream=false
OUTBOUND DATA: streamId=3 padding=0 endStream=false length=7 bytes=00000000020801
OUTBOUND DATA: streamId=3 padding=0 endStream=true length=0 bytes=
INBOUND HEADERS: streamId=3 headers=GrpcHttp2ResponseHeaders[:status: 200, server: servicetalk-grpc/, content-type: application/grpc+proto, grpc-status: 12, grpc-message: Method grpc.netty.Compat/clientStreamingCall is unimplemented] padding=0 endStream=false
INBOUND DATA: streamId=3 padding=0 endStream=true length=0 bytes=
```

After Change:

```
OUTBOUND HEADERS: streamId=3 headers=GrpcHttp2OutboundHeaders[:authority: 127.0.0.1:62421, :path: /grpc.netty.Compat/clientStreamingCall, :method: POST, :scheme: http, content-type: application/grpc, te: trailers, user-agent: grpc-java-netty/1.64.1, grpc-accept-encoding: gzip] streamDependency=0 weight=16 exclusive=false padding=0 endStream=false
OUTBOUND DATA: streamId=3 padding=0 endStream=false length=7 bytes=00000000020801
OUTBOUND DATA: streamId=3 padding=0 endStream=true length=0 bytes=
INBOUND HEADERS: streamId=3 headers=GrpcHttp2ResponseHeaders[:status: 200, server: servicetalk-grpc/, content-type: application/grpc+proto, grpc-status: 12, grpc-message: Method grpc.netty.Compat/clientStreamingCall is unimplemented, content-length: 0] padding=0 endStream=true
```

Author: mgodave

servicetalk-grpc-api/src/main/java/io/servicetalk/grpc/api/GrpcRouter.java

@@ -90,6 +90,7 @@
 import static io.servicetalk.grpc.api.GrpcUtils.setStatus;
 import static io.servicetalk.grpc.api.GrpcUtils.setStatusOk;
 import static io.servicetalk.grpc.api.GrpcUtils.validateContentType;
+import static io.servicetalk.grpc.internal.GrpcContextKeys.TRAILERS_ONLY_RESPONSE;
 import static io.servicetalk.http.api.HttpApiConversions.toStreamingHttpService;
 import static io.servicetalk.http.api.HttpExecutionStrategies.customStrategyBuilder;
 import static io.servicetalk.http.api.HttpExecutionStrategies.defaultStrategy;
@@ -708,6 +709,7 @@ public void handle(final HttpServiceContext ctx, final BlockingStreamingHttpRequ
                                         methodDescriptor.httpPath(), t);
                                 HttpHeaders trailers;
                                 if (grpcResponse == null || (trailers = grpcResponse.trailers()) == null) {
+                                    response.context().put(TRAILERS_ONLY_RESPONSE, Boolean.TRUE);
                                     setStatus(response.headers(), t, allocator);
                                     // Use HTTP response to avoid setting "OK" in trailers and allocating a serializer
                                     response.sendMetaData().close();

servicetalk-grpc-internal/src/main/java/io/servicetalk/grpc/internal/GrpcContextKeys.java

@@ -0,0 +1,36 @@
+/*
+ * Copyright © 2019, 2021 Apple Inc. and the ServiceTalk project authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.servicetalk.grpc.internal;
+
+import io.servicetalk.context.api.ContextMap;
+
+import static io.servicetalk.context.api.ContextMap.Key.newKey;
+
+/**
+ * All {@link ContextMap.Key}(s) defined for gRPC.
+ */
+public final class GrpcContextKeys {
+    /**
+     * For the blocking server this key allows the router to notify an upstream filter that it is safe to consolidate
+     * tailing empty data frames when set to true.
+     *
+     */
+    public static final ContextMap.Key<Boolean> TRAILERS_ONLY_RESPONSE =
+            newKey("TRAILERS_ONLY_RESPONSE", Boolean.class);
+
+    private GrpcContextKeys() {
+    }
+}

servicetalk-grpc-netty/src/main/java/io/servicetalk/grpc/netty/DefaultGrpcServerBuilder.java

@@ -145,7 +145,6 @@ private Single<GrpcServerContext> doListen(final GrpcServiceFactory<?> serviceFa
     private ExecutionContextInterceptorHttpServerBuilder preBuild() {
         final ExecutionContextInterceptorHttpServerBuilder interceptor =
                 new ExecutionContextInterceptorHttpServerBuilder(httpServerBuilderSupplier.get());
-
         interceptor.appendNonOffloadingServiceFilter(GrpcExceptionMapperServiceFilter.INSTANCE);
 
         directCallInitializer.initialize(interceptor);
@@ -154,6 +153,8 @@ private ExecutionContextInterceptorHttpServerBuilder preBuild() {
         }
         initializer.initialize(interceptor);
 
+        interceptor.appendServiceFilter(GrpcEnforceTrailersOnlyResponseServiceFilter.INSTANCE);
+
         return interceptor;
     }
 

servicetalk-grpc-netty/src/main/java/io/servicetalk/grpc/netty/GrpcEnforceTrailersOnlyResponseServiceFilter.java

@@ -0,0 +1,63 @@
+/*
+ * Copyright © 2019, 2021 Apple Inc. and the ServiceTalk project authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.servicetalk.grpc.netty;
+
+import io.servicetalk.concurrent.api.Single;
+import io.servicetalk.http.api.HttpExecutionStrategies;
+import io.servicetalk.http.api.HttpExecutionStrategy;
+import io.servicetalk.http.api.HttpResponse;
+import io.servicetalk.http.api.HttpServiceContext;
+import io.servicetalk.http.api.StreamingHttpRequest;
+import io.servicetalk.http.api.StreamingHttpResponse;
+import io.servicetalk.http.api.StreamingHttpResponseFactory;
+import io.servicetalk.http.api.StreamingHttpService;
+import io.servicetalk.http.api.StreamingHttpServiceFilter;
+import io.servicetalk.http.api.StreamingHttpServiceFilterFactory;
+
+import static io.servicetalk.grpc.internal.GrpcContextKeys.TRAILERS_ONLY_RESPONSE;
+
+final class GrpcEnforceTrailersOnlyResponseServiceFilter implements StreamingHttpServiceFilterFactory {
+    static final GrpcEnforceTrailersOnlyResponseServiceFilter INSTANCE =
+            new GrpcEnforceTrailersOnlyResponseServiceFilter();
+
+    private GrpcEnforceTrailersOnlyResponseServiceFilter() {
+    }
+
+    @Override
+    public StreamingHttpServiceFilter create(StreamingHttpService service) {
+        return new StreamingHttpServiceFilter(service) {
+            @Override
+            public Single<StreamingHttpResponse> handle(final HttpServiceContext ctx,
+                                                        final StreamingHttpRequest request,
+                                                        final StreamingHttpResponseFactory responseFactory) {
+                return delegate().handle(ctx, request, responseFactory).flatMap(response -> {
+                    Single<StreamingHttpResponse> mappedResponse;
+                    if (Boolean.TRUE.equals(response.context().get(TRAILERS_ONLY_RESPONSE))) {
+                        mappedResponse = response.toResponse().map(HttpResponse::toStreamingResponse);
+                    } else {
+                        mappedResponse = Single.succeeded(response);
+                    }
+                    return mappedResponse.shareContextOnSubscribe();
+                });
+            }
+        };
+    }
+
+    @Override
+    public HttpExecutionStrategy requiredOffloads() {
+        return HttpExecutionStrategies.offloadNone();
+    }
+}

servicetalk-grpc-netty/src/test/java/io/servicetalk/grpc/netty/ProtocolCompatibilityTest.java

@@ -268,10 +268,6 @@ private static Collection<Arguments> clientServerParams() {
         for (boolean isClientServiceTalk : TRUE_FALSE) {
             for (boolean isServerServiceTalk : TRUE_FALSE) {
                 for (boolean isServerBlocking : TRUE_FALSE) {
-                    if (!isClientServiceTalk && isServerServiceTalk && isServerBlocking) {
-                        // TODO there appears to be a potential bug in this combination. Separate bug filed.
-                        continue;
-                    }
                     if (isServerServiceTalk || !isServerBlocking) {
                         args.add(Arguments.of(isClientServiceTalk, isServerServiceTalk, isServerBlocking));
                     }