Memory leak on intercepting server streaming call

[azemliankinlivevision]

Version: 0.2.3

Steps to reproduce:

1. Intercept long-running server streaming calls with TracingServerInterceptor
2. Watch memory consumption increase indefinitely (take a heap dump every few minutes)

Description:
TracingServerInterceptor creates a new span on every call from the client and closes it only when onCancel or onComplete occurs. In the case of server streaming, it means that Span object lives as long as the client-server connection exists and accumulates all the LogData objects that happen in this period of time inside Span internal state.

In our particular case, we have about 400 connections, processing 1 streaming response per second on average. It leads to OOM error every 3-4 hours (having a memory limit of about 5Gb). As a workaround, we could turn off the tracing of server streaming calls:

@Override
public <ReqT, RespT> ServerCall.Listener<ReqT> interceptCall(ServerCall<ReqT, RespT> call, Metadata headers, ServerCallHandler<ReqT, RespT> next) {

    if (call.getMethodDescriptor().getType() == MethodDescriptor.MethodType.SERVER_STREAMING) {
        return noopServerCallListener();
    }
  ...

for the long-running server streaming calls, it may be no sense of tracing. So I believe it should be optional (controlled by some property)

Another option is to create and log a new span on server streaming response:

@Override
public void sendMessage(RespT message) {
    if (streaming || verbose) {
        span.log(
                ImmutableMap.<String, Object>builder()
                        .put(Fields.EVENT, GrpcFields.SERVER_CALL_SEND_MESSAGE)
                        .put(Fields.MESSAGE, "Server sent response message")
                        .build());
    }

    if (call.getMethodDescriptor().getType() == MethodDescriptor.MethodType.SERVER_STREAMING) {
        Span streamResponseSpan = getSpanFromHeaders(headerMap, operationNameConstructor.constructOperationName(call.getMethodDescriptor()));
        try (Scope ignored = tracer.scopeManager().activate(streamResponseSpan)) {
            super.sendMessage(message);
        } finally {
            streamResponseSpan.finish();
        }
    } else {
        try (Scope ignored = tracer.scopeManager().activate(span)) {
            super.sendMessage(message);
        }
    }
}

Both workarounds work for us. Please advise if we miss something or if you have a better solution.