Skip to content

Commit d1375ba

Browse files
EvenLjjliujianjun.ljjJust-CJ
authored
V5.13.3 release to v5.14.0 (#1486)
* fix triple header and usedId problem * Increase the extensibility of the header conversion context. * fix the triple header context transfer issue * fix the triple header context transfer issue (#1476) * fix triple POJO stream parent interface method not cache stream call type issue (#1481) Co-authored-by: liujianjun.ljj <[email protected]> * fix method timeout resolve problem (#1478) Co-authored-by: liujianjun.ljj <[email protected]> * fix triple stream deserialize issue when in serverless scene (#1480) Co-authored-by: liujianjun.ljj <[email protected]> * fix: hessian deserialize support sofa.serialize.dynamic.load.enable (#1482) (cherry picked from commit d20062f) Co-authored-by: 辰霖 <[email protected]> * modify permitKeepAliveTime and permitKeepAliveWithoutCalls to avoid too many ping problem (#1483) Co-authored-by: liujianjun.ljj <[email protected]> * Fix serialization and deserialization classloader issues in serverless scene (#1484) Co-authored-by: liujianjun.ljj <[email protected]> * enhance triple tracelog in server (#1477) Co-authored-by: liujianjun.ljj <[email protected]> * 5.13.3 release --------- Co-authored-by: liujianjun.ljj <[email protected]> Co-authored-by: 辰霖 <[email protected]>
1 parent 7d1a052 commit d1375ba

File tree

20 files changed

+497
-126
lines changed

20 files changed

+497
-126
lines changed

‎core/api/src/main/java/com/alipay/sofa/rpc/common/RpcConstants.java

+3-3
Original file line numberDiff line numberDiff line change
@@ -128,15 +128,15 @@ public class RpcConstants {
128128
/**
129129
* 调用方式：客户端流
130130
*/
131-
public static final String INVOKER_TYPE_CLIENT_STREAMING = "clientStream";
131+
public static final String INVOKER_TYPE_CLIENT_STREAMING = "client_stream";
132132
/**
133133
* 调用方式：服务端流
134134
*/
135-
public static final String INVOKER_TYPE_SERVER_STREAMING = "serverStream";
135+
public static final String INVOKER_TYPE_SERVER_STREAMING = "server_stream";
136136
/**
137137
* 调用方式：双向流
138138
*/
139-
public static final String INVOKER_TYPE_BI_STREAMING = "bidirectionalStream";
139+
public static final String INVOKER_TYPE_BI_STREAMING = "bi_stream";
140140

141141
/**
142142
* Hessian序列化 [不推荐]

‎core/api/src/main/java/com/alipay/sofa/rpc/config/ConsumerConfig.java

+5-2
Original file line numberDiff line numberDiff line change
@@ -933,8 +933,11 @@ public int getMethodRetries(String methodName) {
933933
* @return the time out
934934
*/
935935
public int getMethodTimeout(String methodName) {
936-
return (Integer) getMethodConfigValue(methodName, RpcConstants.CONFIG_KEY_TIMEOUT,
937-
getTimeout());
936+
Object methodTimeout = getMethodConfigValue(methodName, RpcConstants.CONFIG_KEY_TIMEOUT);
937+
if (methodTimeout == null || ((Integer) methodTimeout) == 0) {
938+
return getTimeout();
939+
}
940+
return (Integer) methodTimeout;
938941
}
939942

940943
/**

‎core/api/src/main/java/com/alipay/sofa/rpc/core/request/SofaRequest.java

+5-2
Original file line numberDiff line numberDiff line change
@@ -306,7 +306,10 @@ public SofaRequest setData(AbstractByteBuf data) {
306306
* @return 如果是Future和Callback，是异步请求
307307
*/
308308
public boolean isAsync() {
309-
return invokeType != null && (RpcConstants.INVOKER_TYPE_CALLBACK.equals(invokeType)
310-
|| RpcConstants.INVOKER_TYPE_FUTURE.equals(invokeType));
309+
return RpcConstants.INVOKER_TYPE_CALLBACK.equals(invokeType)
310+
|| RpcConstants.INVOKER_TYPE_FUTURE.equals(invokeType)
311+
|| RpcConstants.INVOKER_TYPE_BI_STREAMING.equals(invokeType)
312+
|| RpcConstants.INVOKER_TYPE_SERVER_STREAMING.equals(invokeType)
313+
|| RpcConstants.INVOKER_TYPE_CLIENT_STREAMING.equals(invokeType);
311314
}
312315
}

‎core/api/src/test/java/com/alipay/sofa/rpc/config/ConsumerConfigTest.java

+56
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,56 @@
1+
/*
2+
* Licensed to the Apache Software Foundation (ASF) under one or more
3+
* contributor license agreements. See the NOTICE file distributed with
4+
* this work for additional information regarding copyright ownership.
5+
* The ASF licenses this file to You under the Apache License, Version 2.0
6+
* (the "License"); you may not use this file except in compliance with
7+
* the License. You may obtain a copy of the License at
8+
*
9+
* http://www.apache.org/licenses/LICENSE-2.0
10+
*
11+
* Unless required by applicable law or agreed to in writing, software
12+
* distributed under the License is distributed on an "AS IS" BASIS,
13+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14+
* See the License for the specific language governing permissions and
15+
* limitations under the License.
16+
*/
17+
package com.alipay.sofa.rpc.config;
18+
19+
import com.alipay.sofa.rpc.invoke.Invoker;
20+
import org.junit.Assert;
21+
import org.junit.Test;
22+
23+
import java.util.ArrayList;
24+
import java.util.List;
25+
26+
/**
27+
* @author Even
28+
* @date 2025/3/4 21:36
29+
*/
30+
public class ConsumerConfigTest {
31+
32+
@Test
33+
public void testMethodTimeout() {
34+
ConsumerConfig<Invoker> consumerConfig = new ConsumerConfig<>();
35+
consumerConfig.setTimeout(4000);
36+
consumerConfig.setInterfaceId(Invoker.class.getName());
37+
consumerConfig.getConfigValueCache(true);
38+
Assert.assertEquals(4000, consumerConfig.getMethodTimeout("invoke"));
39+
40+
List<MethodConfig> methodConfigs = new ArrayList<>();
41+
MethodConfig methodConfig = new MethodConfig();
42+
methodConfig.setName("invoke");
43+
methodConfigs.add(methodConfig);
44+
consumerConfig.setMethods(methodConfigs);
45+
consumerConfig.getConfigValueCache(true);
46+
Assert.assertEquals(4000, consumerConfig.getMethodTimeout("invoke"));
47+
48+
methodConfig.setTimeout(5000);
49+
consumerConfig.getConfigValueCache(true);
50+
Assert.assertEquals(5000, consumerConfig.getMethodTimeout("invoke"));
51+
52+
methodConfig.setTimeout(-1);
53+
consumerConfig.getConfigValueCache(true);
54+
Assert.assertEquals(-1, consumerConfig.getMethodTimeout("invoke"));
55+
}
56+
}

‎remoting/pom.xml

+5
Original file line numberDiff line numberDiff line change
@@ -88,6 +88,11 @@
8888
<artifactId>sofa-rpc-log-common-tools</artifactId>
8989
<version>${project.parent.version}</version>
9090
</dependency>
91+
<dependency>
92+
<groupId>com.alipay.sofa</groupId>
93+
<artifactId>sofa-rpc-tracer-opentracing</artifactId>
94+
<version>${project.parent.version}</version>
95+
</dependency>
9196
</dependencies>
9297
</dependencyManagement>
9398

‎remoting/remoting-triple/pom.xml

+4
Original file line numberDiff line numberDiff line change
@@ -66,5 +66,9 @@
6666
<groupId>com.alipay.sofa</groupId>
6767
<artifactId>tracer-core</artifactId>
6868
</dependency>
69+
<dependency>
70+
<groupId>com.alipay.sofa</groupId>
71+
<artifactId>sofa-rpc-tracer-opentracing</artifactId>
72+
</dependency>
6973
</dependencies>
7074
</project>

‎remoting/remoting-triple/src/main/java/com/alipay/sofa/rpc/interceptor/ClientHeaderClientInterceptor.java

+59-9
Original file line numberDiff line numberDiff line change
@@ -16,12 +16,20 @@
1616
*/
1717
package com.alipay.sofa.rpc.interceptor;
1818

19+
import com.alipay.common.tracer.core.context.trace.SofaTraceContext;
20+
import com.alipay.common.tracer.core.holder.SofaTraceContextHolder;
21+
import com.alipay.common.tracer.core.span.SofaTracerSpan;
1922
import com.alipay.sofa.rpc.config.ConsumerConfig;
23+
import com.alipay.sofa.rpc.context.RpcInternalContext;
2024
import com.alipay.sofa.rpc.context.RpcInvokeContext;
2125
import com.alipay.sofa.rpc.context.RpcRunningState;
2226
import com.alipay.sofa.rpc.core.request.SofaRequest;
27+
import com.alipay.sofa.rpc.core.response.SofaResponse;
28+
import com.alipay.sofa.rpc.event.ClientAsyncReceiveEvent;
29+
import com.alipay.sofa.rpc.event.EventBus;
2330
import com.alipay.sofa.rpc.server.triple.TripleContants;
2431
import com.alipay.sofa.rpc.tracer.sofatracer.TripleTracerAdapter;
32+
import com.alipay.sofa.rpc.utils.TripleExceptionUtils;
2533
import io.grpc.CallOptions;
2634
import io.grpc.Channel;
2735
import io.grpc.ClientCall;
@@ -59,12 +67,15 @@ public <ReqT, RespT> ClientCall<ReqT, RespT> interceptCall(MethodDescriptor<ReqT
5967

6068
@Override
6169
public void start(Listener<RespT> responseListener, Metadata requestHeader) {
62-
70+
RpcInternalContext internalContext = RpcInternalContext.getContext();
6371
RpcInvokeContext context = RpcInvokeContext.getContext();
6472
SofaRequest sofaRequest = (SofaRequest) context.get(TripleContants.SOFA_REQUEST_KEY);
6573

66-
ConsumerConfig consumerConfig = (ConsumerConfig) context.get(TripleContants.SOFA_CONSUMER_CONFIG_KEY);
67-
TripleTracerAdapter.beforeSend(sofaRequest, consumerConfig, requestHeader);
74+
ConsumerConfig<?> consumerConfig = (ConsumerConfig<?>) context
75+
.get(TripleContants.SOFA_CONSUMER_CONFIG_KEY);
76+
TripleTracerAdapter.beforeSend(sofaRequest, consumerConfig, requestHeader, method);
77+
SofaTraceContext sofaTraceContext = SofaTraceContextHolder.getSofaTraceContext();
78+
SofaTracerSpan clientSpan = sofaTraceContext.getCurrentSpan();
6879
if (RpcRunningState.isDebugMode()) {
6980
LOGGER.info("[2]prepare to send from client:{}", requestHeader);
7081
}
@@ -80,18 +91,48 @@ public void onHeaders(Metadata responseHeader) {
8091

8192
@Override
8293
public void onMessage(RespT message) {
83-
if (RpcRunningState.isDebugMode()) {
84-
LOGGER.info("[4]response message received from server:{}", message);
94+
// onMessage -> onNext()
95+
try {
96+
if (sofaRequest.isAsync()) {
97+
RpcInvokeContext.setContext(context);
98+
sofaTraceContext.push(clientSpan);
99+
}
100+
if (RpcRunningState.isDebugMode()) {
101+
LOGGER.info("[4]response message received from server:{}", message);
102+
}
103+
super.onMessage(message);
104+
} finally {
105+
if (sofaRequest.isAsync()) {
106+
sofaTraceContext.clear();
107+
RpcInvokeContext.removeContext();
108+
}
85109
}
86-
super.onMessage(message);
87110
}
88111

89112
@Override
90113
public void onClose(Status status, Metadata trailers) {
91-
if (RpcRunningState.isDebugMode()) {
92-
LOGGER.info("[5]response close received from server:{},trailers:{}", status, trailers);
114+
// onClose -> onComplete() or onError()
115+
try {
116+
if (sofaRequest.isAsync()) {
117+
RpcInvokeContext.setContext(context);
118+
sofaTraceContext.push(clientSpan);
119+
}
120+
if (RpcRunningState.isDebugMode()) {
121+
LOGGER.info("[5]response close received from server:{},trailers:{}", status, trailers);
122+
}
123+
super.onClose(status, trailers);
124+
} finally {
125+
if (sofaRequest.isAsync()) {
126+
Throwable throwable = TripleExceptionUtils.getThrowableFromStatus(status);
127+
RpcInternalContext.setContext(internalContext);
128+
if (EventBus.isEnable(ClientAsyncReceiveEvent.class)) {
129+
EventBus.post(new ClientAsyncReceiveEvent(consumerConfig, null,
130+
sofaRequest, new SofaResponse(), throwable));
131+
}
132+
RpcInvokeContext.removeContext();
133+
RpcInternalContext.removeAllContext();
134+
}
93135
}
94-
super.onClose(status, trailers);
95136
}
96137

97138
@Override
@@ -104,6 +145,15 @@ public void onReady() {
104145
}, requestHeader);
105146
}
106147

148+
@Override
149+
public void sendMessage(ReqT message) {
150+
try {
151+
super.sendMessage(message);
152+
} catch (Throwable t) {
153+
LOGGER.error("Client invoke grpc sendMessage meet error:", t);
154+
throw t;
155+
}
156+
}
107157
};
108158
}
109159
}

0 commit comments

Comments
 (0)