Merge branch 'refs/heads/3.3' into feat/spring_boot_observability

# Conflicts: # dubbo-spring-boot/dubbo-spring-boot-starters/observability/pom.xml
CrazyHZM · Sep 27, 2024 · 88f1fea · 88f1fea
2 parents 6a79d79 + e1cfc03
commit 88f1fea
Show file tree

Hide file tree

Showing 23 changed files with 217 additions and 52 deletions.
diff --git a/dubbo-common/src/main/java/org/apache/dubbo/config/nested/RestConfig.java b/dubbo-common/src/main/java/org/apache/dubbo/config/nested/RestConfig.java
@@ -67,6 +67,11 @@ public class RestConfig implements Serializable {
      */
     private String jsonFramework;
 
+    /**
+     * The disallowed content-types.
+     */
+    private String[] disallowedContentTypes;
+
     /**
      *  The cors configuration.
      */
@@ -133,6 +138,14 @@ public void setJsonFramework(String jsonFramework) {
         this.jsonFramework = jsonFramework;
     }
 
+    public String[] getDisallowedContentTypes() {
+        return disallowedContentTypes;
+    }
+
+    public void setDisallowedContentTypes(String[] disallowedContentTypes) {
+        this.disallowedContentTypes = disallowedContentTypes;
+    }
+
     public CorsConfig getCors() {
         return cors;
     }

diff --git a/.../config/spring6/beans/factory/annotation/ReferenceAnnotationWithAotBeanPostProcessor.java b/.../config/spring6/beans/factory/annotation/ReferenceAnnotationWithAotBeanPostProcessor.java
@@ -261,7 +261,17 @@ private CodeBlock generateStatementForElement(
                 Class<?> c = referenceElement.getInjectedType();
                 AotUtils.registerSerializationForService(c, hints);
                 hints.reflection().registerType(TypeReference.of(c), MemberCategory.INVOKE_PUBLIC_METHODS);
+                // need to enumerate all interfaces by the proxy
+                hints.proxies().registerJdkProxy(c, EchoService.class, Destroyable.class);
                 hints.proxies().registerJdkProxy(c, EchoService.class, Destroyable.class, GenericService.class);
+                hints.proxies()
+                        .registerJdkProxy(
+                                c,
+                                EchoService.class,
+                                Destroyable.class,
+                                SpringProxy.class,
+                                Advised.class,
+                                DecoratingProxy.class);
                 hints.proxies()
                         .registerJdkProxy(
                                 c,

diff --git a/dubbo-demo/dubbo-demo-native/dubbo-demo-native-consumer/pom.xml b/dubbo-demo/dubbo-demo-native/dubbo-demo-native-consumer/pom.xml
@@ -29,7 +29,7 @@
 
   <properties>
     <skip_maven_deploy>true</skip_maven_deploy>
-    <curator5_version>5.1.0</curator5_version>
+    <curator5_version>5.7.0</curator5_version>
     <zookeeper_version>3.8.4</zookeeper_version>
   </properties>
 

diff --git a/dubbo-demo/dubbo-demo-native/dubbo-demo-native-provider/pom.xml b/dubbo-demo/dubbo-demo-native/dubbo-demo-native-provider/pom.xml
@@ -29,7 +29,7 @@
 
   <properties>
     <skip_maven_deploy>true</skip_maven_deploy>
-    <curator5_version>5.1.0</curator5_version>
+    <curator5_version>5.7.0</curator5_version>
     <zookeeper_version>3.8.4</zookeeper_version>
   </properties>
 

diff --git a/dubbo-demo/dubbo-demo-spring-boot/pom.xml b/dubbo-demo/dubbo-demo-spring-boot/pom.xml
@@ -36,7 +36,7 @@
     <skip_maven_deploy>true</skip_maven_deploy>
     <spring-boot.version>2.7.18</spring-boot.version>
     <spring-boot-maven-plugin.version>2.7.18</spring-boot-maven-plugin.version>
-    <micrometer-core.version>1.13.3</micrometer-core.version>
+    <micrometer-core.version>1.13.4</micrometer-core.version>
   </properties>
 
   <dependencyManagement>

diff --git a/dubbo-dependencies-bom/pom.xml b/dubbo-dependencies-bom/pom.xml
@@ -99,7 +99,7 @@
     <httpclient_version>4.5.14</httpclient_version>
     <httpcore_version>4.4.16</httpcore_version>
     <fastjson_version>1.2.83</fastjson_version>
-    <fastjson2_version>2.0.52</fastjson2_version>
+    <fastjson2_version>2.0.53</fastjson2_version>
     <zookeeper_version>3.7.2</zookeeper_version>
     <curator_version>5.7.0</curator_version>
     <curator_test_version>2.12.0</curator_test_version>
@@ -121,9 +121,9 @@
     <commons_lang3_version>3.17.0</commons_lang3_version>
     <envoy_api_version>0.1.35</envoy_api_version>
     <micrometer.version>1.13.4</micrometer.version>
-    <opentelemetry.version>1.39.0</opentelemetry.version>
-    <zipkin-reporter.version>3.4.0</zipkin-reporter.version>
-    <micrometer-tracing.version>1.3.3</micrometer-tracing.version>
+    <opentelemetry.version>1.42.1</opentelemetry.version>
+    <zipkin-reporter.version>3.4.2</zipkin-reporter.version>
+    <micrometer-tracing.version>1.3.4</micrometer-tracing.version>
     <t_digest.version>3.3</t_digest.version>
     <prometheus_client.version>0.16.0</prometheus_client.version>
     <reactive.version>1.0.4</reactive.version>
@@ -136,9 +136,9 @@
     <codehaus-jackson_version>1.9.13</codehaus-jackson_version>
     <tomcat_embed_version>8.5.100</tomcat_embed_version>
     <nacos_version>2.4.2</nacos_version>
-    <sentinel.version>1.8.6</sentinel.version>
+    <sentinel.version>1.8.8</sentinel.version>
     <seata.version>1.8.0</seata.version>
-    <grpc.version>1.66.0</grpc.version>
+    <grpc.version>1.68.0</grpc.version>
     <grpc_contrib_verdion>0.8.1</grpc_contrib_verdion>
     <jprotoc_version>1.2.2</jprotoc_version>
     <mustache_version>0.9.14</mustache_version>
@@ -149,14 +149,14 @@
     <logback_version>1.2.13</logback_version>
     <!-- Fix the bug of log4j refer:https://github.com/apache/logging-log4j2/pull/608 -->
     <log4j2_version>2.24.0</log4j2_version>
-    <commons_io_version>2.16.1</commons_io_version>
-    <commons-codec_version>1.16.0</commons-codec_version>
-    <groovy_version>4.0.21</groovy_version>
+    <commons_io_version>2.17.0</commons_io_version>
+    <commons-codec_version>1.17.1</commons-codec_version>
+    <groovy_version>4.0.23</groovy_version>
 
     <!-- Test libs -->
     <junit_jupiter_version>5.9.3</junit_jupiter_version>
     <junit_version>4.13.2</junit_version>
-    <awaitility_version>4.2.0</awaitility_version>
+    <awaitility_version>4.2.2</awaitility_version>
     <hamcrest_version>2.2</hamcrest_version>
     <cglib_version>2.2.2</cglib_version>
     <embedded_redis_version>1.4.3</embedded_redis_version>
@@ -183,7 +183,7 @@
     <spotless.action>check</spotless.action>
     <dubbo-shared-resources.version>1.0.0</dubbo-shared-resources.version>
     <palantirJavaFormat.version>2.38.0</palantirJavaFormat.version>
-    <open_feign_version>3.1.5</open_feign_version>
+    <open_feign_version>3.1.9</open_feign_version>
     <jakarta.xml.bind-api.version>4.0.2</jakarta.xml.bind-api.version>
     <jaxb-runtime.version>2.4.0-b180830.0438</jaxb-runtime.version>
 

diff --git a/dubbo-maven-plugin/pom.xml b/dubbo-maven-plugin/pom.xml
@@ -75,7 +75,7 @@
     <dependency>
       <groupId>commons-io</groupId>
       <artifactId>commons-io</artifactId>
-      <version>2.16.1</version>
+      <version>2.17.0</version>
     </dependency>
 
     <!-- Takes no effect for this dependency. To notify github dependencies bot to update ${protobuf-java.version} property.  -->

diff --git a/dubbo-plugin/dubbo-compiler/src/main/java/org/apache/dubbo/gen/AbstractGenerator.java b/dubbo-plugin/dubbo-compiler/src/main/java/org/apache/dubbo/gen/AbstractGenerator.java
@@ -210,14 +210,14 @@ private List<PluginProtos.CodeGeneratorResponse.File> generateFiles(List<Service
         return allServiceFiles;
     }
 
-    protected boolean enableMultipleTemplateFiles() {
+    protected boolean useMultipleTemplate(boolean multipleFiles) {
         return false;
     }
 
     private List<PluginProtos.CodeGeneratorResponse.File> buildFile(ServiceContext context) {
         List<PluginProtos.CodeGeneratorResponse.File> files = new ArrayList<>();
 
-        if (context.multipleFiles && enableMultipleTemplateFiles()) {
+        if (useMultipleTemplate(context.multipleFiles)) {
             String content = applyTemplate(getTemplateFileName(), context);
             String dir = absoluteDir(context);
 

diff --git a/...o-plugin/dubbo-compiler/src/main/java/org/apache/dubbo/gen/tri/Dubbo3TripleGenerator.java b/...o-plugin/dubbo-compiler/src/main/java/org/apache/dubbo/gen/tri/Dubbo3TripleGenerator.java
@@ -51,7 +51,7 @@ protected String getSingleTemplateFileName() {
     }
 
     @Override
-    protected boolean enableMultipleTemplateFiles() {
+    protected boolean useMultipleTemplate(boolean multipleFiles) {
         return true;
     }
 }
diff --git a/...ompiler/src/main/java/org/apache/dubbo/gen/tri/reactive/ReactorDubbo3TripleGenerator.java b/...ompiler/src/main/java/org/apache/dubbo/gen/tri/reactive/ReactorDubbo3TripleGenerator.java
@@ -51,7 +51,7 @@ protected String getSingleTemplateFileName() {
     }
 
     @Override
-    protected boolean enableMultipleTemplateFiles() {
+    protected boolean useMultipleTemplate(boolean multipleFiles) {
         return true;
     }
 }
diff --git a/dubbo-plugin/dubbo-compiler/src/main/resources/Dubbo3TripleStub.mustache b/dubbo-plugin/dubbo-compiler/src/main/resources/Dubbo3TripleStub.mustache
@@ -30,6 +30,7 @@ import org.apache.dubbo.rpc.model.MethodDescriptor;
 import org.apache.dubbo.rpc.model.ServiceDescriptor;
 import org.apache.dubbo.rpc.model.StubMethodDescriptor;
 import org.apache.dubbo.rpc.model.StubServiceDescriptor;
+import org.apache.dubbo.rpc.service.Destroyable;
 import org.apache.dubbo.rpc.stub.BiStreamMethodHandler;
 import org.apache.dubbo.rpc.stub.ServerStreamMethodHandler;
 import org.apache.dubbo.rpc.stub.StubInvocationUtil;
@@ -130,13 +131,18 @@ public final class {{className}} {
     {{/biStreamingWithoutClientStreamMethods}}
     }
 
-    public static class {{interfaceClassName}}Stub implements {{interfaceClassName}}{
+    public static class {{interfaceClassName}}Stub implements {{interfaceClassName}}, Destroyable {
         private final Invoker<{{interfaceClassName}}> invoker;
 
         public {{interfaceClassName}}Stub(Invoker<{{interfaceClassName}}> invoker) {
             this.invoker = invoker;
         }
 
+        @Override
+        public void $destroy() {
+              invoker.destroy();
+         }
+
     {{#unaryMethods}}
         {{#javaDoc}}
         {{{javaDoc}}}

diff --git a/...-triple-servlet/src/main/java/org/apache/dubbo/rpc/protocol/tri/servlet/TripleFilter.java b/...-triple-servlet/src/main/java/org/apache/dubbo/rpc/protocol/tri/servlet/TripleFilter.java
@@ -125,8 +125,10 @@ private void handleHttp1(HttpServletRequest request, HttpServletResponse respons
                     channel, ServletExchanger.getUrl(), FrameworkModel.defaultModel());
             channel.setGrpc(false);
             context.setTimeout(resolveTimeout(request, false));
-            listener.onMetadata(new HttpMetadataAdapter(request));
             ServletInputStream is = request.getInputStream();
+            response.getOutputStream().setWriteListener(new TripleWriteListener(channel));
+
+            listener.onMetadata(new HttpMetadataAdapter(request));
             listener.onData(new Http1InputMessage(
                     is.available() == 0 ? StreamUtils.EMPTY : new ByteArrayInputStream(StreamUtils.readBytes(is))));
         } catch (Throwable t) {

diff --git a/...-api/src/main/java/org/apache/dubbo/remoting/api/connection/AbstractConnectionClient.java b/...-api/src/main/java/org/apache/dubbo/remoting/api/connection/AbstractConnectionClient.java
@@ -61,13 +61,17 @@ public final void increase() {
     /**
      * Increments the reference count by 1.
      */
-    public final AbstractConnectionClient retain() {
+    public final boolean retain() {
         long oldCount = COUNTER_UPDATER.getAndIncrement(this);
         if (oldCount <= 0) {
             COUNTER_UPDATER.getAndDecrement(this);
-            throw new AssertionError("This instance has been destroyed");
+            logger.info(
+                    "Retain failed, because connection " + remote
+                            + " has been destroyed but not yet removed, will create a new one instead."
+                            + " Check logs below to confirm that this connection finally gets removed to make sure there's no potential memory leak!");
+            return false;
         }
-        return this;
+        return true;
     }
 
     /**
@@ -77,6 +81,7 @@ public boolean release() {
         long remainingCount = COUNTER_UPDATER.decrementAndGet(this);
 
         if (remainingCount == 0) {
+            logger.info("Destroying connection to {}, because the reference count reaches 0", remote);
             destroy();
             return true;
         } else if (remainingCount <= -1) {

diff --git a/...c/main/java/org/apache/dubbo/remoting/api/connection/SingleProtocolConnectionManager.java b/...c/main/java/org/apache/dubbo/remoting/api/connection/SingleProtocolConnectionManager.java
@@ -17,6 +17,8 @@
 package org.apache.dubbo.remoting.api.connection;
 
 import org.apache.dubbo.common.URL;
+import org.apache.dubbo.common.logger.ErrorTypeAwareLogger;
+import org.apache.dubbo.common.logger.LoggerFactory;
 import org.apache.dubbo.remoting.ChannelHandler;
 import org.apache.dubbo.remoting.Constants;
 import org.apache.dubbo.rpc.model.FrameworkModel;
@@ -26,6 +28,9 @@
 import java.util.function.Consumer;
 
 public class SingleProtocolConnectionManager implements ConnectionManager {
+    private static final ErrorTypeAwareLogger logger =
+            LoggerFactory.getErrorTypeAwareLogger(SingleProtocolConnectionManager.class);
+
     public static final String NAME = "single";
 
     private final ConcurrentMap<String, AbstractConnectionClient> connections = new ConcurrentHashMap<>(16);
@@ -42,21 +47,34 @@ public AbstractConnectionClient connect(URL url, ChannelHandler handler) {
             throw new IllegalArgumentException("url == null");
         }
         return connections.compute(url.getAddress(), (address, conn) -> {
+            String transport = url.getParameter(Constants.TRANSPORTER_KEY, "netty4");
             if (conn == null) {
-                String transport = url.getParameter(Constants.TRANSPORTER_KEY, "netty4");
-                ConnectionManager manager = frameworkModel
-                        .getExtensionLoader(ConnectionManager.class)
-                        .getExtension(transport);
-                final AbstractConnectionClient connectionClient = manager.connect(url, handler);
-                connectionClient.addCloseListener(() -> connections.remove(address, connectionClient));
-                return connectionClient;
+                return createAbstractConnectionClient(url, handler, address, transport);
             } else {
-                conn.retain();
+                boolean shouldReuse = conn.retain();
+                if (!shouldReuse) {
+                    logger.info("Trying to create a new connection for {}.", address);
+                    return createAbstractConnectionClient(url, handler, address, transport);
+                }
                 return conn;
             }
         });
     }
 
+    private AbstractConnectionClient createAbstractConnectionClient(
+            URL url, ChannelHandler handler, String address, String transport) {
+        ConnectionManager manager =
+                frameworkModel.getExtensionLoader(ConnectionManager.class).getExtension(transport);
+        final AbstractConnectionClient connectionClient = manager.connect(url, handler);
+        connectionClient.addCloseListener(() -> {
+            logger.info(
+                    "Remove closed connection (with reference count==0) for address {}, a new one will be created for upcoming RPC requests routing to this address.",
+                    address);
+            connections.remove(address, connectionClient);
+        });
+        return connectionClient;
+    }
+
     @Override
     public void forEachConnection(Consumer<AbstractConnectionClient> connectionConsumer) {
         connections.values().forEach(connectionConsumer);

diff --git a/...oting-http12/src/main/java/org/apache/dubbo/remoting/http12/message/codec/CodecUtils.java b/...oting-http12/src/main/java/org/apache/dubbo/remoting/http12/message/codec/CodecUtils.java
@@ -17,17 +17,24 @@
 package org.apache.dubbo.remoting.http12.message.codec;
 
 import org.apache.dubbo.common.URL;
+import org.apache.dubbo.common.config.Configuration;
+import org.apache.dubbo.common.config.ConfigurationUtils;
 import org.apache.dubbo.common.utils.Assert;
+import org.apache.dubbo.common.utils.StringUtils;
 import org.apache.dubbo.remoting.http12.exception.UnsupportedMediaTypeException;
 import org.apache.dubbo.remoting.http12.message.HttpMessageDecoder;
 import org.apache.dubbo.remoting.http12.message.HttpMessageDecoderFactory;
 import org.apache.dubbo.remoting.http12.message.HttpMessageEncoder;
 import org.apache.dubbo.remoting.http12.message.HttpMessageEncoderFactory;
+import org.apache.dubbo.rpc.Constants;
 import org.apache.dubbo.rpc.model.FrameworkModel;
 
+import java.util.Collections;
+import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
 import java.util.Optional;
+import java.util.Set;
 import java.util.concurrent.ConcurrentHashMap;
 
 public final class CodecUtils {
@@ -37,13 +44,18 @@ public final class CodecUtils {
     private final List<HttpMessageEncoderFactory> encoderFactories;
     private final Map<String, Optional<HttpMessageEncoderFactory>> encoderCache = new ConcurrentHashMap<>();
     private final Map<String, Optional<HttpMessageDecoderFactory>> decoderCache = new ConcurrentHashMap<>();
+    private Set<String> disallowedContentTypes = Collections.emptySet();
 
     public CodecUtils(FrameworkModel frameworkModel) {
         this.frameworkModel = frameworkModel;
         decoderFactories = frameworkModel.getActivateExtensions(HttpMessageDecoderFactory.class);
         encoderFactories = frameworkModel.getActivateExtensions(HttpMessageEncoderFactory.class);
-        decoderFactories.forEach(f -> decoderCache.putIfAbsent(f.mediaType().getName(), Optional.of(f)));
-        encoderFactories.forEach(f -> encoderCache.putIfAbsent(f.mediaType().getName(), Optional.of(f)));
+
+        Configuration configuration = ConfigurationUtils.getGlobalConfiguration(frameworkModel.defaultApplication());
+        String contentTypes = configuration.getString(Constants.H2_SETTINGS_DISALLOWED_CONTENT_TYPES, null);
+        if (contentTypes != null) {
+            disallowedContentTypes = new HashSet<>(StringUtils.tokenizeToList(contentTypes));
+        }
     }
 
     public HttpMessageDecoder determineHttpMessageDecoder(URL url, String mediaType) {
@@ -69,9 +81,10 @@ public HttpMessageEncoder determineHttpMessageEncoder(String mediaType) {
     public Optional<HttpMessageDecoderFactory> determineHttpMessageDecoderFactory(String mediaType) {
         Assert.notNull(mediaType, "mediaType must not be null");
         return decoderCache.computeIfAbsent(mediaType, k -> {
-            for (HttpMessageDecoderFactory decoderFactory : decoderFactories) {
-                if (decoderFactory.supports(k)) {
-                    return Optional.of(decoderFactory);
+            for (HttpMessageDecoderFactory factory : decoderFactories) {
+                if (factory.supports(k)
+                        && !disallowedContentTypes.contains(factory.mediaType().getName())) {
+                    return Optional.of(factory);
                 }
             }
             return Optional.empty();
@@ -81,9 +94,10 @@ public Optional<HttpMessageDecoderFactory> determineHttpMessageDecoderFactory(St
     public Optional<HttpMessageEncoderFactory> determineHttpMessageEncoderFactory(String mediaType) {
         Assert.notNull(mediaType, "mediaType must not be null");
         return encoderCache.computeIfAbsent(mediaType, k -> {
-            for (HttpMessageEncoderFactory encoderFactory : encoderFactories) {
-                if (encoderFactory.supports(k)) {
-                    return Optional.of(encoderFactory);
+            for (HttpMessageEncoderFactory factory : encoderFactories) {
+                if (factory.supports(k)
+                        && !disallowedContentTypes.contains(factory.mediaType().getName())) {
+                    return Optional.of(factory);
                 }
             }
             return Optional.empty();

diff --git a/dubbo-remoting/dubbo-remoting-zookeeper-curator5/pom.xml b/dubbo-remoting/dubbo-remoting-zookeeper-curator5/pom.xml
@@ -30,7 +30,7 @@
   <properties>
     <skip_maven_deploy>false</skip_maven_deploy>
     <zookeeper_version>3.7.2</zookeeper_version>
-    <curator5_version>5.1.0</curator5_version>
+    <curator5_version>5.7.0</curator5_version>
   </properties>
   <dependencies>
     <dependency>

diff --git a/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/Constants.java b/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/Constants.java
@@ -109,6 +109,7 @@ public interface Constants {
     String H2_SETTINGS_BUILTIN_SERVICE_INIT = "dubbo.tri.builtin.service.init";
 
     String H2_SETTINGS_JSON_FRAMEWORK_NAME = "dubbo.protocol.triple.rest.json-framework";
+    String H2_SETTINGS_DISALLOWED_CONTENT_TYPES = "dubbo.protocol.triple.rest.disallowed-content-types";
 
     String H2_SETTINGS_VERBOSE_ENABLED = "dubbo.protocol.triple.verbose";
     String H2_SETTINGS_SERVLET_ENABLED = "dubbo.protocol.triple.servlet.enabled";