@Test public void testConcurrentIncreaseTokenSize() { when(redisKeeperServer.getRedisKeeperServerState()).thenReturn(new RedisKeeperServerStateActive(redisKeeperServer)); when(redisMaster.isKeeper()).thenReturn(true); when(keeperConfig.getLeakyBucketInitSize()).thenReturn(3); CompositeLeakyBucket leakyBucket = new CompositeLeakyBucket(keeperConfig, metaServerKeeperService, keeperContainerService); when(resourceManager.getLeakyBucket()).thenReturn(leakyBucket); listener = new LeakyBucketBasedMasterReplicationListener(redisMasterReplication, redisKeeperServer, resourceManager, scheduled); for(int i = 0; i < keeperConfig.getLeakyBucketInitSize(); i++) { Assert.assertTrue(listener.canSendPsync()); } Assert.assertFalse(listener.canSendPsync()); when(keeperConfig.getMetaServerAddress()).thenReturn("http: MetaServerKeeperService.KeeperContainerTokenStatusResponse response = new MetaServerKeeperService .KeeperContainerTokenStatusResponse(keeperConfig.getLeakyBucketInitSize() + 1); when(metaServerKeeperService.refreshKeeperContainerTokenStatus(any())).thenReturn(response); leakyBucket.refresh(); Assert.assertEquals(keeperConfig.getLeakyBucketInitSize() + 1, leakyBucket.getTotalSize()); Assert.assertTrue(listener.canSendPsync()); Assert.assertFalse(listener.canSendPsync()); }
@Override public boolean canSendPsync() { if (redisMasterReplication.redisMaster().isKeeper() && (redisKeeperServer.getRedisKeeperServerState() instanceof RedisKeeperServerStateActive)) { if(replDownShortly()) { logger.info("[canSendPsync][repl down short] let psync pass"); return true; } KeeperStats keeperStats = redisKeeperServer.getKeeperMonitor().getKeeperStats(); if(keeperStats.getLastPsyncFailReason() != null && keeperStats.getLastPsyncFailReason() != PsyncFailReason.TOKEN_LACK && psyncEverSucceed.get() != INIT_STATE && !isPsyncEverSucceed()) { logger.warn("[canSendPsync]never succeed, let it psync, {}", psyncEverSucceed.get()); return true; } if(resourceManager.getLeakyBucket().tryAcquire()) { holdToken.set(true); recordPsyncSendTime(); } else { logger.warn("[canSendPsync]psync wont send as no token is available [port:{}]", redisKeeperServer.getListeningPort()); keeperStats.setLastPsyncFailReason(PsyncFailReason.TOKEN_LACK); keeperStats.increasePsyncSendFail(); return false; } } return true; }
LeakyBucketBasedMasterReplicationListener implements RedisMasterReplication.RedisMasterReplicationObserver { @Override public boolean canSendPsync() { if (redisMasterReplication.redisMaster().isKeeper() && (redisKeeperServer.getRedisKeeperServerState() instanceof RedisKeeperServerStateActive)) { if(replDownShortly()) { logger.info("[canSendPsync][repl down short] let psync pass"); return true; } KeeperStats keeperStats = redisKeeperServer.getKeeperMonitor().getKeeperStats(); if(keeperStats.getLastPsyncFailReason() != null && keeperStats.getLastPsyncFailReason() != PsyncFailReason.TOKEN_LACK && psyncEverSucceed.get() != INIT_STATE && !isPsyncEverSucceed()) { logger.warn("[canSendPsync]never succeed, let it psync, {}", psyncEverSucceed.get()); return true; } if(resourceManager.getLeakyBucket().tryAcquire()) { holdToken.set(true); recordPsyncSendTime(); } else { logger.warn("[canSendPsync]psync wont send as no token is available [port:{}]", redisKeeperServer.getListeningPort()); keeperStats.setLastPsyncFailReason(PsyncFailReason.TOKEN_LACK); keeperStats.increasePsyncSendFail(); return false; } } return true; } }
LeakyBucketBasedMasterReplicationListener implements RedisMasterReplication.RedisMasterReplicationObserver { @Override public boolean canSendPsync() { if (redisMasterReplication.redisMaster().isKeeper() && (redisKeeperServer.getRedisKeeperServerState() instanceof RedisKeeperServerStateActive)) { if(replDownShortly()) { logger.info("[canSendPsync][repl down short] let psync pass"); return true; } KeeperStats keeperStats = redisKeeperServer.getKeeperMonitor().getKeeperStats(); if(keeperStats.getLastPsyncFailReason() != null && keeperStats.getLastPsyncFailReason() != PsyncFailReason.TOKEN_LACK && psyncEverSucceed.get() != INIT_STATE && !isPsyncEverSucceed()) { logger.warn("[canSendPsync]never succeed, let it psync, {}", psyncEverSucceed.get()); return true; } if(resourceManager.getLeakyBucket().tryAcquire()) { holdToken.set(true); recordPsyncSendTime(); } else { logger.warn("[canSendPsync]psync wont send as no token is available [port:{}]", redisKeeperServer.getListeningPort()); keeperStats.setLastPsyncFailReason(PsyncFailReason.TOKEN_LACK); keeperStats.increasePsyncSendFail(); return false; } } return true; } LeakyBucketBasedMasterReplicationListener(RedisMasterReplication redisMasterReplication,
                                                     RedisKeeperServer redisKeeperServer,
                                                     KeeperResourceManager resourceManager,
                                                     ScheduledExecutorService scheduled); }
LeakyBucketBasedMasterReplicationListener implements RedisMasterReplication.RedisMasterReplicationObserver { @Override public boolean canSendPsync() { if (redisMasterReplication.redisMaster().isKeeper() && (redisKeeperServer.getRedisKeeperServerState() instanceof RedisKeeperServerStateActive)) { if(replDownShortly()) { logger.info("[canSendPsync][repl down short] let psync pass"); return true; } KeeperStats keeperStats = redisKeeperServer.getKeeperMonitor().getKeeperStats(); if(keeperStats.getLastPsyncFailReason() != null && keeperStats.getLastPsyncFailReason() != PsyncFailReason.TOKEN_LACK && psyncEverSucceed.get() != INIT_STATE && !isPsyncEverSucceed()) { logger.warn("[canSendPsync]never succeed, let it psync, {}", psyncEverSucceed.get()); return true; } if(resourceManager.getLeakyBucket().tryAcquire()) { holdToken.set(true); recordPsyncSendTime(); } else { logger.warn("[canSendPsync]psync wont send as no token is available [port:{}]", redisKeeperServer.getListeningPort()); keeperStats.setLastPsyncFailReason(PsyncFailReason.TOKEN_LACK); keeperStats.increasePsyncSendFail(); return false; } } return true; } LeakyBucketBasedMasterReplicationListener(RedisMasterReplication redisMasterReplication,
                                                     RedisKeeperServer redisKeeperServer,
                                                     KeeperResourceManager resourceManager,
                                                     ScheduledExecutorService scheduled); @Override void onMasterConnected(); @Override boolean canSendPsync(); @Override void onMasterDisconnected(); @Override void endWriteRdb(); @Override void onContinue(String requestReplId, String responseReplId); @Override void onFullSync(); @Override void reFullSync(); @Override void onDumpFinished(); @Override void onDumpFail(); @Override void beginWriteRdb(EofType eofType, long masterRdbOffset); }
LeakyBucketBasedMasterReplicationListener implements RedisMasterReplication.RedisMasterReplicationObserver { @Override public boolean canSendPsync() { if (redisMasterReplication.redisMaster().isKeeper() && (redisKeeperServer.getRedisKeeperServerState() instanceof RedisKeeperServerStateActive)) { if(replDownShortly()) { logger.info("[canSendPsync][repl down short] let psync pass"); return true; } KeeperStats keeperStats = redisKeeperServer.getKeeperMonitor().getKeeperStats(); if(keeperStats.getLastPsyncFailReason() != null && keeperStats.getLastPsyncFailReason() != PsyncFailReason.TOKEN_LACK && psyncEverSucceed.get() != INIT_STATE && !isPsyncEverSucceed()) { logger.warn("[canSendPsync]never succeed, let it psync, {}", psyncEverSucceed.get()); return true; } if(resourceManager.getLeakyBucket().tryAcquire()) { holdToken.set(true); recordPsyncSendTime(); } else { logger.warn("[canSendPsync]psync wont send as no token is available [port:{}]", redisKeeperServer.getListeningPort()); keeperStats.setLastPsyncFailReason(PsyncFailReason.TOKEN_LACK); keeperStats.increasePsyncSendFail(); return false; } } return true; } LeakyBucketBasedMasterReplicationListener(RedisMasterReplication redisMasterReplication,
                                                     RedisKeeperServer redisKeeperServer,
                                                     KeeperResourceManager resourceManager,
                                                     ScheduledExecutorService scheduled); @Override void onMasterConnected(); @Override boolean canSendPsync(); @Override void onMasterDisconnected(); @Override void endWriteRdb(); @Override void onContinue(String requestReplId, String responseReplId); @Override void onFullSync(); @Override void reFullSync(); @Override void onDumpFinished(); @Override void onDumpFail(); @Override void beginWriteRdb(EofType eofType, long masterRdbOffset);  }