CR: Gsoc routing support

changes/en-us/2.x.md

@@ -20,7 +20,7 @@ Add changes here for all PR submitted to the 2.x branch.
 
 ### feature:
 
-- [[#PR_NO](https://github.com/seata/seata/pull/PR_NO)] support XXX
+- [[#7533](https://github.com/apache/incubator-seata/pull/7533)] support metadata-based registration and discovery, support namingserver type
 
 
 ### bugfix:

changes/zh-cn/2.x.md

@@ -19,8 +19,7 @@
 <!-- 请根据PR的类型添加 `变更记录` 到以下对应位置(feature/bugfix/optimize/test) 下 -->
 
 ### feature:
-
-- [[#PR_NO](https://github.com/seata/seata/pull/PR_NO)] 支持 XXX
+- [[#7533](https://github.com/apache/incubator-seata/pull/7533)] 支持元数据注册与发现能力并支持namingserver类型
 
 
 ### bugfix:

common/src/main/java/org/apache/seata/common/metadata/ServiceInstance.java

@@ -0,0 +1,135 @@
+/*
+ * 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.seata.common.metadata;
+
+import org.apache.seata.common.util.NetUtil;
+
+import java.net.InetSocketAddress;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Set;
+
+/**
+ * entity for packaging inetSocketAddress and metadata for loadBalance
+ */
+public class ServiceInstance {
+    private InetSocketAddress address;
+    private Map<String, Object> metadata;
+
+    public ServiceInstance(InetSocketAddress address, Map<String, Object> metadata) {
+        this.address = address;
+        this.metadata = metadata;
+    }
+
+    public ServiceInstance(Instance instance) {
+        this.address = new InetSocketAddress(
+                instance.getTransaction().getHost(), instance.getTransaction().getPort());
+        this.metadata = instance.getMetadata();
+    }
+
+    public ServiceInstance(InetSocketAddress address) {
+        this.address = address;
+    }
+
+    public InetSocketAddress getAddress() {
+        return address;
+    }
+
+    public void setAddress(InetSocketAddress address) {
+        this.address = address;
+    }
+
+    public Map<String, Object> getMetadata() {
+        return metadata;
+    }
+
+    public void setMetadata(Map<String, Object> metadata) {
+        this.metadata = metadata;
+    }
+
+    /**
+     * Converts a list of InetSocketAddress to a list of ServiceInstance.
+     * @param addresses list of InetSocketAddress
+     * @return list of ServiceInstance
+     */
+    public static List<ServiceInstance> convertToServiceInstanceSet(List<InetSocketAddress> addresses) {
+        List<ServiceInstance> serviceInstances = new ArrayList<>();
+        if (addresses != null && !addresses.isEmpty()) {
+            for (InetSocketAddress address : addresses) {
+                NetUtil.validAddress(address);
+                serviceInstances.add(new ServiceInstance(address, null));
+            }
+        }
+        return serviceInstances;
+    }
+
+    /**
+     * Converts a set of InetSocketAddress to a set of ServiceInstance in RedisRegistryServiceImpl.
+     * @param addresses set of InetSocketAddress
+     * @return set of ServiceInstance
+     */
+    public static Set<ServiceInstance> convertToServiceInstanceSet(Set<InetSocketAddress> addresses) {
+        Set<ServiceInstance> serviceInstances = new HashSet<>();
+        if (addresses != null && !addresses.isEmpty()) {
+            for (InetSocketAddress address : addresses) {
+                NetUtil.validAddress(address);
+                serviceInstances.add(new ServiceInstance(address, null));
+            }
+        }
+        return serviceInstances;
+    }
+
+    /**
+     * Creates a ServiceInstance from an InetSocketAddress and a Map<String, String> of metadata.
+     * @param address the InetSocketAddress
+     * @param stringMap the map of string metadata
+     * @return a new ServiceInstance
+     */
+    public static ServiceInstance fromStringMap(InetSocketAddress address, Map<String, String> stringMap) {
+        Map<String, Object> metadata = new HashMap<>();
+        if (stringMap != null) {
+            metadata.putAll(stringMap);
+        }
+        return new ServiceInstance(address, metadata);
+    }
+
+    @Override
+    public boolean equals(Object o) {
+        if (this == o) {
+            return true;
+        }
+        if (o == null || getClass() != o.getClass()) {
+            return false;
+        }
+        ServiceInstance that = (ServiceInstance) o;
+        return Objects.equals(address, that.address) && Objects.equals(metadata, that.metadata);
+    }
+
+    @Override
+    public int hashCode() {
+        return Objects.hash(address, metadata);
+    }
+
+    @Override
+    public String toString() {
+        return "ServiceInstance{" + "address=" + address + ", metadata=" + metadata + '}';
+    }
+}

common/src/test/java/org/apache/seata/common/metadata/ServiceInstanceTest.java

@@ -0,0 +1,123 @@
+/*
+ * 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.seata.common.metadata;
+
+import org.junit.jupiter.api.Test;
+
+import java.net.InetSocketAddress;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+public class ServiceInstanceTest {
+
+    private final InetSocketAddress address1 = new InetSocketAddress("127.0.0.1", 8091);
+    private final InetSocketAddress address2 = new InetSocketAddress("127.0.0.1", 8092);
+
+    @Test
+    public void testConstructorAndGetters() {
+        Map<String, Object> metadata = new HashMap<>();
+        metadata.put("key1", "value1");
+
+        ServiceInstance instance1 = new ServiceInstance(address1, metadata);
+
+        assertEquals(address1, instance1.getAddress());
+        assertEquals(metadata, instance1.getMetadata());
+
+        Instance instance = Instance.getInstance();
+        instance.setTransaction(new Node.Endpoint("127.0.0.1", 8093));
+
+        ServiceInstance instance2 = new ServiceInstance(Instance.getInstance());
+
+        assertEquals(
+                Instance.getInstance().getTransaction().getHost(),
+                instance2.getAddress().getAddress().getHostAddress());
+        assertEquals(
+                Instance.getInstance().getTransaction().getPort(),
+                instance2.getAddress().getPort());
+
+        instance.setTransaction(null); // clean up after test
+    }
+
+    @Test
+    public void testConvertToServiceInstanceList() {
+        List<InetSocketAddress> addresses = new ArrayList<>();
+        addresses.add(address1);
+        addresses.add(address2);
+
+        List<ServiceInstance> serviceInstances = ServiceInstance.convertToServiceInstanceSet(addresses);
+
+        assertEquals(2, serviceInstances.size());
+        assertEquals(address1, serviceInstances.get(0).getAddress());
+        assertEquals(address2, serviceInstances.get(1).getAddress());
+    }
+
+    @Test
+    public void testConvertToServiceInstanceSet() {
+        Set<InetSocketAddress> addresses = new HashSet<>();
+        addresses.add(address1);
+        addresses.add(address2);
+
+        Set<ServiceInstance> serviceInstances = ServiceInstance.convertToServiceInstanceSet(addresses);
+
+        assertEquals(2, serviceInstances.size());
+    }
+
+    @Test
+    public void testSetAddressAndSetMetadata() {
+        ServiceInstance instance = new ServiceInstance(address1);
+        instance.setAddress(address2);
+        instance.setMetadata(new HashMap<>());
+
+        assertEquals(address2, instance.getAddress());
+        assertNotNull(instance.getMetadata());
+    }
+
+    @Test
+    public void testFromStringMap() {
+        Map<String, String> stringMap = new HashMap<>();
+        stringMap.put("stringKey", "stringValue");
+
+        ServiceInstance instance = ServiceInstance.fromStringMap(address1, stringMap);
+
+        assertEquals(address1, instance.getAddress());
+        assertNotNull(instance.getMetadata());
+        assertEquals("stringValue", instance.getMetadata().get("stringKey"));
+    }
+
+    @Test
+    public void testEqualsAndToString() {
+        ServiceInstance instance1 = new ServiceInstance(address1);
+        ServiceInstance instance2 = new ServiceInstance(address1);
+        ServiceInstance instance3 = new ServiceInstance(address2);
+
+        assertTrue(instance1.equals(instance2));
+        assertTrue(instance1.equals(instance1));
+        assertFalse(instance1.equals(instance3));
+        assertFalse(instance1.equals("string"));
+
+        assertTrue(instance1.toString().contains("8091"));
+    }
+}

core/src/main/java/org/apache/seata/core/rpc/netty/AbstractNettyRemotingClient.java

@@ -27,6 +27,7 @@
 import io.netty.handler.timeout.IdleStateEvent;
 import org.apache.seata.common.exception.FrameworkErrorCode;
 import org.apache.seata.common.exception.FrameworkException;
+import org.apache.seata.common.metadata.ServiceInstance;
 import org.apache.seata.common.thread.NamedThreadFactory;
 import org.apache.seata.common.util.CollectionUtils;
 import org.apache.seata.common.util.NetUtil;
@@ -48,6 +49,7 @@
 import org.apache.seata.core.rpc.processor.RemotingProcessor;
 import org.apache.seata.discovery.loadbalance.LoadBalanceFactory;
 import org.apache.seata.discovery.registry.RegistryFactory;
+import org.apache.seata.discovery.routing.RoutingManager;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -287,9 +289,13 @@ protected String loadBalance(String transactionServiceGroup, Object msg) {
         InetSocketAddress address = null;
         try {
             @SuppressWarnings("unchecked")
-            List<InetSocketAddress> inetSocketAddressList =
+            List<ServiceInstance> serviceInstances =
                     RegistryFactory.getInstance().aliveLookup(transactionServiceGroup);
-            address = this.doSelect(inetSocketAddressList, msg);
+
+            // Apply routing filter
+            serviceInstances = applyRoutingFilter(serviceInstances, transactionServiceGroup, msg);
+
+            address = this.doSelect(serviceInstances, msg);
         } catch (Exception ex) {
             LOGGER.error("Select the address failed: {}", ex.getMessage());
         }
@@ -299,12 +305,38 @@ protected String loadBalance(String transactionServiceGroup, Object msg) {
         return NetUtil.toStringAddress(address);
     }
 
-    protected InetSocketAddress doSelect(List<InetSocketAddress> list, Object msg) throws Exception {
+    /**
+     * Apply routing filter
+     * @param serviceInstances original service instances list
+     * @param transactionServiceGroup transaction service group
+     * @param msg message object
+     * @return filtered service instances list
+     */
+    private List<ServiceInstance> applyRoutingFilter(
+            List<ServiceInstance> serviceInstances, String transactionServiceGroup, Object msg) {
+        try {
+            // Get routing manager
+            RoutingManager routingManager = RoutingManager.getInstance();
+
+            // Get transaction ID
+            String xid = getXid(msg);
+
+            // Execute routing filter
+            return routingManager.filter(serviceInstances);
+        } catch (Exception e) {
+            LOGGER.warn("Routing filter failed, using original service instances: {}", e.getMessage());
+            return serviceInstances;
+        }
+    }
+
+    protected InetSocketAddress doSelect(List<ServiceInstance> list, Object msg) throws Exception {
         if (CollectionUtils.isNotEmpty(list)) {
             if (list.size() > 1) {
-                return LoadBalanceFactory.getInstance().select(list, getXid(msg));
+                return LoadBalanceFactory.getInstance()
+                        .select(list, getXid(msg))
+                        .getAddress();
             } else {
-                return list.get(0);
+                return list.get(0).getAddress();
             }
         }
         return null;