Skip to content

Commit 0278a01

Browse files
carryxyhchickenlj
carryxyh
authored and
chickenlj
committed
Merge pull request #2024, binding attachment before a clusterInvoker invoke.
Fixes #1978
1 parent 0882c83 commit 0278a01

File tree

6 files changed

+116
-46
lines changed

6 files changed

+116
-46
lines changed

‎dubbo-cluster/src/main/java/com/alibaba/dubbo/rpc/cluster/support/AbstractClusterInvoker.java‎

Lines changed: 10 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -25,6 +25,8 @@
2525
import com.alibaba.dubbo.common.utils.NetUtils;
2626
import com.alibaba.dubbo.rpc.Invocation;
2727
import com.alibaba.dubbo.rpc.Invoker;
28+
import com.alibaba.dubbo.rpc.RpcContext;
29+
import com.alibaba.dubbo.rpc.RpcInvocation;
2830
import com.alibaba.dubbo.rpc.Result;
2931
import com.alibaba.dubbo.rpc.RpcException;
3032
import com.alibaba.dubbo.rpc.cluster.Directory;
@@ -33,6 +35,7 @@
3335

3436
import java.util.ArrayList;
3537
import java.util.List;
38+
import java.util.Map;
3639
import java.util.concurrent.atomic.AtomicBoolean;
3740

3841
/**
@@ -225,6 +228,13 @@ private Invoker<T> reselect(LoadBalance loadbalance, Invocation invocation,
225228
public Result invoke(final Invocation invocation) throws RpcException {
226229
checkWhetherDestroyed();
227230
LoadBalance loadbalance = null;
231+
232+
// binding attachments into invocation.
233+
Map<String, String> contextAttachments = RpcContext.getContext().getAttachments();
234+
if (contextAttachments != null && contextAttachments.size() != 0) {
235+
((RpcInvocation) invocation).addAttachments(contextAttachments);
236+
}
237+
228238
List<Invoker<T>> invokers = list(invocation);
229239
if (invokers != null && !invokers.isEmpty()) {
230240
loadbalance = ExtensionLoader.getExtensionLoader(LoadBalance.class).getExtension(invokers.get(0).getUrl()

‎dubbo-cluster/src/main/java/com/alibaba/dubbo/rpc/cluster/support/ForkingClusterInvoker.java‎

Lines changed: 43 additions & 38 deletions
Original file line numberDiff line numberDiff line change
@@ -57,50 +57,55 @@ public ForkingClusterInvoker(Directory<T> directory) {
5757
@Override
5858
@SuppressWarnings({"unchecked", "rawtypes"})
5959
public Result doInvoke(final Invocation invocation, List<Invoker<T>> invokers, LoadBalance loadbalance) throws RpcException {
60-
checkInvokers(invokers, invocation);
61-
final List<Invoker<T>> selected;
62-
final int forks = getUrl().getParameter(Constants.FORKS_KEY, Constants.DEFAULT_FORKS);
63-
final int timeout = getUrl().getParameter(Constants.TIMEOUT_KEY, Constants.DEFAULT_TIMEOUT);
64-
if (forks <= 0 || forks >= invokers.size()) {
65-
selected = invokers;
66-
} else {
67-
selected = new ArrayList<Invoker<T>>();
68-
for (int i = 0; i < forks; i++) {
69-
// TODO. Add some comment here, refer chinese version for more details.
70-
Invoker<T> invoker = select(loadbalance, invocation, invokers, selected);
71-
if (!selected.contains(invoker)) {//Avoid add the same invoker several times.
72-
selected.add(invoker);
60+
try {
61+
checkInvokers(invokers, invocation);
62+
final List<Invoker<T>> selected;
63+
final int forks = getUrl().getParameter(Constants.FORKS_KEY, Constants.DEFAULT_FORKS);
64+
final int timeout = getUrl().getParameter(Constants.TIMEOUT_KEY, Constants.DEFAULT_TIMEOUT);
65+
if (forks <= 0 || forks >= invokers.size()) {
66+
selected = invokers;
67+
} else {
68+
selected = new ArrayList<Invoker<T>>();
69+
for (int i = 0; i < forks; i++) {
70+
// TODO. Add some comment here, refer chinese version for more details.
71+
Invoker<T> invoker = select(loadbalance, invocation, invokers, selected);
72+
if (!selected.contains(invoker)) {//Avoid add the same invoker several times.
73+
selected.add(invoker);
74+
}
7375
}
7476
}
75-
}
76-
RpcContext.getContext().setInvokers((List) selected);
77-
final AtomicInteger count = new AtomicInteger();
78-
final BlockingQueue<Object> ref = new LinkedBlockingQueue<Object>();
79-
for (final Invoker<T> invoker : selected) {
80-
executor.execute(new Runnable() {
81-
@Override
82-
public void run() {
83-
try {
84-
Result result = invoker.invoke(invocation);
85-
ref.offer(result);
86-
} catch (Throwable e) {
87-
int value = count.incrementAndGet();
88-
if (value >= selected.size()) {
89-
ref.offer(e);
77+
RpcContext.getContext().setInvokers((List) selected);
78+
final AtomicInteger count = new AtomicInteger();
79+
final BlockingQueue<Object> ref = new LinkedBlockingQueue<Object>();
80+
for (final Invoker<T> invoker : selected) {
81+
executor.execute(new Runnable() {
82+
@Override
83+
public void run() {
84+
try {
85+
Result result = invoker.invoke(invocation);
86+
ref.offer(result);
87+
} catch (Throwable e) {
88+
int value = count.incrementAndGet();
89+
if (value >= selected.size()) {
90+
ref.offer(e);
91+
}
9092
}
9193
}
94+
});
95+
}
96+
try {
97+
Object ret = ref.poll(timeout, TimeUnit.MILLISECONDS);
98+
if (ret instanceof Throwable) {
99+
Throwable e = (Throwable) ret;
100+
throw new RpcException(e instanceof RpcException ? ((RpcException) e).getCode() : 0, "Failed to forking invoke provider " + selected + ", but no luck to perform the invocation. Last error is: " + e.getMessage(), e.getCause() != null ? e.getCause() : e);
92101
}
93-
});
94-
}
95-
try {
96-
Object ret = ref.poll(timeout, TimeUnit.MILLISECONDS);
97-
if (ret instanceof Throwable) {
98-
Throwable e = (Throwable) ret;
99-
throw new RpcException(e instanceof RpcException ? ((RpcException) e).getCode() : 0, "Failed to forking invoke provider " + selected + ", but no luck to perform the invocation. Last error is: " + e.getMessage(), e.getCause() != null ? e.getCause() : e);
102+
return (Result) ret;
103+
} catch (InterruptedException e) {
104+
throw new RpcException("Failed to forking invoke provider " + selected + ", but no luck to perform the invocation. Last error is: " + e.getMessage(), e);
100105
}
101-
return (Result) ret;
102-
} catch (InterruptedException e) {
103-
throw new RpcException("Failed to forking invoke provider " + selected + ", but no luck to perform the invocation. Last error is: " + e.getMessage(), e);
106+
} finally {
107+
// clear attachments which is binding to current thread.
108+
RpcContext.getContext().clearAttachments();
104109
}
105110
}
106111
}

‎dubbo-cluster/src/test/java/com/alibaba/dubbo/rpc/cluster/support/AbstractClusterInvokerTest.java‎

Lines changed: 26 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -23,6 +23,7 @@
2323
import com.alibaba.dubbo.rpc.Invocation;
2424
import com.alibaba.dubbo.rpc.Invoker;
2525
import com.alibaba.dubbo.rpc.Result;
26+
import com.alibaba.dubbo.rpc.RpcContext;
2627
import com.alibaba.dubbo.rpc.RpcException;
2728
import com.alibaba.dubbo.rpc.RpcInvocation;
2829
import com.alibaba.dubbo.rpc.cluster.Directory;
@@ -132,6 +133,31 @@ protected Result doInvoke(Invocation invocation, List invokers, LoadBalance load
132133

133134
}
134135

136+
@Test
137+
public void testBindingAttachment() {
138+
final String attachKey = "attach";
139+
final String attachValue = "value";
140+
141+
// setup attachment
142+
RpcContext.getContext().setAttachment(attachKey, attachValue);
143+
Map<String, String> attachments = RpcContext.getContext().getAttachments();
144+
Assert.assertTrue("set attachment failed!", attachments != null && attachments.size() == 1);
145+
146+
cluster = new AbstractClusterInvoker(dic) {
147+
@Override
148+
protected Result doInvoke(Invocation invocation, List invokers, LoadBalance loadbalance)
149+
throws RpcException {
150+
// attachment will be bind to invocation
151+
String value = invocation.getAttachment(attachKey);
152+
Assert.assertTrue("binding attachment failed!", value != null && value.equals(attachValue));
153+
return null;
154+
}
155+
};
156+
157+
// invoke
158+
cluster.invoke(invocation);
159+
}
160+
135161
@Test
136162
public void testSelect_Invokersize0() throws Exception {
137163
{

‎dubbo-cluster/src/test/java/com/alibaba/dubbo/rpc/cluster/support/FailoverClusterInvokerTest.java‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -166,7 +166,7 @@ public void testNoInvoke() {
166166
* then we should reselect from the latest invokers before retry.
167167
*/
168168
@Test
169-
public void testInvokerDestoryAndReList() {
169+
public void testInvokerDestroyAndReList() {
170170
final URL url = URL.valueOf("test://localhost/" + Demo.class.getName() + "?loadbalance=roundrobin&retries=" + retries);
171171
RpcException exception = new RpcException(RpcException.TIMEOUT_EXCEPTION);
172172
MockInvoker<Demo> invoker1 = new MockInvoker<Demo>(Demo.class, url);

‎dubbo-cluster/src/test/java/com/alibaba/dubbo/rpc/cluster/support/ForkingClusterInvokerTest.java‎

Lines changed: 35 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -19,25 +19,29 @@
1919
import com.alibaba.dubbo.common.URL;
2020
import com.alibaba.dubbo.rpc.Invoker;
2121
import com.alibaba.dubbo.rpc.Result;
22+
import com.alibaba.dubbo.rpc.RpcContext;
2223
import com.alibaba.dubbo.rpc.RpcException;
2324
import com.alibaba.dubbo.rpc.RpcInvocation;
2425
import com.alibaba.dubbo.rpc.RpcResult;
2526
import com.alibaba.dubbo.rpc.cluster.Directory;
2627

27-
import junit.framework.Assert;
28+
import org.junit.Assert;
2829
import org.junit.Before;
2930
import org.junit.Test;
3031

3132
import java.util.ArrayList;
3233
import java.util.List;
34+
import java.util.Map;
3335

36+
import static org.hamcrest.CoreMatchers.is;
3437
import static org.junit.Assert.assertFalse;
38+
import static org.junit.Assert.assertSame;
39+
import static org.junit.Assert.assertThat;
3540
import static org.mockito.BDDMockito.given;
3641
import static org.mockito.Mockito.mock;
3742

3843
/**
3944
* ForkingClusterInvokerTest
40-
*
4145
*/
4246
@SuppressWarnings("unchecked")
4347
public class ForkingClusterInvokerTest {
@@ -71,6 +75,7 @@ public void setUp() throws Exception {
7175
invokers.add(invoker3);
7276

7377
}
78+
7479
private void resetInvokerToException() {
7580
given(invoker1.invoke(invocation)).willThrow(new RuntimeException());
7681
given(invoker1.getUrl()).willReturn(url);
@@ -106,7 +111,7 @@ private void resetInvokerToNoException() {
106111
}
107112

108113
@Test
109-
public void testInvokeExceptoin() {
114+
public void testInvokeException() {
110115
resetInvokerToException();
111116
ForkingClusterInvoker<ForkingClusterInvokerTest> invoker = new ForkingClusterInvoker<ForkingClusterInvokerTest>(
112117
dic);
@@ -115,20 +120,44 @@ public void testInvokeExceptoin() {
115120
invoker.invoke(invocation);
116121
Assert.fail();
117122
} catch (RpcException expected) {
118-
Assert.assertTrue(expected.getMessage().contains("Failed to forking invoke provider"));
123+
assertThat(expected.getMessage().contains("Failed to forking invoke provider"), is(true));
124+
assertFalse(expected.getCause() instanceof RpcException);
125+
}
126+
}
127+
128+
@Test
129+
public void testClearRpcContext() {
130+
resetInvokerToException();
131+
ForkingClusterInvoker<ForkingClusterInvokerTest> invoker = new ForkingClusterInvoker<ForkingClusterInvokerTest>(
132+
dic);
133+
134+
String attachKey = "attach";
135+
String attachValue = "value";
136+
137+
RpcContext.getContext().setAttachment(attachKey, attachValue);
138+
139+
Map<String, String> attachments = RpcContext.getContext().getAttachments();
140+
assertThat("set attachment failed!", attachments != null && attachments.size() == 1, is(true));
141+
try {
142+
invoker.invoke(invocation);
143+
Assert.fail();
144+
} catch (RpcException expected) {
145+
assertThat(expected.getMessage().contains("Failed to forking invoke provider"), is(true));
119146
assertFalse(expected.getCause() instanceof RpcException);
120147
}
148+
Map<String, String> afterInvoke = RpcContext.getContext().getAttachments();
149+
assertThat(afterInvoke != null && afterInvoke.size() == 0, is(true));
121150
}
122151

123152
@Test()
124-
public void testInvokeNoExceptoin() {
153+
public void testInvokeNoException() {
125154

126155
resetInvokerToNoException();
127156

128157
ForkingClusterInvoker<ForkingClusterInvokerTest> invoker = new ForkingClusterInvoker<ForkingClusterInvokerTest>(
129158
dic);
130159
Result ret = invoker.invoke(invocation);
131-
Assert.assertSame(result, ret);
160+
assertSame(result, ret);
132161
}
133162

134163
}

‎dubbo-rpc/dubbo-rpc-api/src/main/java/com/alibaba/dubbo/rpc/protocol/AbstractInvoker.java‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -135,7 +135,7 @@ public Result invoke(Invocation inv) throws RpcException {
135135
invocation.addAttachmentsIfAbsent(attachment);
136136
}
137137
Map<String, String> contextAttachments = RpcContext.getContext().getAttachments();
138-
if (contextAttachments != null) {
138+
if (contextAttachments != null && contextAttachments.size() != 0) {
139139
/**
140140
* invocation.addAttachmentsIfAbsent(context){@link RpcInvocation#addAttachmentsIfAbsent(Map)}should not be used here,
141141
* because the {@link RpcContext#setAttachment(String, String)} is passed in the Filter when the call is triggered

0 commit comments

Comments
 (0)
⚡