diff --git a/dubbo-remoting/dubbo-remoting-api/src/main/java/org/apache/dubbo/remoting/exchange/codec/ExchangeCodec.java b/dubbo-remoting/dubbo-remoting-api/src/main/java/org/apache/dubbo/remoting/exchange/codec/ExchangeCodec.java index 44cd564f57..6fdbbea497 100644 --- a/dubbo-remoting/dubbo-remoting-api/src/main/java/org/apache/dubbo/remoting/exchange/codec/ExchangeCodec.java +++ b/dubbo-remoting/dubbo-remoting-api/src/main/java/org/apache/dubbo/remoting/exchange/codec/ExchangeCodec.java @@ -42,6 +42,8 @@ import org.apache.dubbo.remoting.transport.ExceedPayloadLimitException; import java.io.ByteArrayInputStream; import java.io.IOException; import java.io.InputStream; +import java.text.SimpleDateFormat; +import java.util.Date; /** * ExchangeCodec. @@ -167,7 +169,7 @@ public class ExchangeCodec extends TelnetCodec { data = decodeEventData(channel, CodecSupport.deserialize(channel.getUrl(), new ByteArrayInputStream(eventPayload), proto), eventPayload); } } else { - data = decodeResponseData(channel, CodecSupport.deserialize(channel.getUrl(), is, proto), getRequestData(id)); + data = decodeResponseData(channel, CodecSupport.deserialize(channel.getUrl(), is, proto), getRequestData(channel, res, id)); } res.setResult(data); } else { @@ -209,16 +211,21 @@ public class ExchangeCodec extends TelnetCodec { } } - protected Object getRequestData(long id) { + protected Object getRequestData(Channel channel, Response response, long id) { DefaultFuture future = DefaultFuture.getFuture(id); - if (future == null) { - return null; + if (future != null) { + Request req = future.getRequest(); + if (req != null) { + return req.getData(); + } } - Request req = future.getRequest(); - if (req == null) { - return null; - } - return req.getData(); + + logger.warn("The timeout response finally returned at " + + (new SimpleDateFormat("yyyy-MM-dd HH:mm:ss.SSS").format(new Date())) + + ", response status is " + response.getStatus() + ", response id is " + response.getId() + + (channel == null ? "" : ", channel: " + channel.getLocalAddress() + + " -> " + channel.getRemoteAddress()) + ", please check provider side for detailed result."); + throw new IllegalArgumentException("Failed to find any request match the response, response id: " + id); } protected void encodeRequest(Channel channel, ChannelBuffer buffer, Request req) throws IOException { @@ -427,7 +434,7 @@ public class ExchangeCodec extends TelnetCodec { try { if (eventBytes != null) { int dataLen = eventBytes.length; - int threshold = ConfigurationUtils.getSystemConfiguration(channel.getUrl().getScopeModel()).getInt("deserialization.event.size", 50); + int threshold = ConfigurationUtils.getSystemConfiguration(channel.getUrl().getScopeModel()).getInt("deserialization.event.size", 15); if (dataLen > threshold) { throw new IllegalArgumentException("Event data too long, actual size " + threshold + ", threshold " + threshold + " rejected for security consideration."); } diff --git a/dubbo-remoting/dubbo-remoting-api/src/test/java/org/apache/dubbo/remoting/codec/ExchangeCodecTest.java b/dubbo-remoting/dubbo-remoting-api/src/test/java/org/apache/dubbo/remoting/codec/ExchangeCodecTest.java index 495092d8bb..54247f28af 100644 --- a/dubbo-remoting/dubbo-remoting-api/src/test/java/org/apache/dubbo/remoting/codec/ExchangeCodecTest.java +++ b/dubbo-remoting/dubbo-remoting-api/src/test/java/org/apache/dubbo/remoting/codec/ExchangeCodecTest.java @@ -30,11 +30,13 @@ import org.apache.dubbo.remoting.buffer.ChannelBuffers; import org.apache.dubbo.remoting.exchange.Request; import org.apache.dubbo.remoting.exchange.Response; import org.apache.dubbo.remoting.exchange.codec.ExchangeCodec; +import org.apache.dubbo.remoting.exchange.support.DefaultFuture; import org.apache.dubbo.remoting.telnet.codec.TelnetCodec; import org.junit.jupiter.api.Assertions; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; +import org.mockito.Mockito; import java.io.ByteArrayOutputStream; import java.io.IOException; @@ -137,6 +139,8 @@ public class ExchangeCodecTest extends TelnetCodecTest { @Test public void test_Decode_Error_Length() throws IOException { + DefaultFuture future = DefaultFuture.newFuture(Mockito.mock(Channel.class), new Request(0), 100000, null); + byte[] header = new byte[]{MAGIC_HIGH, MAGIC_LOW, 0x02, 20, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0}; Person person = new Person(); byte[] request = getRequestBytes(person, header); @@ -148,6 +152,8 @@ public class ExchangeCodecTest extends TelnetCodecTest { Assertions.assertEquals(person, obj.getResult()); //only decode necessary bytes Assertions.assertEquals(request.length, buffer.readerIndex()); + + future.cancel(); } @Test @@ -226,6 +232,7 @@ public class ExchangeCodecTest extends TelnetCodecTest { @Test public void test_Decode_Return_Response_Person() throws IOException { + DefaultFuture future = DefaultFuture.newFuture(Mockito.mock(Channel.class), new Request(0), 100000, null); //00000010-response/oneway/hearbeat=false/hessian |20-stats=ok|id=0|length=0 byte[] header = new byte[]{MAGIC_HIGH, MAGIC_LOW, 2, 20, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0}; Person person = new Person(); @@ -235,6 +242,8 @@ public class ExchangeCodecTest extends TelnetCodecTest { Assertions.assertEquals(20, obj.getStatus()); Assertions.assertEquals(person, obj.getResult()); System.out.println(obj); + + future.cancel(); } @Test //The status input has a problem, and the read information is wrong when the serialization is serialized. @@ -324,6 +333,7 @@ public class ExchangeCodecTest extends TelnetCodecTest { @Test public void test_Header_Response_NoSerializationFlag() throws IOException { + DefaultFuture future = DefaultFuture.newFuture(Mockito.mock(Channel.class), new Request(0), 100000, null); //00000010-response/oneway/hearbeat=false/noset |20-stats=ok|id=0|length=0 byte[] header = new byte[]{MAGIC_HIGH, MAGIC_LOW, (byte) 0x02, 20, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0}; Person person = new Person(); @@ -333,10 +343,13 @@ public class ExchangeCodecTest extends TelnetCodecTest { Assertions.assertEquals(20, obj.getStatus()); Assertions.assertEquals(person, obj.getResult()); System.out.println(obj); + + future.cancel(); } @Test public void test_Header_Response_Heartbeat() throws IOException { + DefaultFuture future = DefaultFuture.newFuture(Mockito.mock(Channel.class), new Request(0), 100000, null); //00000010-response/oneway/hearbeat=true |20-stats=ok|id=0|length=0 byte[] header = new byte[]{MAGIC_HIGH, MAGIC_LOW, 0x02, 20, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0}; Person person = new Person(); @@ -346,6 +359,8 @@ public class ExchangeCodecTest extends TelnetCodecTest { Assertions.assertEquals(20, obj.getStatus()); Assertions.assertEquals(person, obj.getResult()); System.out.println(obj); + + future.cancel(); } @Test @@ -371,6 +386,7 @@ public class ExchangeCodecTest extends TelnetCodecTest { @Test public void test_Encode_Response() throws IOException { + DefaultFuture future = DefaultFuture.newFuture(Mockito.mock(Channel.class), new Request(1001), 100000, null); ChannelBuffer encodeBuffer = ChannelBuffers.dynamicBuffer(1024); Channel channel = getClientSideChannel(url); Response response = new Response(); @@ -396,6 +412,7 @@ public class ExchangeCodecTest extends TelnetCodecTest { // encode response verson ?? // Assertions.assertEquals(response.getProtocolVersion(), obj.getVersion()); + future.cancel(); } @Test diff --git a/dubbo-rpc/dubbo-rpc-dubbo/src/main/java/org/apache/dubbo/rpc/protocol/dubbo/DubboCodec.java b/dubbo-rpc/dubbo-rpc-dubbo/src/main/java/org/apache/dubbo/rpc/protocol/dubbo/DubboCodec.java index 913138dc66..652a5f9e42 100644 --- a/dubbo-rpc/dubbo-rpc-dubbo/src/main/java/org/apache/dubbo/rpc/protocol/dubbo/DubboCodec.java +++ b/dubbo-rpc/dubbo-rpc-dubbo/src/main/java/org/apache/dubbo/rpc/protocol/dubbo/DubboCodec.java @@ -101,12 +101,12 @@ public class DubboCodec extends ExchangeCodec { DecodeableRpcResult result; if (channel.getUrl().getParameter(DECODE_IN_IO_THREAD_KEY, DEFAULT_DECODE_IN_IO_THREAD)) { result = new DecodeableRpcResult(channel, res, is, - (Invocation) getRequestData(id), proto); + (Invocation) getRequestData(channel, res, id), proto); result.decode(); } else { result = new DecodeableRpcResult(channel, res, new UnsafeByteArrayInputStream(readMessageData(is)), - (Invocation) getRequestData(id), proto); + (Invocation) getRequestData(channel, res, id), proto); } data = result; } diff --git a/dubbo-rpc/dubbo-rpc-dubbo/src/test/java/org/apache/dubbo/rpc/protocol/dubbo/DubboCountCodecTest.java b/dubbo-rpc/dubbo-rpc-dubbo/src/test/java/org/apache/dubbo/rpc/protocol/dubbo/DubboCountCodecTest.java index b531bb1aa0..212efd5fd4 100644 --- a/dubbo-rpc/dubbo-rpc-dubbo/src/test/java/org/apache/dubbo/rpc/protocol/dubbo/DubboCountCodecTest.java +++ b/dubbo-rpc/dubbo-rpc-dubbo/src/test/java/org/apache/dubbo/rpc/protocol/dubbo/DubboCountCodecTest.java @@ -22,6 +22,7 @@ import org.apache.dubbo.remoting.buffer.ChannelBuffer; import org.apache.dubbo.remoting.buffer.ChannelBuffers; import org.apache.dubbo.remoting.exchange.Request; import org.apache.dubbo.remoting.exchange.Response; +import org.apache.dubbo.remoting.exchange.support.DefaultFuture; import org.apache.dubbo.remoting.exchange.support.MultiMessage; import org.apache.dubbo.rpc.AppResponse; import org.apache.dubbo.rpc.RpcInvocation; @@ -32,7 +33,9 @@ import org.apache.dubbo.rpc.protocol.dubbo.support.DemoService; import org.junit.jupiter.api.Assertions; import org.junit.jupiter.api.Test; +import java.util.ArrayList; import java.util.Iterator; +import java.util.List; import static org.apache.dubbo.rpc.Constants.INPUT_KEY; import static org.apache.dubbo.rpc.Constants.OUTPUT_KEY; @@ -45,23 +48,25 @@ public class DubboCountCodecTest { ChannelBuffer buffer = ChannelBuffers.buffer(1024); Channel channel = new MockChannel(); Assertions.assertEquals(Codec2.DecodeResult.NEED_MORE_INPUT, dubboCountCodec.decode(channel, buffer)); + List futures = new ArrayList<>(); for (int i = 0; i < 10; i++) { - Request request = new Request(1); + Request request = new Request(i); + futures.add(DefaultFuture.newFuture(channel, request, 1000, null)); RpcInvocation rpcInvocation = new RpcInvocation(null, "echo", DemoService.class.getName(), "", new Class[]{String.class}, new String[]{"yug"}); request.setData(rpcInvocation); dubboCountCodec.encode(channel, buffer, request); } for (int i = 0; i < 10; i++) { - Response response = new Response(1); + Response response = new Response(i); AppResponse appResponse = new AppResponse(i); response.setResult(appResponse); dubboCountCodec.encode(channel, buffer, response); } MultiMessage multiMessage = (MultiMessage) dubboCountCodec.decode(channel, buffer); - Assertions.assertEquals(multiMessage.size(), 20); + Assertions.assertEquals(20, multiMessage.size()); int requestCount = 0; int responseCount = 0; Iterator iterator = multiMessage.iterator(); @@ -79,6 +84,8 @@ public class DubboCountCodecTest { } Assertions.assertEquals(requestCount, 10); Assertions.assertEquals(responseCount, 10); + + futures.forEach(DefaultFuture::cancel); } }