#19996

Committed 24 Sep 2025 12:08AM UTC coverage: 88.575% (+0.03%) from 88.543%

Build # #19996

Build Type

push

github

Committed by

web-flow

Commit Message

Implement otel retry metrics (#12064)

implements [A96](https://github.com/grpc/proposal/pull/488/files)

Run Details

34731 of 39211 relevant lines covered (88.57%)

0.89 hits per line

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

86.11

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

/*
 * Copyright 2016 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 static com.google.common.base.Preconditions.checkNotNull;

import com.google.common.annotations.VisibleForTesting;
import io.grpc.CallOptions;
import io.grpc.Channel;
import io.grpc.ClientCall;
import io.grpc.ClientStreamTracer;
import io.grpc.Context;
import io.grpc.InternalConfigSelector;
import io.grpc.Metadata;
import io.grpc.MethodDescriptor;
import io.grpc.Status;
import io.grpc.internal.ClientCallImpl.ClientStreamProvider;
import io.grpc.internal.ClientStreamListener.RpcProgress;
import java.util.concurrent.Executor;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.atomic.AtomicReference;

final class SubchannelChannel extends Channel {
  @VisibleForTesting
  static final Status NOT_READY_ERROR =
      Status.UNAVAILABLE.withDescription("Subchannel is NOT READY");
  @VisibleForTesting
  static final Status WAIT_FOR_READY_ERROR =
      Status.UNAVAILABLE.withDescription(
          "wait-for-ready RPC is not supported on Subchannel.asChannel()");
  private static final FailingClientTransport notReadyTransport =
      new FailingClientTransport(NOT_READY_ERROR, RpcProgress.MISCARRIED);
  private final InternalSubchannel subchannel;
  private final Executor executor;
  private final ScheduledExecutorService deadlineCancellationExecutor;
  private final CallTracer callsTracer;
  private final AtomicReference<InternalConfigSelector> configSelector;

  private final ClientStreamProvider transportProvider = new ClientStreamProvider() {
      @Override
      public ClientStream newStream(MethodDescriptor<?, ?> method,
          CallOptions callOptions, Metadata headers, Context context) {
        ClientTransport transport = subchannel.getTransport();
        if (transport == null) {
          transport = notReadyTransport;
        }
        ClientStreamTracer[] tracers = GrpcUtil.getClientStreamTracers(
            callOptions, headers, 0, /* isTransparentRetry= */ false,
            /* isHedging= */ false);
        Context origContext = context.attach();
        try {
          return transport.newStream(method, headers, callOptions, tracers);
        } finally {
          context.detach(origContext);
        }
      }
    };

  SubchannelChannel(
      InternalSubchannel subchannel, Executor executor,
      ScheduledExecutorService deadlineCancellationExecutor, CallTracer callsTracer,
      AtomicReference<InternalConfigSelector> configSelector) {
    this.subchannel = checkNotNull(subchannel, "subchannel");
    this.executor = checkNotNull(executor, "executor");
    this.deadlineCancellationExecutor =
        checkNotNull(deadlineCancellationExecutor, "deadlineCancellationExecutor");
    this.callsTracer = checkNotNull(callsTracer, "callsTracer");
    this.configSelector = checkNotNull(configSelector, "configSelector");
  }

  @Override
  public <RequestT, ResponseT> ClientCall<RequestT, ResponseT> newCall(
      MethodDescriptor<RequestT, ResponseT> methodDescriptor, CallOptions callOptions) {
    final Executor effectiveExecutor =
        callOptions.getExecutor() == null ? executor : callOptions.getExecutor();
    if (callOptions.isWaitForReady()) {
      return new ClientCall<RequestT, ResponseT>() {
        @Override
        public void start(final ClientCall.Listener<ResponseT> listener, Metadata headers) {
          effectiveExecutor.execute(new Runnable() {
              @Override
              public void run() {
                listener.onClose(WAIT_FOR_READY_ERROR, new Metadata());
              }
            });
        }

        @Override
        public void request(int numMessages) {}

        @Override
        public void cancel(String message, Throwable cause) {}

        @Override
        public void halfClose() {}

        @Override
        public void sendMessage(RequestT message) {}
      };
    }
    return new ClientCallImpl<>(methodDescriptor,
        effectiveExecutor,
        callOptions.withOption(GrpcUtil.CALL_OPTIONS_RPC_OWNED_BY_BALANCER, Boolean.TRUE),
        transportProvider, deadlineCancellationExecutor, callsTracer, configSelector.get());
  }

  @Override
  public String authority() {
    return subchannel.getAuthority();
  }
}

1	/*
2	* Copyright 2016 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 static com.google.common.base.Preconditions.checkNotNull;
20
21	import com.google.common.annotations.VisibleForTesting;
22	import io.grpc.CallOptions;
23	import io.grpc.Channel;
24	import io.grpc.ClientCall;
25	import io.grpc.ClientStreamTracer;
26	import io.grpc.Context;
27	import io.grpc.InternalConfigSelector;
28	import io.grpc.Metadata;
29	import io.grpc.MethodDescriptor;
30	import io.grpc.Status;
31	import io.grpc.internal.ClientCallImpl.ClientStreamProvider;
32	import io.grpc.internal.ClientStreamListener.RpcProgress;
33	import java.util.concurrent.Executor;
34	import java.util.concurrent.ScheduledExecutorService;
35	import java.util.concurrent.atomic.AtomicReference;
36
37	final class SubchannelChannel extends Channel {
38	@VisibleForTesting
39	static final Status NOT_READY_ERROR =	1✔
40	Status.UNAVAILABLE.withDescription("Subchannel is NOT READY");	1✔
41	@VisibleForTesting
42	static final Status WAIT_FOR_READY_ERROR =	1✔
43	Status.UNAVAILABLE.withDescription(	1✔
44	"wait-for-ready RPC is not supported on Subchannel.asChannel()");
45	private static final FailingClientTransport notReadyTransport =	1✔
46	new FailingClientTransport(NOT_READY_ERROR, RpcProgress.MISCARRIED);
47	private final InternalSubchannel subchannel;
48	private final Executor executor;
49	private final ScheduledExecutorService deadlineCancellationExecutor;
50	private final CallTracer callsTracer;
51	private final AtomicReference<InternalConfigSelector> configSelector;
52
53	private final ClientStreamProvider transportProvider = new ClientStreamProvider() {	1✔
54	@Override
55	public ClientStream newStream(MethodDescriptor<?, ?> method,
56	CallOptions callOptions, Metadata headers, Context context) {
57	ClientTransport transport = subchannel.getTransport();	1✔
58	if (transport == null) {	1✔
59	transport = notReadyTransport;	1✔
60	}
61	ClientStreamTracer[] tracers = GrpcUtil.getClientStreamTracers(	1✔
62	callOptions, headers, 0, /* isTransparentRetry= */ false,
63	/* isHedging= */ false);
64	Context origContext = context.attach();	1✔
65	try {
66	return transport.newStream(method, headers, callOptions, tracers);	1✔
67	} finally {
68	context.detach(origContext);	1✔
69	}
70	}
71	};
72
73	SubchannelChannel(
74	InternalSubchannel subchannel, Executor executor,
75	ScheduledExecutorService deadlineCancellationExecutor, CallTracer callsTracer,
76	AtomicReference<InternalConfigSelector> configSelector) {	1✔
77	this.subchannel = checkNotNull(subchannel, "subchannel");	1✔
78	this.executor = checkNotNull(executor, "executor");	1✔
79	this.deadlineCancellationExecutor =	1✔
80	checkNotNull(deadlineCancellationExecutor, "deadlineCancellationExecutor");	1✔
81	this.callsTracer = checkNotNull(callsTracer, "callsTracer");	1✔
82	this.configSelector = checkNotNull(configSelector, "configSelector");	1✔
83	}	1✔
84
85	@Override
86	public <RequestT, ResponseT> ClientCall<RequestT, ResponseT> newCall(
87	MethodDescriptor<RequestT, ResponseT> methodDescriptor, CallOptions callOptions) {
88	final Executor effectiveExecutor =
89	callOptions.getExecutor() == null ? executor : callOptions.getExecutor();	1✔
90	if (callOptions.isWaitForReady()) {	1✔
91	return new ClientCall<RequestT, ResponseT>() {	1✔
92	@Override
93	public void start(final ClientCall.Listener<ResponseT> listener, Metadata headers) {
94	effectiveExecutor.execute(new Runnable() {	1✔
95	@Override
96	public void run() {
97	listener.onClose(WAIT_FOR_READY_ERROR, new Metadata());	1✔
98	}	1✔
99	});
100	}	1✔
101
102	@Override
103	public void request(int numMessages) {}	×
104
105	@Override
106	public void cancel(String message, Throwable cause) {}	×
107
108	@Override
109	public void halfClose() {}	×
110
111	@Override
112	public void sendMessage(RequestT message) {}	×
113	};
114	}
115	return new ClientCallImpl<>(methodDescriptor,	1✔
116	effectiveExecutor,
117	callOptions.withOption(GrpcUtil.CALL_OPTIONS_RPC_OWNED_BY_BALANCER, Boolean.TRUE),	1✔
118	transportProvider, deadlineCancellationExecutor, callsTracer, configSelector.get());	1✔
119	}
120
121	@Override
122	public String authority() {
123	return subchannel.getAuthority();	×
124	}
125	}

grpc / grpc-java / #19996

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