PsyncHandler extends AbstractCommandHandler { protected void innerDoHandle(final String[] args, final RedisSlave redisSlave, RedisKeeperServer redisKeeperServer) { KeeperConfig keeperConfig = redisKeeperServer.getKeeperConfig(); KeeperRepl keeperRepl = redisKeeperServer.getKeeperRepl(); Long offsetRequest = Long.valueOf(args[1]); String replIdRequest = args[0]; if(replIdRequest.equals("?")){ doFullSync(redisSlave); }else if(replIdRequest.equals(keeperRepl.replId()) || (replIdRequest.equals(keeperRepl.replId2()) && offsetRequest <= keeperRepl.secondReplIdOffset())){ Long beginOffset = keeperRepl.getBeginOffset(); Long endOffset = keeperRepl.getEndOffset(); if(offsetRequest < beginOffset){ logger.info("[innerDoHandle][offset < beginOffset][begin, end, request]{}, {}, {}" , beginOffset , endOffset, offsetRequest); redisSlave.getRedisKeeperServer().getKeeperMonitor().getKeeperStats().increatePartialSyncError(); doFullSync(redisSlave); }else if(offsetRequest > endOffset +1){ logger.info("[innerDoHandle][wait for offset]{}, {} > {} + 1", redisSlave, offsetRequest, endOffset); waitForoffset(args, redisSlave, keeperRepl.replId(), offsetRequest); }else{ if(endOffset - offsetRequest < keeperConfig.getReplicationStoreMaxCommandsToTransferBeforeCreateRdb()) { doPartialSync(redisSlave, keeperRepl.replId(), offsetRequest); } else { logger.info("[innerDoHandle][too much commands to transfer]{} - {} < {}", endOffset, offsetRequest, keeperConfig.getReplicationStoreMaxCommandsToTransferBeforeCreateRdb()); redisSlave.getRedisKeeperServer().getKeeperMonitor().getKeeperStats().increatePartialSyncError(); doFullSync(redisSlave); } } }else{ logger.info("current repl info: {}", keeperRepl); redisSlave.getRedisKeeperServer().getKeeperMonitor().getKeeperStats().increatePartialSyncError(); doFullSync(redisSlave); } }  @Override String[] getCommands(); static final int WAIT_OFFSET_TIME_MILLI; }
@Test public void testFull(){ String args[] = new String[]{ "?", "-1" }; psyncHandler.innerDoHandle(args, redisSlave, redisKeeperServer); Assert.assertEquals(FULL, function); args = new String[]{ replId, String.valueOf((begin - 1)) }; psyncHandler.innerDoHandle(args, redisSlave, redisKeeperServer); Assert.assertEquals(FULL, function); args = new String[]{ randomString(40), String.valueOf((begin - 1)) }; psyncHandler.innerDoHandle(args, redisSlave, redisKeeperServer); Assert.assertEquals(FULL, function); args = new String[]{ replId2, String.valueOf(secondReplOffset + 1) }; psyncHandler.innerDoHandle(args, redisSlave, redisKeeperServer); Assert.assertEquals(FULL, function); }
@Test public void testPartial(){ String args[] = new String[]{ replId, String.valueOf(begin) }; psyncHandler.innerDoHandle(args, redisSlave, redisKeeperServer); Assert.assertEquals(PARTIAL, function); args = new String[]{ replId, String.valueOf(begin + 1) }; psyncHandler.innerDoHandle(args, redisSlave, redisKeeperServer); Assert.assertEquals(PARTIAL, function); args = new String[]{ replId, String.valueOf(end + 1) }; psyncHandler.innerDoHandle(args, redisSlave, redisKeeperServer); Assert.assertEquals(PARTIAL, function); args = new String[]{ replId2, String.valueOf(secondReplOffset) }; psyncHandler.innerDoHandle(args, redisSlave, redisKeeperServer); Assert.assertEquals(PARTIAL, function); }
@Test public void testWait(){ String args[] = new String[]{ replId, String.valueOf(end + 2) }; psyncHandler.innerDoHandle(args, redisSlave, redisKeeperServer); Assert.assertEquals(WAIT, function); }
@Test public void testTooMuchCommands(){ String args[] = new String[]{ replId, String.valueOf(begin) }; long transfter = end - begin; when(KeeperConfig.getReplicationStoreMaxCommandsToTransferBeforeCreateRdb()).thenReturn(transfter); psyncHandler.innerDoHandle(args, redisSlave, redisKeeperServer); Assert.assertEquals(FULL, function); transfter = end - begin + 1; when(KeeperConfig.getReplicationStoreMaxCommandsToTransferBeforeCreateRdb()).thenReturn(transfter); psyncHandler.innerDoHandle(args, redisSlave, redisKeeperServer); Assert.assertEquals(PARTIAL, function); }