natas_hw
/
dubbo
0
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

fix tri filter onError (#11133) 

* fix tri filter onError

* fix tri filter onError

* fix tri filter onError

* fix ut
This commit is contained in:
earthchen 2022-12-14 21:34:56 +08:00 committed by GitHub
parent e219c1f4da
commit 342e4f5bb6
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
2 changed files with 14 additions and 16 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

19
dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/DeadlineFuture.java
View File

@ -115,7 +115,7 @@ public class DeadlineFuture extends CompletableFuture<AppResponse> {
@Override
public boolean cancel(boolean mayInterruptIfRunning) {
timeoutTask.cancel();
doReceived(TriRpcStatus.CANCELLED, null);
doReceived(TriRpcStatus.CANCELLED, new AppResponse(TriRpcStatus.CANCELLED.asException()));
return true;
}
@ -127,12 +127,13 @@ public class DeadlineFuture extends CompletableFuture<AppResponse> {
if (isDone() || isCancelled() || isCompletedExceptionally()) {
return;
}
if (status.isOk()) {
this.complete(appResponse);
} else {
this.completeExceptionally(
status.appendDescription("RemoteAddress:" + address).asException());
}
// Still needs to be discussed here, but for now, that's it
// Remove the judgment of status is ok,
// because the completelyExceptionally method will lead to the onError method in the filter,
// but there are also exceptions in the onResponse in the filter,which is a bit confusing.
// We recommend only handling onResponse in which onError is called for handling
this.complete(appResponse);
// the result is returning, but the caller thread may still waiting
// to avoid endless waiting for whatever reason, notify caller thread to return.
@ -178,7 +179,9 @@ public class DeadlineFuture extends CompletableFuture<AppResponse> {
private void notifyTimeout() {
final TriRpcStatus status = TriRpcStatus.DEADLINE_EXCEEDED.withDescription(
getTimeoutMessage());
DeadlineFuture.this.doReceived(status, null);
AppResponse timeoutResponse = new AppResponse();
timeoutResponse.setException(status.asException());
DeadlineFuture.this.doReceived(status, timeoutResponse);
}
}

11
dubbo-rpc/dubbo-rpc-triple/src/test/java/org/apache/dubbo/rpc/protocol/tri/DeadlineFutureTest.java
View File

@ -28,8 +28,6 @@ import org.junit.jupiter.api.Test;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
import static org.junit.jupiter.api.Assertions.fail;
class DeadlineFutureTest {
@Test
@ -40,12 +38,9 @@ class DeadlineFutureTest {
DeadlineFuture timeout = DeadlineFuture.newFuture(service, method, address, 10,
ImmediateEventExecutor.INSTANCE);
TimeUnit.MILLISECONDS.sleep(20);
try {
timeout.get();
fail();
} catch (ExecutionException e) {
Assertions.assertTrue(e.getCause() instanceof StatusRpcException);
}
AppResponse timeoutResponse = timeout.get();
Assertions.assertTrue(timeoutResponse.getException() instanceof StatusRpcException);
DeadlineFuture success = DeadlineFuture.newFuture(service, method, address, 1000,
ImmediateEventExecutor.INSTANCE);