add swallow info log

ctripcorp · Jan 23, 2024 · 1959955 · 1959955
1 parent d3278d9
commit 1959955
Show file tree

Hide file tree

Showing 2 changed files with 18 additions and 3 deletions.
diff --git a/...-core/src/test/java/com/ctrip/xpipe/redis/core/redis/parser/GeneralRedisOpParserTest.java b/...-core/src/test/java/com/ctrip/xpipe/redis/core/redis/parser/GeneralRedisOpParserTest.java
@@ -68,6 +68,11 @@ public void testUnknowParse() {
         Assert.assertEquals(RedisOpType.UNKNOWN, redisOp.getOpType());
         Assert.assertNull(redisOp.getOpGtid());
         Assert.assertArrayEquals(rawOpArgs, redisOp.buildRawOpArgs());
+        byte[][] rawOpArgs1 = {"credis_flushall".getBytes()};
+        redisOp = parser.parse(rawOpArgs1);
+        Assert.assertEquals(RedisOpType.UNKNOWN, redisOp.getOpType());
+        Assert.assertNull(redisOp.getOpGtid());
+        Assert.assertArrayEquals(rawOpArgs1, redisOp.buildRawOpArgs());
 
         RedisSingleKeyOp redisSingleKeyOp = (RedisSingleKeyOp) redisOp;
         Assert.assertNull(redisSingleKeyOp.getKey());

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
@@ -301,8 +301,8 @@ protected boolean shouldFilter(RedisOp redisOp) {
             }
         }
         if (redisOp.getOpType().isSwallow()) {
-            logger.info("[onRedisOp] swallow redisOp: {}", redisOp);
-            EventMonitor.DEFAULT.logEvent("APPLIER.SWALLOW.OP", String.format("swallow redisOp %s", redisOp.toString()));
+            logger.info("[onRedisOp] swallow redisOp: {}", redisOp.toString());
+            EventMonitor.DEFAULT.logEvent("APPLIER.SWALLOW.OP",  new String(redisOp.buildRawOpArgs()[0]));
         }
         return redisOp.getOpType().isSwallow();
     }
@@ -333,7 +333,7 @@ private void doOnRedisOp(RedisOp redisOp, long commandOffsetToAccumulate) {
         }
 
         if (shouldFilter(redisOp)) {
-            gtid_executed.get().add(redisOp.getOpGtid());
+            addLwm(redisOp);
             offsetRecorder.addAndGet(commandOffsetToAccumulate);
             return;
         }
@@ -349,6 +349,16 @@ private void doOnRedisOp(RedisOp redisOp, long commandOffsetToAccumulate) {
         }
     }
 
+    private void addLwm(RedisOp redisOp) {
+        stateThread.execute(() -> {
+            try {
+                gtid_executed.get().add(redisOp.getOpGtid());
+            } catch (Throwable t) {
+                logger.error("[onRedisOp] unlikely - error", t);
+            }
+        });
+    }
+
     @Override
     public void onRedisOp(RedisOp redisOp) {
         doOnRedisOp(redisOp, 0);