Repository: yarpc/yab
Pull request #342 (2021-05-18 16:43:26)

Yab Release v0.20.0


Patch:
diff --git a/CHANGELOG.md b/CHANGELOG.md
index 4fa08b7..0e66e8d 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -1,6 +1,10 @@
 Changelog
 =========
 
+# 0.20.0 (2021-05-18)
+* Add `stream-delay-close-send` option which delays client send stream closure.
+* New: gRPC details are now printed along with the error if there are any.
+
 # 0.19.1 (2021-04-02)
 * Fix byte parsing to allow 8-bit signed integers to match the Thrift spec & other language implementations.
 
diff --git a/README.md b/README.md
index 76724c0..f955925 100644
--- a/README.md
+++ b/README.md
@@ -1,7 +1,7 @@
 # yab [![Build Status][ci-img]][ci] [![Coverage Status][cov-img]][cov]
 
 `yab` (Yet Another Benchmarker) is a tool to benchmark YARPC services. It
-currently supports making Thrift requests to both HTTP and TChannel services.
+currently supports making Thrift requests to both HTTP and [TChannel](https://github.com/uber/tchannel) services, as well as Protobuf requests to [gRPC](https://grpc.io/) services.
 
 `yab` is currently in **beta** status.
 
@@ -73,7 +73,11 @@ Request Options:
       --multiplexed-thrift       Enables the Thrift TMultiplexedProtocol used
                                  by services that host multiple Thrift services
                                  on a single endpoint.
-      --stream-interval=         Interval between consecutive stream message sends, applicable separately to every stream request opened on a connection.
+      --stream-interval=         Interval between consecutive stream message sends,
+                                 applicable separately to every stream request
+                                 opened on a connection.
+      --stream-delay-close-send= Delay the closure of send stream once all the
+                                 request messages have been sent.
 
 Transport Options:
   -s, --service=                 The TChannel/Hyperbahn service name
diff --git a/encoding/encoding.go b/encoding/encoding.go
index 571cdbe..086e624 100644
--- a/encoding/encoding.go
+++ b/encoding/encoding.go
@@ -69,6 +69,13 @@ type Serializer interface {
 	MethodType() MethodType
 }
 
+// ProtoErrorDeserializer deserializes errors.
+// Error details are very specific to proto.
+type ProtoErrorDeserializer interface {
+	// Error converts an error into something that can be displayed to a user.
+	ErrorDetails(err error) ([]interface{}, error)
+}
+
 // StreamRequestReader interface exposes method to read multiple request body
 type StreamRequestReader interface {
 	// NextBody returns the encoded request body if available, and if not, returns an
diff --git a/encoding/protobuf.go b/encoding/protobuf.go
index 556e886..2ac9dc3 100644
--- a/encoding/protobuf.go
+++ b/encoding/protobuf.go
@@ -17,6 +17,8 @@ import (
 	"github.com/jhump/protoreflect/desc"
 	"github.com/jhump/protoreflect/dynamic"
 	"go.uber.org/yarpc/pkg/procedure"
+	"go.uber.org/yarpc/yarpcerrors"
+	"google.golang.org/genproto/googleapis/rpc/status"
 )
 
 type protoSerializer struct {
@@ -106,6 +108,49 @@ func (p protoSerializer) Encoding() Encoding {
 	return Protobuf
 }
 
+func (p protoSerializer) ErrorDetails(err error) ([]interface{}, error) {
+	// Here we use yarpcerrors since the transport layer of yab is using yarpc as well
+	if !yarpcerrors.IsStatus(err) {
+		return nil, nil
+	}
+
+	yerr := yarpcerrors.FromError(err)
+	if len(yerr.Details()) == 0 {
+		return nil, nil
+	}
+
+	errStatus := &status.Status{}
+	if err := proto.Unmarshal(yerr.Details(), errStatus); err != nil {
+		return nil, fmt.Errorf("could not unmarshal error details %s", err.Error())
+	}
+
+	details := []interface{}{}
+	for _, detail := range errStatus.Details {
+		// By default we set to the value of the proto detail message to its byte values.
+		// It is possible that YAB will not be able to resolve the type message.
+		// This can happen when an error is being bubbled up in a chain of services.
+		// For instance, let's say we have A -> B -> C.
+		// If A does not have registered detail messages descriptors from C and B blindly bubbled up
+		// errors from C, YAB will not be able to resolve the type of the details based
+		// on the descriptors given by A (through the reflection server).
+		var value interface{} = detail.Value
+
+		rdetail, rerr := p.anyResolver.Resolve(detail.TypeUrl)
+		if rerr == nil {
+			if err := proto.Unmarshal(detail.Value, rdetail); err != nil {
+				return nil, fmt.Errorf("could not unmarshal error detail message %s", err.Error())
+			}
+			value = rdetail
+		}
+
+		details = append(details, map[string]interface{}{
+			detail.TypeUrl: value,
+		})
+	}
+
+	return details, nil
+}
+
 func (p protoSerializer) MethodType() MethodType {
 	if p.method.IsClientStreaming() && p.method.IsServerStreaming() {
 		return BidirectionalStream
diff --git a/go.mod b/go.mod
index 8285e3a..bc5b4c1 100644
--- a/go.mod
+++ b/go.mod
@@ -43,6 +43,7 @@ require (
 	go.uber.org/zap v1.10.0
 	golang.org/x/net v0.0.0-20190926025831-c00fd9afed17
 	golang.org/x/text v0.3.1-0.20180511172408-5c1cf69b5978 // indirect
+	google.golang.org/genproto v0.0.0-20180817151627-c66870c02cf8 // indirect
 	google.golang.org/grpc v1.24.0
 	gopkg.in/yaml.v2 v2.2.2
 )
diff --git a/handler.go b/handler.go
index 6888023..61cdd56 100644
--- a/handler.go
+++ b/handler.go
@@ -212,7 +212,7 @@ func makeClientStream(ctx context.Context, stream *yarpctransport.ClientStream,
 	}
 
 	if err == io.EOF {
-		err = closeSendStream(ctx, stream)
+		err = closeSendStream(ctx, stream, opts.DelayCloseSendStream.Duration())
 	}
 	if err != nil {
 		return err
@@ -241,7 +241,7 @@ func makeBidiStream(ctx context.Context, cancel context.CancelFunc, stream *yarp
 			var reqBody []byte
 			reqBody, err = streamIO.NextRequest()
 			if err == io.EOF {
-				err = closeSendStream(ctx, stream)
+				err = closeSendStream(ctx, stream, opts.DelayCloseSendStream.Duration())
 				break
 			}
 			if err != nil {
@@ -317,8 +317,16 @@ func receiveStreamMessage(ctx context.Context, stream *yarpctransport.ClientStre
 }
 
 // closeSendStream closes the stream from the client side while
-// stream can continue to receive messages from server
-func closeSendStream(ctx context.Context, stream *yarpctransport.ClientStream) error {
+// stream can continue to receive messages from server. If non-zero delay is
+// passed, stream is closed after the delay.
+func closeSendStream(ctx context.Context, stream *yarpctransport.ClientStream, delayCloseSendStream time.Duration) error {
+	if delayCloseSendStream != 0 {
+		select {
+		case <-time.After(delayCloseSendStream):
+		case <-ctx.Done():
+		}
+	}
+
 	// YARPC stream.Close method internally invokes closeSend on gRPC clientStream.
 	if err := stream.Close(ctx); err != nil {
 		return fmt.Errorf("Failed to close send stream: %v", err)
diff --git a/main.go b/main.go
index 6df616c..58a0584 100644
--- a/main.go
+++ b/main.go
@@ -21,6 +21,7 @@
 package main
 
 import (
+	"bytes"
 	"context"
 	"encoding/json"
 	"errors"
@@ -472,7 +473,26 @@ func makeContextWithTrace(ctx context.Context, t transport.Transport, request *t
 func makeInitialRequest(out output, transport transport.Transport, serializer encoding.Serializer, req *transport.Request) {
 	response, err := makeRequestWithTracePriority(transport, req, 1)
 	if err != nil {
-		out.Fatalf("Failed while making call: %v\n", err)
+		buffer := bytes.NewBufferString(err.Error())
+
+		if errorSerializer, ok := serializer.(encoding.ProtoErrorDeserializer); ok {
+			details, derr := errorSerializer.ErrorDetails(err)
+			if derr != nil {
+				out.Fatalf("Failed to get protobuf error details %s", derr.Error())
+			}
+
+			if len(details) > 0 {
+				bs, merr := json.MarshalIndent(map[string]interface{}{"details": details}, "", "  ")
+				if merr != nil {
+					out.Fatalf("Failed to convert protobuf error details to JSON: %v\nMap: %+v\n", merr, details)
+				}
+				buffer.WriteString("\n")
+				buffer.Write(bs)
+				buffer.WriteString("\n")
+			}
+		}
+
+		out.Fatalf("Failed while making call: %s\n", buffer.String())
 	}
 
 	// responseMap converts the Thrift bytes response to a map.
diff --git a/options.go b/options.go
index 8513ad3..bb03194 100644
--- a/options.go
+++ b/options.go
@@ -76,7 +76,8 @@ type RequestOptions struct {
 
 // StreamRequestOptions are stream request related options
 type StreamRequestOptions struct {
-	Interval timeMillisFlag `long:"stream-interval" description:"Interval between consecutive stream message sends, applicable separately to every stream request opened on a connection."`
+	Interval             timeMillisFlag `long:"stream-interval" description:"Interval between consecutive stream message sends, applicable separately to every stream request opened on a connection."`
+	DelayCloseSendStream timeMillisFlag `long:"stream-delay-close-send" description:"Delay the closure of send stream once all the request messages have been sent."`
 }
 
 // TransportOptions are transport related options.
diff --git a/protobuf/source_reflection.go b/protobuf/source_reflection.go
index ea02445..dda4685 100644
--- a/protobuf/source_reflection.go
+++ b/protobuf/source_reflection.go
@@ -64,14 +64,17 @@ func NewDescriptorProviderReflection(args ReflectionArgs) (DescriptorProvider, e
 		routingHeaders.Append(ygrpc.RoutingKeyHeader, args.RoutingKey)
 	}
 
-	metadataContext := metadata.NewOutgoingContext(context.Background(), routingHeaders)
+	ctx, cancel := context.WithTimeout(context.Background(), args.Timeout)
+	metadataContext := metadata.NewOutgoingContext(ctx, routingHeaders)
 	return &grpcreflectSource{
-		client: grpcreflect.NewClient(metadataContext, pbClient),
+		client:     grpcreflect.NewClient(metadataContext, pbClient),
+		cancelFunc: cancel,
 	}, nil
 }
 
 type grpcreflectSource struct {
-	client *grpcreflect.Client
+	client     *grpcreflect.Client
+	cancelFunc context.CancelFunc
 }
 
 func (s *grpcreflectSource) FindMessage(messageType string) (*desc.MessageDescriptor, error) {
@@ -83,19 +86,23 @@ func (s *grpcreflectSource) FindMessage(messageType string) (*desc.MessageDescri
 		return nil, nil
 	}
 
+	if err != nil {
+		return nil, wrapReflectionError(err)
+	}
+
 	return msg, err
 }
 
 func (s *grpcreflectSource) FindService(fullyQualifiedName string) (*desc.ServiceDescriptor, error) {
 	service, err := s.client.ResolveService(fullyQualifiedName)
 	if err != nil {
-		available, availableErr := s.client.ListServices()
-		if availableErr != nil {
-			err = availableErr
+		if !grpcreflect.IsElementNotFoundError(err) {
+			return nil, wrapReflectionError(err)
 		}
 
-		if !grpcreflect.IsElementNotFoundError(err) {
-			return nil, err
+		available, availableErr := s.client.ListServices()
+		if availableErr != nil && !grpcreflect.IsElementNotFoundError(availableErr) {
+			return nil, wrapReflectionError(availableErr)
 		}
 
 		return nil, encodingerror.NotFound{
@@ -111,5 +118,10 @@ func (s *grpcreflectSource) FindService(fullyQualifiedName string) (*desc.Servic
 }
 
 func (s *grpcreflectSource) Close() {
+	s.cancelFunc()
 	s.client.Reset()
 }
+
+func wrapReflectionError(err error) error {
+	return fmt.Errorf("error in protobuf reflection: %v", err)
+}
diff --git a/version.go b/version.go
index b129020..b2088fe 100644
--- a/version.go
+++ b/version.go
@@ -22,4 +22,4 @@ package main
 
 // versionString is the sem-ver version string for yab.
 // It will be bumped explicitly on releases.
-var versionString = "0.19.1"
+var versionString = "0.20.0"


Test patch:
diff --git a/encoding/protobuf_test.go b/encoding/protobuf_test.go
index 0dbb55b..67c4ce9 100644
--- a/encoding/protobuf_test.go
+++ b/encoding/protobuf_test.go
@@ -19,6 +19,8 @@ import (
 	"github.com/jhump/protoreflect/dynamic"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
+	"go.uber.org/yarpc/yarpcerrors"
+	"google.golang.org/genproto/googleapis/rpc/status"
 )
 
 func TestNewProtobuf(t *testing.T) {
@@ -159,6 +161,138 @@ func TestProtobufRequest(t *testing.T) {
 	}
 }
 
+func TestProtobufErrorDetails(t *testing.T) {
+	source, err := protobuf.NewDescriptorProviderFileDescriptorSetBins("../testdata/protobuf/simple/simple.proto.bin")
+	require.NoError(t, err)
+	anySource, err := protobuf.NewDescriptorProviderFileDescriptorSetBins("../testdata/protobuf/any/any.proto.bin")
+	require.NoError(t, err)
+
+	grpcStatusBytesFunc := func(typeUrl string, msg proto.Message) []byte {
+		valueContent, err := proto.Marshal(msg)
+		require.NoError(t, err)
+
+		a := &any.Any{
+			TypeUrl: typeUrl,
+			Value:   valueContent,
+		}
+
+		s := &status.Status{}
+		s.Details = []*any.Any{a}
+		bytes, err := proto.Marshal(s)
+		require.NoError(t, err)
+		return bytes
+	}
+
+	grpcStatusCorrupedBytesFunc := func(typeUrl string, msg proto.Message) []byte {
+		valueContent, err := proto.Marshal(msg)
+		require.NoError(t, err)
+
+		a := &any.Any{
+			TypeUrl: typeUrl,
+			Value:   valueContent[:1],
+		}
+
+		s := &status.Status{}
+		s.Details = []*any.Any{a}
+		bytes, err := proto.Marshal(s)
+		require.NoError(t, err)
+		return bytes
+	}
+
+	tests := []struct {
+		desc          string
+		method        string
+		source        protobuf.DescriptorProvider
+		err           error
+		wantOutAsJSON string
+		wantErr       error
+	}{
+		{
+			desc:          "nil error",
+			source:        source,
+			method:        "Bar/Baz",
+			err:           nil,
+			wantErr:       nil,
+			wantOutAsJSON: "null",
+		},
+		{
+			desc:          "std error",
+			source:        source,
+			method:        "Bar/Baz",
+			err:           fmt.Errorf("this is a test error"),
+			wantErr:       nil,
+			wantOutAsJSON: "null",
+		},
+		{
+			desc:          "yarpc error with no details",
+			source:        source,
+			method:        "Bar/Baz",
+			err:           yarpcerrors.FromError(fmt.Errorf("this is a test error")),
+			wantErr:       nil,
+			wantOutAsJSON: "null",
+		},
+		{
+			desc:          "yarpc error bad details bytes",
+			source:        source,
+			method:        "Bar/Baz",
+			err:           yarpcerrors.FromError(fmt.Errorf("this is a test error")).WithDetails([]byte{0x8, 0x1, 0x12}),
+			wantErr:       fmt.Errorf("could not unmarshal error details unexpected EOF"),
+			wantOutAsJSON: "null",
+		},
+		{
+			desc:   "yarpc error with  1 detail",
+			source: anySource,
+			method: "BarAny/BazAny",
+			err: yarpcerrors.FromError(fmt.Errorf("this is a test error")).WithDetails(grpcStatusBytesFunc("type.googleapis.com/FooAny", &tany.FooAny{
+				Value: 10,
+			})),
+			wantErr:       nil,
+			wantOutAsJSON: `[{"type.googleapis.com/FooAny":{"value":10}}]`,
+		},
+		{
+			desc:   "yarpc error with detail which can not be resolved",
+			source: anySource,
+			method: "BarAny/BazAny",
+			err: yarpcerrors.FromError(fmt.Errorf("this is a test error")).WithDetails(grpcStatusBytesFunc("uri/not/identifiable", &tany.FooAny{
+				Value: 10,
+			})),
+			wantErr:       nil,
+			wantOutAsJSON: `[{"uri/not/identifiable":{"V":"CAo="}}]`,
+		},
+		{
+			desc:   "yarpc error with detail with corrupted bytes in any detail message",
+			source: anySource,
+			method: "BarAny/BazAny",
+			err: yarpcerrors.FromError(fmt.Errorf("this is a test error")).WithDetails(grpcStatusCorrupedBytesFunc("type.googleapis.com/FooAny", &tany.FooAny{
+				Value: 10,
+			})),
+			wantErr:       fmt.Errorf("could not unmarshal error detail message unexpected EOF"),
+			wantOutAsJSON: ``,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.desc, func(t *testing.T) {
+			serializer, err := NewProtobuf(tt.method, tt.source)
+			require.NoError(t, err, "Failed to create serializer")
+			errorSerializer, ok := serializer.(ProtoErrorDeserializer)
+			require.True(t, ok)
+
+			result, err := errorSerializer.ErrorDetails(tt.err)
+
+			if tt.wantErr != nil {
+				require.Error(t, err, tt.desc)
+				assert.Contains(t, err.Error(), tt.wantErr.Error(), "%v: invalid error", tt.desc)
+			} else {
+				require.NoError(t, err, tt.desc)
+				r, err := json.Marshal(result)
+				require.NoError(t, err)
+				assert.Equal(t, tt.wantOutAsJSON, string(r))
+			}
+		})
+	}
+}
+
 func TestProtobufResponse(t *testing.T) {
 	source, err := protobuf.NewDescriptorProviderFileDescriptorSetBins("../testdata/protobuf/simple/simple.proto.bin")
 	require.NoError(t, err)
diff --git a/handler_test.go b/handler_test.go
index 79240ad..ed146ce 100644
--- a/handler_test.go
+++ b/handler_test.go
@@ -31,6 +31,7 @@ import (
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 	"github.com/yarpc/yab/encoding"
+	"go.uber.org/yarpc/api/transport"
 )
 
 type mockStreamReader struct {
@@ -152,3 +153,70 @@ func TestIntervalWaiter(t *testing.T) {
 		}
 	})
 }
+
+type mockStreamCloser struct {
+	closeErr error
+}
+
+func (m mockStreamCloser) Close(context.Context) error { return m.closeErr }
+
+func (mockStreamCloser) Context() context.Context { return context.Background() }
+
+func (mockStreamCloser) Request() *transport.StreamRequest { return nil }
+
+func (mockStreamCloser) SendMessage(context.Context, *transport.StreamMessage) error { return nil }
+
+func (mockStreamCloser) ReceiveMessage(context.Context) (*transport.StreamMessage, error) {
+	return nil, nil
+}
+
+func TestCloseSendStream(t *testing.T) {
+	tests := []struct {
+		description string
+		delay       time.Duration
+		err         error
+	}{
+		{
+			description: "close without delay",
+		},
+		{
+			description: "close with error",
+			err:         errors.New("test error"),
+		},
+		{
+			description: "close with delay",
+			delay:       time.Millisecond * 50,
+		},
+	}
+
+	for _, test := range tests {
+		t.Run(test.description, func(t *testing.T) {
+			stream, err := transport.NewClientStream(mockStreamCloser{closeErr: test.err})
+			require.NoError(t, err, "unexpected client stream error")
+
+			start := time.Now()
+			err = closeSendStream(context.Background(), stream, test.delay)
+			end := time.Now()
+
+			if test.err != nil {
+				require.Error(t, err, "unexpected nil error from close")
+				assert.Contains(t, err.Error(), test.err.Error(), "unexpected error message from close")
+			} else {
+				assert.NoError(t, err, "unexpected error from close")
+			}
+
+			assert.True(t, end.After(start.Add(test.delay)), "unexpected execution time")
+		})
+	}
+
+	t.Run("must not delay when context is done", func(t *testing.T) {
+		stream, err := transport.NewClientStream(mockStreamCloser{})
+		require.NoError(t, err, "unexpected client stream error")
+
+		ctx, cancel := context.WithCancel(context.Background())
+		cancel()
+		start := time.Now()
+		require.NoError(t, closeSendStream(ctx, stream, time.Second), "unexpected error from close")
+		assert.True(t, time.Now().Before(start.Add(time.Second)), "expected close to execute immediately")
+	})
+}
diff --git a/integration_test.go b/integration_test.go
index 3c85067..2ce1dba 100644
--- a/integration_test.go
+++ b/integration_test.go
@@ -55,6 +55,9 @@ import (
 	ytchan "go.uber.org/yarpc/transport/tchannel"
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/reflection"
+	rpb "google.golang.org/grpc/reflection/grpc_reflection_v1alpha"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/status"
 )
 
 //go:generate thriftrw --plugin=yarpc --out ./testdata/yarpc ./testdata/integration.thrift
@@ -378,6 +381,18 @@ func (s *simpleService) Baz(c context.Context, in *simple.Foo) (*simple.Foo, err
 		return in, nil
 	}
 
+	if in.Test == -1 {
+		st := status.New(codes.InvalidArgument, "invalid username")
+		st, err := st.WithDetails(in)
+		if err != nil {
+			// If this errored, it will always error
+			// here, so better panic so we can figure
+			// out why than have this silently passing.
+			panic(fmt.Sprintf("Unexpected error: %v", err))
+		}
+		return nil, st.Err()
+	}
+
 	return nil, fmt.Errorf("negative input")
 }
 
@@ -1001,72 +1016,113 @@ test: 1
 	}
 }
 
-func TestGRPCStreamWithStreamIntervalOption(t *testing.T) {
-	t.Run("client stream", func(t *testing.T) {
-		addr, server := setupGRPCServer(t, &simpleService{
+func TestGRPCStreamWithBoundedExecutionTime(t *testing.T) {
+	tests := []struct {
+		desc         string
+		opts         Options
+		wantExecTime time.Duration
+
+		returnOutput  []simple.Foo
+		expectedInput []simple.Foo
+	}{
+		{
+			desc: "client side streaming with stream interval option",
+			opts: Options{
+				ROpts: RequestOptions{
+					FileDescriptorSet: []string{"testdata/protobuf/simple/simple.proto.bin"},
+					Procedure:         "Bar/ClientStream",
+					Timeout:           timeMillisFlag(time.Second * 5),
+					RequestJSON:       `{"test":1}{"test":2}{"test":-1}`,
+					StreamRequestOptions: StreamRequestOptions{
+						Interval: timeMillisFlag(time.Millisecond * 100),
+					},
+				},
+			},
 			returnOutput:  []simple.Foo{{Test: 4}},
 			expectedInput: []simple.Foo{{Test: 1}, {Test: 2}, {Test: -1}},
-		})
-		defer server.Stop()
-
-		opts := Options{
-			ROpts: RequestOptions{
-				FileDescriptorSet: []string{"testdata/protobuf/simple/simple.proto.bin"},
-				Procedure:         "Bar/ClientStream",
-				Timeout:           timeMillisFlag(time.Second * 5),
-				RequestJSON:       `{"test":1}{"test":2}{"test":-1}`,
-				StreamRequestOptions: StreamRequestOptions{
-					Interval: timeMillisFlag(time.Millisecond * 1000),
+			// Test must run for more than 200ms as there are three requests and
+			// it must take 200ms totally between consecutive messages.
+			wantExecTime: time.Millisecond * 200,
+		},
+		{
+			desc: "bidirectional streaming with stream interval option",
+			opts: Options{
+				ROpts: RequestOptions{
+					FileDescriptorSet: []string{"testdata/protobuf/simple/simple.proto.bin"},
+					Procedure:         "Bar/BidiStream",
+					Timeout:           timeMillisFlag(time.Second * 5),
+					RequestJSON:       `{"test":1}{"test":2}{"test":-1}`,
+					StreamRequestOptions: StreamRequestOptions{
+						Interval: timeMillisFlag(time.Millisecond * 100),
+					},
 				},
 			},
-			TOpts: TransportOptions{
-				ServiceName: "foo",
-				Peers:       []string{"grpc://" + addr.String()},
-			},
-		}
-
-		now := time.Now()
-		_, gotErr := runTestWithOpts(opts)
-		duration := time.Now().Sub(now)
-
-		assert.Empty(t, gotErr)
-		// test must run for more than 2 seconds as there are three requests and
-		// it must take 2 seconds totally between consecutive messages
-		assert.True(t, duration > (time.Second*2), "Unexpected time taken on wait: %v", duration)
-	})
-
-	t.Run("bidirectional stream", func(t *testing.T) {
-		addr, server := setupGRPCServer(t, &simpleService{
 			returnOutput:  []simple.Foo{{Test: 4}, {Test: 9}},
 			expectedInput: []simple.Foo{{Test: 1}, {Test: 2}, {Test: -1}},
-		})
-		defer server.Stop()
-
-		opts := Options{
-			ROpts: RequestOptions{
-				FileDescriptorSet: []string{"testdata/protobuf/simple/simple.proto.bin"},
-				Procedure:         "Bar/BidiStream",
-				Timeout:           timeMillisFlag(time.Second * 5),
-				RequestJSON:       `{"test":1}{"test":2}{"test":-1}`,
-				StreamRequestOptions: StreamRequestOptions{
-					Interval: timeMillisFlag(time.Millisecond * 1000),
+			// Test must run for more than 200ms as there are three requests and
+			// it must take 200ms totally between consecutive messages.
+			wantExecTime: time.Millisecond * 200,
+		},
+		{
+			desc: "client side streaming with close send stream delay",
+			opts: Options{
+				ROpts: RequestOptions{
+					FileDescriptorSet: []string{"testdata/protobuf/simple/simple.proto.bin"},
+					Procedure:         "Bar/ClientStream",
+					Timeout:           timeMillisFlag(time.Second * 5),
+					RequestJSON:       `{"test":1}{"test":2}{"test":-1}`,
+					StreamRequestOptions: StreamRequestOptions{
+						DelayCloseSendStream: timeMillisFlag(time.Millisecond * 100),
+					},
 				},
 			},
-			TOpts: TransportOptions{
-				ServiceName: "foo",
-				Peers:       []string{"grpc://" + addr.String()},
+			returnOutput:  []simple.Foo{{Test: 4}},
+			expectedInput: []simple.Foo{{Test: 1}, {Test: 2}, {Test: -1}},
+			// Test must run for more than 100ms as send stream closure has a delay
+			// of 100ms.
+			wantExecTime: time.Millisecond * 100,
+		},
+		{
+			desc: "bidirectional streaming with close send stream delay",
+			opts: Options{
+				ROpts: RequestOptions{
+					FileDescriptorSet: []string{"testdata/protobuf/simple/simple.proto.bin"},
+					Procedure:         "Bar/BidiStream",
+					Timeout:           timeMillisFlag(time.Second * 5),
+					RequestJSON:       `{"test":1}{"test":2}`,
+					StreamRequestOptions: StreamRequestOptions{
+						DelayCloseSendStream: timeMillisFlag(time.Millisecond * 100),
+					},
+				},
 			},
-		}
+			returnOutput:  []simple.Foo{{Test: 4}, {Test: 9}},
+			expectedInput: []simple.Foo{{Test: 1}, {Test: 2}, {Test: -1}},
+			// Test must run for more than 100ms as send stream closure has a delay
+			// of 100ms.
+			wantExecTime: time.Millisecond * 100,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.desc, func(t *testing.T) {
+			addr, server := setupGRPCServer(t, &simpleService{
+				expectedInput: tt.expectedInput,
+				returnOutput:  tt.returnOutput,
+			})
+			defer server.Stop()
 
-		now := time.Now()
-		_, gotErr := runTestWithOpts(opts)
-		duration := time.Now().Sub(now)
+			tt.opts.TOpts = TransportOptions{
+				ServiceName: "foo",
+				Peers:       []string{"grpc://" + addr.String()},
+			}
+			now := time.Now()
+			_, gotErr := runTestWithOpts(tt.opts)
+			duration := time.Now().Sub(now)
 
-		assert.Empty(t, gotErr)
-		// test must run for more than 2 seconds as there are three requests and
-		// it must take 2 seconds totally between consecutive messages
-		assert.True(t, duration > (time.Second*2), "Unexpected time taken on wait: %v", duration)
-	})
+			assert.Empty(t, gotErr)
+			assert.True(t, duration > tt.wantExecTime, "Unexpected time taken: %v", duration)
+		})
+	}
 }
 
 func TestGRPCReflectionSource(t *testing.T) {
@@ -1092,7 +1148,13 @@ func TestGRPCReflectionSource(t *testing.T) {
 					Peers:       []string{addr.String()},
 				},
 			},
-			wantRes: `"test": 1`,
+			wantRes: `{
+  "body": {
+    "test": 1
+  }
+}
+
+`,
 		},
 		{
 			desc: "success (no scheme in peer)",
@@ -1107,7 +1169,13 @@ func TestGRPCReflectionSource(t *testing.T) {
 					Peers:       []string{addr.String()},
 				},
 			},
-			wantRes: `"test": 1`,
+			wantRes: `{
+  "body": {
+    "test": 1
+  }
+}
+
+`,
 		},
 		{
 			desc: "return error",
@@ -1122,14 +1190,83 @@ func TestGRPCReflectionSource(t *testing.T) {
 					Peers:       []string{addr.String()},
 				},
 			},
-			wantErr: "negative input",
+			wantErr: "Failed while making call: code:unknown message:negative input\n",
+		},
+		{
+			desc: "return error with details",
+			opts: Options{
+				ROpts: RequestOptions{
+					Procedure:   "Bar/Baz",
+					Timeout:     timeMillisFlag(time.Second),
+					RequestJSON: `{"test":-1}`,
+				},
+				TOpts: TransportOptions{
+					ServiceName: "foo",
+					Peers:       []string{addr.String()},
+				},
+			},
+			wantErr: `Failed while making call: code:invalid-argument message:invalid username
+{
+  "details": [
+    {
+      "type.googleapis.com/Foo": {
+        "test": -1
+      }
+    }
+  ]
+}
+
+`,
 		},
 	}
 	for _, tt := range tests {
 		t.Run(tt.desc, func(t *testing.T) {
 			gotOut, gotErr := runTestWithOpts(tt.opts)
-			assert.Contains(t, gotErr, tt.wantErr)
-			assert.Contains(t, gotOut, tt.wantRes)
+			assert.Equal(t, gotErr, tt.wantErr)
+			assert.Equal(t, gotOut, tt.wantRes)
 		})
 	}
 }
+
+type protoReflectService struct {
+	waitChan chan struct{} // If non-nil, handler waits on the channel
+}
+
+func (p protoReflectService) ServerReflectionInfo(s rpb.ServerReflection_ServerReflectionInfoServer) error {
+	if p.waitChan != nil {
+		// Wait on the given channel or until stream context completes.
+		select {
+		case <-s.Context().Done():
+		case <-p.waitChan:
+		}
+	}
+	return nil
+}
+
+func TestGPCReflectionTimeout(t *testing.T) {
+	ln, err := net.Listen("tcp", "127.0.0.1:0")
+	require.NoError(t, err)
+
+	waitChan := make(chan struct{})
+	reflectSvc := protoReflectService{waitChan: waitChan}
+	server := grpc.NewServer()
+	rpb.RegisterServerReflectionServer(server, reflectSvc)
+	go server.Serve(ln)
+
+	defer server.Stop()
+	defer close(waitChan)
+
+	gotOut, gotErr := runTestWithOpts(Options{
+		ROpts: RequestOptions{
+			Procedure:   "Bar/Baz",
+			Timeout:     timeMillisFlag(time.Second * 1),
+			RequestJSON: `{"test":0}`,
+		},
+		TOpts: TransportOptions{
+			ServiceName: "foo",
+			Peers:       []string{ln.Addr().String()},
+		},
+	})
+	assert.Empty(t, gotOut, "Expected empty stdout")
+	assert.Contains(t, gotErr, "error in protobuf reflection: rpc error: code = DeadlineExceeded desc = context deadline exceeded", "Expected deadline exceeded error")
+}
diff --git a/protobuf/source_reflection_test.go b/protobuf/source_reflection_test.go
index a33097c..141be5d 100644
--- a/protobuf/source_reflection_test.go
+++ b/protobuf/source_reflection_test.go
@@ -1,6 +1,7 @@
 package protobuf
 
 import (
+	"errors"
 	"net"
 	"testing"
 	"time"
@@ -195,8 +196,42 @@ func TestReflectionRoutingHeaders(t *testing.T) {
 	}
 }
 
+func TestE2eErrors(t *testing.T) {
+	ln, err := net.Listen("tcp", "127.0.0.1:0")
+	require.NoError(t, err)
+	defer ln.Close()
+
+	s := grpc.NewServer()
+	drs := &dummyReflectionServer{returnErr: errors.New("test error")}
+	rpb.RegisterServerReflectionServer(s, drs)
+	go s.Serve(ln)
+
+	// Ensure that all streams are closed by the end of the test.
+	defer s.GracefulStop()
+
+	source, err := NewDescriptorProviderReflection(ReflectionArgs{
+		Timeout: time.Second,
+		Peers:   []string{ln.Addr().String()},
+	})
+	require.NoError(t, err)
+	defer source.Close()
+
+	t.Run("test error handling in find method", func(t *testing.T) {
+		desc, err := source.FindMessage("test")
+		require.Nil(t, desc, "unexpected message descriptor")
+		assert.EqualError(t, err, "error in protobuf reflection: rpc error: code = Unknown desc = test error")
+	})
+
+	t.Run("test error handling in find service", func(t *testing.T) {
+		desc, err := source.FindService("test")
+		require.Nil(t, desc, "unexpected message descriptor")
+		assert.EqualError(t, err, "error in protobuf reflection: rpc error: code = Unknown desc = test error")
+	})
+}
+
 type dummyReflectionServer struct {
-	md metadata.MD
+	md        metadata.MD
+	returnErr error
 }
 
 func (s *dummyReflectionServer) Reset() {
@@ -204,6 +239,10 @@ func (s *dummyReflectionServer) Reset() {
 }
 
 func (s *dummyReflectionServer) ServerReflectionInfo(r rpb.ServerReflection_ServerReflectionInfoServer) error {
+	if s.returnErr != nil {
+		return s.returnErr
+	}
+
 	if md, ok := metadata.FromIncomingContext(r.Context()); ok {
 		s.md = md
 	}