Merge pull request #883 from ctripcorp/bugfix/hailu_repair_fullgc

repair Increment full gc

redis/redis-keeper/src/main/java/com/ctrip/xpipe/redis/keeper/applier/command/SequenceCommand.java

@@ -11,7 +11,6 @@
 
 /**
  * @author Slight
- *
  * Make sure to execute this command on stateThread.
  * Jan 31, 2022 1:06 PM
  */
@@ -44,7 +43,7 @@ public SequenceCommand(Collection<SequenceCommand<?>> pasts, Command<V> inner, E
 
     private void executeSelf() {
         CommandFuture<V> future = inner.execute(workerThreads);
-        future.addListener((f)->{
+        future.addListener((f) -> {
             if (f.isSuccess()) {
                 stateThread.execute(() -> {
                     try {
@@ -65,10 +64,13 @@ private void executeSelf() {
     }
 
     private void nextAfter(Collection<SequenceCommand<?>> pasts) {
-        for (SequenceCommand<?> past : pasts) {
-            past.future().addListener((f)->{
+        final int size = pasts.size();
+        List<SequenceCommand> temp = Lists.newArrayList(pasts);
+        for (SequenceCommand<?> past : temp) {
+            past.future().addListener((f) -> {
                 if (f.isSuccess()) {
-                    if (++complete == pasts.size()) {
+                    pasts.remove(past);
+                    if (++complete == size) {
                         executeSelf();
                     }
                 } else {

redis/redis-keeper/src/main/java/com/ctrip/xpipe/redis/keeper/applier/sequence/DefaultSequenceController.java

@@ -154,8 +154,8 @@ private void submitNoneKeyCommand(RedisOpCommand<?> command, long commandOffset)
 
         /* do some stuff when finish */
 
-        releaseMemoryThresholdWhenDone(current, command.redisOp().estimatedSize());
         increaseOffsetWhenSuccess(current, commandOffset);
+        releaseMemoryThresholdWhenDone(current, command.redisOp().estimatedSize());
 
         /* run self */
 
@@ -189,8 +189,8 @@ private void submitSingleKeyCommand(RedisOpDataCommand<?> command, long commandO
 
         /* do some stuff when finish */
 
-        releaseMemoryThresholdWhenDone(current, command.redisOp().estimatedSize());
         increaseOffsetWhenSuccess(current, commandOffset);
+        releaseMemoryThresholdWhenDone(current, command.redisOp().estimatedSize());
 
         /* run self */
 
@@ -212,8 +212,8 @@ private void submitMultiKeyCommand(RedisOpDataCommand<?> command, long commandOf
             forgetWhenDone(current, key);
         }
 
-        releaseMemoryThresholdWhenDone(current, command.redisOp().estimatedSize());
         increaseOffsetWhenSuccess(current, commandOffset);
+        releaseMemoryThresholdWhenDone(current, command.redisOp().estimatedSize());
 
         current.execute();
     }
@@ -240,8 +240,8 @@ private void submitObstacle(RedisOpCommand<?> command, long commandOffset) {
 
         /* do some stuff when finish */
 
-        releaseMemoryThresholdWhenDone(current, command.redisOp().estimatedSize());
         increaseOffsetWhenSuccess(current, commandOffset);
+        releaseMemoryThresholdWhenDone(current, command.redisOp().estimatedSize());
 
         /* run self */
 
@@ -266,7 +266,7 @@ private void forgetWhenDone(SequenceCommand<?> sequenceCommand, RedisKey key) {
 
 
     private void releaseMemoryThresholdWhenDone(SequenceCommand<?> sequenceCommand, long memory) {
-        sequenceCommand.future().addListener((f)->{
+        sequenceCommand.future().addListener((f) -> {
             concurrencyThreshold.release();
             memoryThreshold.release(memory);
         });

redis/redis-keeper/src/main/java/com/ctrip/xpipe/redis/keeper/applier/sync/DefaultCommandDispatcher.java

@@ -128,6 +128,7 @@ public void onRdbData(ByteBuf rdbData) {
 
     @Override
     public void endReadRdb(EofType eofType, GtidSet emptyGtidSet, long rdbOffset) {
+        logger.info("[endReadRdb] eofType={}", eofType);
     }
 
     @Override

redis/redis-keeper/src/test/java/com/ctrip/xpipe/redis/keeper/AbstractFakeRedisTest.java

@@ -86,7 +86,7 @@ protected RedisKeeperServer startRedisKeeperServer(int replicationStoreCommandFi
 				commandFileSize, 
 				replicationStoreCommandFileNumToKeep, 
 				replicationStoreMaxCommandsToTransferBeforeCreateRdb, minTimeMilliToGcAfterCreate);
-
+        ((TestKeeperConfig)keeperConfig).setRdbDumpMinIntervalMilli(0);
 		RedisKeeperServer redisKeeperServer = createRedisKeeperServer(keeperConfig);
 		redisKeeperServer.initialize();
 		redisKeeperServer.start();

redis/redis-keeper/src/test/java/com/ctrip/xpipe/redis/keeper/impl/fakeredis/DefaultRedisKeeperServerConnectToFakeRedisTest.java

@@ -42,7 +42,7 @@ public void testReplicationData() throws Exception{
 	@Test
 	public void testNewDumpCommandsLost() throws Exception{
 
-		startKeeperServerAndTestReFullSync(2, allCommandsSize);
+		startKeeperServerAndTestReFullSync(1, allCommandsSize);
 	}
 
 	@Test