Merge branch 'apache-3.2' into apache-3.3

Author: AlbumenJ

dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/client/ServiceDiscoveryRegistryDirectory.java

@@ -343,6 +343,17 @@ private void refreshInvoker(List<URL> invokerUrls) {
                 return;
             }
 
+            int originSize = invokerUrls.size();
+            invokerUrls = invokerUrls.stream().distinct().collect(Collectors.toList());
+            if (invokerUrls.size() != originSize) {
+                logger.info("Received duplicated invoker urls changed event from registry. "
+                        + "Registry type: instance. "
+                        + "Service Key: "
+                        + getConsumerUrl().getServiceKey() + ". "
+                        + "Notify Urls Size : " + originSize + ". "
+                        + "Distinct Urls Size: " + invokerUrls.size() + ".");
+            }
+
             // use local reference to avoid NPE as this.urlInvokerMap will be set null concurrently at
             // destroyAllInvokers().
             Map<ProtocolServiceKeyWithAddress, Invoker<T>> localUrlInvokerMap = this.urlInvokerMap;

dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/integration/RegistryDirectory.java

@@ -310,6 +310,17 @@ private void refreshInvoker(List<URL> invokerUrls) {
                 return;
             }
 
+            int originSize = invokerUrls.size();
+            invokerUrls = invokerUrls.stream().distinct().collect(Collectors.toList());
+            if (invokerUrls.size() != originSize) {
+                logger.info("Received duplicated invoker urls changed event from registry. "
+                        + "Registry type: interface. "
+                        + "Service Key: "
+                        + getConsumerUrl().getServiceKey() + ". "
+                        + "Notify Urls Size : " + originSize + ". "
+                        + "Distinct Urls Size: " + invokerUrls.size() + ".");
+            }
+
             // use local reference to avoid NPE as this.urlInvokerMap will be set null concurrently at
             // destroyAllInvokers().
             Map<URL, Invoker<T>> localUrlInvokerMap = this.urlInvokerMap;

dubbo-remoting/dubbo-remoting-netty4/src/main/java/org/apache/dubbo/remoting/transport/netty4/NettyPortUnificationServerHandler.java

@@ -20,6 +20,7 @@
 import org.apache.dubbo.common.io.Bytes;
 import org.apache.dubbo.common.logger.ErrorTypeAwareLogger;
 import org.apache.dubbo.common.logger.LoggerFactory;
+import org.apache.dubbo.common.ssl.AuthPolicy;
 import org.apache.dubbo.common.ssl.CertManager;
 import org.apache.dubbo.common.ssl.ProviderCert;
 import org.apache.dubbo.remoting.ChannelHandler;
@@ -47,6 +48,7 @@
 import io.netty.handler.ssl.SslHandshakeCompletionEvent;
 import io.netty.util.AttributeKey;
 
+import static org.apache.dubbo.common.constants.LoggerCodeConstants.CONFIG_SSL_CONNECT_INSECURE;
 import static org.apache.dubbo.common.constants.LoggerCodeConstants.INTERNAL_ERROR;
 
 public class NettyPortUnificationServerHandler extends ByteToMessageDecoder {
@@ -122,8 +124,27 @@ protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) t
         ProviderCert providerConnectionConfig =
                 certManager.getProviderConnectionConfig(url, ctx.channel().remoteAddress());
 
-        if (providerConnectionConfig != null && isSsl(in)) {
-            enableSsl(ctx, providerConnectionConfig);
+        if (providerConnectionConfig != null && canDetectSsl(in)) {
+            if (isSsl(in)) {
+                enableSsl(ctx, providerConnectionConfig);
+            } else {
+                // check server should load TLS or not
+                if (providerConnectionConfig.getAuthPolicy() != AuthPolicy.NONE) {
+                    byte[] preface = new byte[in.readableBytes()];
+                    in.readBytes(preface);
+                    LOGGER.error(
+                            CONFIG_SSL_CONNECT_INSECURE,
+                            "client request server without TLS",
+                            "",
+                            String.format(
+                                    "Downstream=%s request without TLS preface, but server require it. " + "preface=%s",
+                                    ctx.channel().remoteAddress(), Bytes.bytes2hex(preface)));
+
+                    // Untrusted connection; discard everything and close the connection.
+                    in.clear();
+                    ctx.close();
+                }
+            }
         } else {
             detectProtocol(ctx, url, channel, in);
         }
@@ -150,6 +171,11 @@ protected void configurePipeline(ChannelHandlerContext ctx, String protocol) thr
         p.remove(this);
     }
 
+    private boolean canDetectSsl(ByteBuf buf) {
+        // at least 5 bytes to determine if data is encrypted
+        return detectSsl && buf.readableBytes() >= 5;
+    }
+
     private boolean isSsl(ByteBuf buf) {
         // at least 5 bytes to determine if data is encrypted
         if (detectSsl && buf.readableBytes() >= 5) {