#20223

Committed 26 Mar 2026 02:56AM UTC coverage: 88.688% (-0.03%) from 88.715%

Build # #20223

Build Type

push

github

Committed by

ejona86

Commit Message

core: Close InputStream in NoopClientStream.writeMessage() to prevent resource leaks

NoopClientStream.writeMessage() silently discards the InputStream without
closing it. When a Marshaller.stream() returns an InputStream backed by a
ref-counted ByteBuf (e.g. Netty's PooledByteBufAllocator), this causes a
direct memory leak.

This affects any code path where writeMessage() is called on a
NoopClientStream or its subclass FailingClientStream:
- Context cancelled before stream start (ClientCallImpl line 197)
- Compressor not found (ClientCallImpl line 219)
- Deadline already exceeded (ClientCallImpl line 262, via FailingClientStream)
- DelayedStream draining buffered messages after cancellation sets
  realStream to NoopClientStream.INSTANCE

The fix calls GrpcUtil.closeQuietly(message) to ensure the InputStream is
always closed, matching the contract in AbstractStream.writeMessage().

Run Details

35491 of 40018 relevant lines covered (88.69%)

0.89 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

68.18

/../core/src/main/java/io/grpc/internal/NoopClientStream.java

/*
 * Copyright 2015 The gRPC 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.grpc.internal;

import io.grpc.Attributes;
import io.grpc.Compressor;
import io.grpc.Deadline;
import io.grpc.DecompressorRegistry;
import io.grpc.Status;
import java.io.InputStream;
import javax.annotation.Nonnull;

/**
 * An implementation of {@link ClientStream} that silently does nothing for the operations.
 */
public class NoopClientStream implements ClientStream {
  public static final NoopClientStream INSTANCE = new NoopClientStream();

  @Override
  public void setAuthority(String authority) {}

  @Override
  public void start(ClientStreamListener listener) {}

  @Override
  public Attributes getAttributes() {
    return Attributes.EMPTY;
  }

  @Override
  public void request(int numMessages) {}

  @Override
  public void writeMessage(InputStream message) {
    GrpcUtil.closeQuietly(message);
  }

  @Override
  public void flush() {}

  @Override
  public boolean isReady() {
    return false;
  }

  @Override
  public void cancel(Status status) {}

  @Override
  public void halfClose() {}

  @Override
  public void setMessageCompression(boolean enable) {
    // noop
  }

  @Override
  public void optimizeForDirectExecutor() {}

  @Override
  public void setCompressor(Compressor compressor) {}

  @Override
  public void setFullStreamDecompression(boolean fullStreamDecompression) {}

  @Override
  public void setDecompressorRegistry(DecompressorRegistry decompressorRegistry) {}

  @Override
  public void setMaxInboundMessageSize(int maxSize) {}

  @Override
  public void setMaxOutboundMessageSize(int maxSize) {}

  @Override
  public void setDeadline(@Nonnull Deadline deadline) {}

  @Override
  public void appendTimeoutInsight(InsightBuilder insight) {
    insight.append("noop");
  }
}

1	/*
2	* Copyright 2015 The gRPC Authors
3	*
4	* Licensed under the Apache License, Version 2.0 (the "License");
5	* you may not use this file except in compliance with the License.
6	* You may obtain a copy of the License at
7	*
8	* http://www.apache.org/licenses/LICENSE-2.0
9	*
10	* Unless required by applicable law or agreed to in writing, software
11	* distributed under the License is distributed on an "AS IS" BASIS,
12	* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13	* See the License for the specific language governing permissions and
14	* limitations under the License.
15	*/
16
17	package io.grpc.internal;
18
19	import io.grpc.Attributes;
20	import io.grpc.Compressor;
21	import io.grpc.Deadline;
22	import io.grpc.DecompressorRegistry;
23	import io.grpc.Status;
24	import java.io.InputStream;
25	import javax.annotation.Nonnull;
26
27	/**
28	* An implementation of {@link ClientStream} that silently does nothing for the operations.
29	*/
30	public class NoopClientStream implements ClientStream {	1✔
31	public static final NoopClientStream INSTANCE = new NoopClientStream();	1✔
32
33	@Override
34	public void setAuthority(String authority) {}	×
35
36	@Override
37	public void start(ClientStreamListener listener) {}	1✔
38
39	@Override
40	public Attributes getAttributes() {
41	return Attributes.EMPTY;	×
42	}
43
44	@Override
45	public void request(int numMessages) {}	1✔
46
47	@Override
48	public void writeMessage(InputStream message) {
49	GrpcUtil.closeQuietly(message);	1✔
50	}	1✔
51
52	@Override
53	public void flush() {}	1✔
54
55	@Override
56	public boolean isReady() {
57	return false;	1✔
58	}
59
60	@Override
61	public void cancel(Status status) {}	1✔
62
63	@Override
64	public void halfClose() {}	1✔
65
66	@Override
67	public void setMessageCompression(boolean enable) {
68	// noop
69	}	1✔
70
71	@Override
72	public void optimizeForDirectExecutor() {}	1✔
73
74	@Override
75	public void setCompressor(Compressor compressor) {}	1✔
76
77	@Override
78	public void setFullStreamDecompression(boolean fullStreamDecompression) {}	×
79
80	@Override
81	public void setDecompressorRegistry(DecompressorRegistry decompressorRegistry) {}	1✔
82
83	@Override
84	public void setMaxInboundMessageSize(int maxSize) {}	×
85
86	@Override
87	public void setMaxOutboundMessageSize(int maxSize) {}	×
88
89	@Override
90	public void setDeadline(@Nonnull Deadline deadline) {}	1✔
91
92	@Override
93	public void appendTimeoutInsight(InsightBuilder insight) {
94	insight.append("noop");	×
95	}	×
96	}

grpc / grpc-java / #20223

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous