register callback once for the full filter chain.

dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/AsyncRpcResult.java

@@ -137,20 +137,10 @@ public Object recreate() throws Throwable {
     }
 
     public Result thenApplyWithContext(Function<Result, Result> fn) {
-        CompletableFuture<Result> future = this.thenApply(fn.compose(beforeContext).andThen(afterContext));
-        AsyncRpcResult nextAsyncRpcResult = new AsyncRpcResult(this);
-        nextAsyncRpcResult.subscribeTo(future);
-        return nextAsyncRpcResult;
-    }
-
-    public void subscribeTo(CompletableFuture<?> future) {
-        future.whenComplete((obj, t) -> {
-            if (t != null) {
-                this.completeExceptionally(t);
-            } else {
-                this.complete((Result) obj);
-            }
-        });
+        this.thenApply(fn.compose(beforeContext).andThen(afterContext));
+        // You may need to return a new Result instance representing the next async stage,
+        // like thenApply will return a new CompletableFuture.
+        return this;
     }
 
     @Override

dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/filter/CallbackRegistrationFilter.java

@@ -0,0 +1,65 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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 org.apache.dubbo.rpc.filter;
+
+import org.apache.dubbo.common.constants.CommonConstants;
+import org.apache.dubbo.common.extension.Activate;
+import org.apache.dubbo.rpc.Filter;
+import org.apache.dubbo.rpc.Invocation;
+import org.apache.dubbo.rpc.Invoker;
+import org.apache.dubbo.rpc.ListenableFilter;
+import org.apache.dubbo.rpc.Result;
+import org.apache.dubbo.rpc.RpcException;
+
+import java.util.List;
+
+@Activate(group = {CommonConstants.PROVIDER, CommonConstants.CONSUMER}, order = -999999)
+public class CallbackRegistrationFilter implements Filter {
+
+    private List<Filter> filters;
+
+    public CallbackRegistrationFilter() {
+    }
+
+    public void setFilters(List<Filter> filters) {
+        this.filters = filters;
+    }
+
+    @Override
+    public Result invoke(Invoker<?> invoker, Invocation invocation) throws RpcException {
+        Result asyncResult = invoker.invoke(invocation);
+
+        asyncResult.thenApplyWithContext(r -> {
+            for (int i = filters.size() - 1; i >= 0; i--) {
+                Filter filter = filters.get(i);
+                // onResponse callback
+                if (filter instanceof ListenableFilter) {
+                    Listener listener = ((ListenableFilter) filter).listener();
+                    if (listener != null) {
+                        listener.onResponse(r, invoker, invocation);
+                    }
+                } else {
+                    filter.onResponse(r, invoker, invocation);
+                }
+            }
+            return r;
+        });
+
+        return asyncResult;
+    }
+
+}

dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/protocol/ProtocolFilterWrapper.java

@@ -27,11 +27,11 @@
 import org.apache.dubbo.rpc.Protocol;
 import org.apache.dubbo.rpc.Result;
 import org.apache.dubbo.rpc.RpcException;
+import org.apache.dubbo.rpc.filter.CallbackRegistrationFilter;
 
 import java.util.List;
 
 import static org.apache.dubbo.common.constants.RegistryConstants.REGISTRY_PROTOCOL;
-
 import static org.apache.dubbo.rpc.Constants.REFERENCE_FILTER_KEY;
 import static org.apache.dubbo.rpc.Constants.SERVICE_FILTER_KEY;
 
@@ -49,12 +49,19 @@ public ProtocolFilterWrapper(Protocol protocol) {
         this.protocol = protocol;
     }
 
+
+
     private static <T> Invoker<T> buildInvokerChain(final Invoker<T> invoker, String key, String group) {
         Invoker<T> last = invoker;
         List<Filter> filters = ExtensionLoader.getExtensionLoader(Filter.class).getActivateExtension(invoker.getUrl(), key, group);
+
         if (!filters.isEmpty()) {
             for (int i = filters.size() - 1; i >= 0; i--) {
                 final Filter filter = filters.get(i);
+                // register callback at CallbackRegistrationFilter
+                if (filter instanceof CallbackRegistrationFilter) {
+                    ((CallbackRegistrationFilter) filter).setFilters(filters);
+                }
                 final Invoker<T> next = last;
                 last = new Invoker<T>() {
 
@@ -88,18 +95,7 @@ public Result invoke(Invocation invocation) throws RpcException {
                             }
                             throw e;
                         }
-                        return asyncResult.thenApplyWithContext(r -> {
-                            // onResponse callback
-                            if (filter instanceof ListenableFilter) {
-                                Filter.Listener listener = ((ListenableFilter) filter).listener();
-                                if (listener != null) {
-                                    listener.onResponse(r, invoker, invocation);
-                                }
-                            } else {
-                                filter.onResponse(r, invoker, invocation);
-                            }
-                            return r;
-                        });
+                        return asyncResult;
                     }
 
                     @Override

dubbo-rpc/dubbo-rpc-api/src/main/resources/META-INF/dubbo/internal/org.apache.dubbo.rpc.Filter

@@ -11,4 +11,5 @@ exception=org.apache.dubbo.rpc.filter.ExceptionFilter
 executelimit=org.apache.dubbo.rpc.filter.ExecuteLimitFilter
 deprecated=org.apache.dubbo.rpc.filter.DeprecatedFilter
 compatible=org.apache.dubbo.rpc.filter.CompatibleFilter
-timeout=org.apache.dubbo.rpc.filter.TimeoutFilter
+timeout=org.apache.dubbo.rpc.filter.TimeoutFilter
+callback-registration=org.apache.dubbo.rpc.filter.CallbackRegistrationFilter