fix: intersect BitList tail invokers in state routing (#16395)

Author: 014-code

dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/router/state/BitList.java

@@ -114,16 +114,26 @@ public synchronized E getByIndex(int index) {
     }
 
     /**
-     * And operation between two bitList. Return a new cloned list.
-     * TailList in source bitList will be totally saved even if it is not appeared in the target bitList.
+     * And operation between two bitList.
      *
      * @param target target bitList
-     * @return this bitList only contains those elements contain in both two list and source bitList's tailList
+     * @return this bitList only contains elements contained in both lists
      */
     public synchronized BitList<E> and(BitList<E> target) {
-        rootSet.and(target.rootSet);
-        if (target.getTailList() != null) {
-            target.getTailList().forEach(this::addToTailList);
+        if (originList == target.originList) {
+            rootSet.and(target.rootSet);
+        } else {
+            BitSet resultSet = new BitSet();
+            for (int bitIndex = rootSet.nextSetBit(0); bitIndex >= 0; bitIndex = rootSet.nextSetBit(bitIndex + 1)) {
+                if (target.contains(originList.get(bitIndex))) {
+                    resultSet.set(bitIndex);
+                }
+            }
+            rootSet.clear();
+            rootSet.or(resultSet);
+        }
+        if (CollectionUtils.isNotEmpty(tailList)) {
+            tailList.removeIf(e -> !target.contains(e));
         }
         return this;
     }

dubbo-cluster/src/test/java/org/apache/dubbo/rpc/cluster/router/condition/ConditionStateRouterTest.java

@@ -196,6 +196,33 @@ void testRoute_matchFilter() {
         Assertions.assertEquals(1, filteredInvokers6.size());
     }
 
+    @Test
+    void testRoute_shouldFilterTailInvokers() {
+        List<Invoker<String>> originInvokers = new ArrayList<Invoker<String>>();
+        Invoker<String> targetInvoker =
+                new MockInvoker<String>(URL.valueOf("dubbo://10.20.3.3:20880/com.foo.BarService"));
+        Invoker<String> tailInvoker =
+                new MockInvoker<String>(URL.valueOf("dubbo://10.20.3.4:20880/com.foo.BarService"));
+        originInvokers.add(targetInvoker);
+        BitList<Invoker<String>> invokers = new BitList<>(originInvokers);
+        invokers.add(tailInvoker);
+
+        StateRouter<String> router = new ConditionStateRouterFactory()
+                .getRouter(
+                        String.class, getRouteUrl("=> host = 10.20.3.3").addParameter(FORCE_KEY, String.valueOf(true)));
+
+        BitList<Invoker<String>> filteredInvokers = router.route(
+                invokers,
+                URL.valueOf("consumer://" + LOCAL_HOST + "/com.foo.BarService"),
+                new RpcInvocation(),
+                false,
+                new Holder<>());
+
+        Assertions.assertEquals(1, filteredInvokers.size());
+        Assertions.assertTrue(filteredInvokers.contains(targetInvoker));
+        Assertions.assertFalse(filteredInvokers.contains(tailInvoker));
+    }
+
     @Test
     void testRoute_methodRoute() {
         Invocation invocation = new RpcInvocation("getFoo", "com.foo.BarService", "", new Class<?>[0], new Object[0]);

dubbo-cluster/src/test/java/org/apache/dubbo/rpc/cluster/router/state/BitListTest.java

@@ -96,10 +96,29 @@ void testIntersect() {
 
         aBitList.add("D");
         intersectBitList = aBitList.and(bBitList);
-        Assertions.assertEquals(3, intersectBitList.size());
+        Assertions.assertEquals(2, intersectBitList.size());
         Assertions.assertEquals(totalList.get(0), intersectBitList.get(0));
         Assertions.assertEquals(totalList.get(1), intersectBitList.get(1));
-        Assertions.assertEquals("D", intersectBitList.get(2));
+        Assertions.assertFalse(intersectBitList.contains("D"));
+    }
+
+    @Test
+    void testIntersectTailList() {
+        List<String> list = Arrays.asList("A", "B");
+        BitList<String> bitList = new BitList<>(list);
+        bitList.add("C");
+        bitList.add("D");
+
+        BitList<String> target = new BitList<>(list);
+        target.remove("B");
+        target.add("D");
+
+        BitList<String> intersectBitList = bitList.and(target);
+        Assertions.assertEquals(2, intersectBitList.size());
+        Assertions.assertEquals("A", intersectBitList.get(0));
+        Assertions.assertEquals("D", intersectBitList.get(1));
+        Assertions.assertFalse(intersectBitList.contains("B"));
+        Assertions.assertFalse(intersectBitList.contains("C"));
     }
 
     @Test