fix applier ut down

ctripcorp · Nov 21, 2023 · e73cc7e · e73cc7e
1 parent 70cba87
commit e73cc7e
Show file tree

Hide file tree

Showing 3 changed files with 7 additions and 5 deletions.
diff --git a/...rc/main/java/com/ctrip/xpipe/redis/keeper/applier/sequence/DefaultSequenceController.java b/...rc/main/java/com/ctrip/xpipe/redis/keeper/applier/sequence/DefaultSequenceController.java
@@ -69,7 +69,7 @@ public class DefaultSequenceController extends AbstractInstanceComponent impleme
 
     public static final long DEFAULT_CONCURRENCY_THRESHOLD = 10000;
 
-    private static final CloseState closeState = new CloseState();
+    private final CloseState closeState = new CloseState();
 
     public DefaultSequenceController() {
         this(DEFAULT_QPS_THRESHOLD, DEFAULT_BYTES_PER_SECOND_THRESHOLD, DEFAULT_MEMORY_THRESHOLD, DEFAULT_CONCURRENCY_THRESHOLD);

diff --git a/...er/src/main/java/com/ctrip/xpipe/redis/keeper/applier/xsync/DefaultCommandDispatcher.java b/...er/src/main/java/com/ctrip/xpipe/redis/keeper/applier/xsync/DefaultCommandDispatcher.java
@@ -293,7 +293,7 @@ protected boolean shouldFilter(RedisOp redisOp) {
                 logger.warn("publish command {} length={} unexpected, filtered", redisOp, length);
                 return true;
             }
-            if (!channel.startsWith("xpipe-hetero-")) {
+            if (!channel.startsWith("xpipe-asymmetric-")) {
                 logger.warn("publish command {} channel: [{}] filtered", redisOp, channel);
                 return true;
             }

diff --git a/...est/java/com/ctrip/xpipe/redis/keeper/applier/sequence/DefaultSequenceControllerTest.java b/...est/java/com/ctrip/xpipe/redis/keeper/applier/sequence/DefaultSequenceControllerTest.java
@@ -14,6 +14,7 @@
 
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
 import java.util.concurrent.atomic.AtomicLong;
 
@@ -29,10 +30,11 @@
  */
 public class DefaultSequenceControllerTest extends AbstractTest {
 
-    DefaultSequenceController controller = new DefaultSequenceController();
+    DefaultSequenceController controller;
 
     @Before
     public void setUp() throws Exception {
+        controller = new DefaultSequenceController();
         controller.stateThread = Executors.newScheduledThreadPool(1,
                 ClusterShardAwareThreadFactory.create("test-cluster", "test-shard", "state-test-thread"));
         controller.workerThreads = Executors.newScheduledThreadPool(8,
@@ -79,8 +81,8 @@ public void multiKeyCommand() throws ExecutionException, InterruptedException, T
 
         controller.submit(new TestMultiDataCommandWrapper(command, executors, Lists.newArrayList(first, second)), 0);
 
-        first.future().get();
-        second.future().get();
+        first.future().get(5, TimeUnit.SECONDS);
+        second.future().get(5, TimeUnit.SECONDS);
 
         waitConditionUntilTimeOut(()->lwmManager.lastSubmitTime >= first.endTime);
         waitConditionUntilTimeOut(()->lwmManager.lastSubmitTime >= second.endTime);