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);  }
@Test public void testCanSendPsyncWithNonActiveState() { when(redisKeeperServer.getRedisKeeperServerState()).thenReturn(new RedisKeeperServerStateBackup(redisKeeperServer)); when(redisMaster.isKeeper()).thenReturn(true); verify(resourceManager, never()).getLeakyBucket(); Assert.assertTrue(listener.canSendPsync()); }
@Test public void testCanSendPsyncWithMasterNotKeeper() { when(redisKeeperServer.getRedisKeeperServerState()).thenReturn(new RedisKeeperServerStateActive(redisKeeperServer)); when(redisMaster.isKeeper()).thenReturn(false); verify(resourceManager, never()).getLeakyBucket(); Assert.assertTrue(listener.canSendPsync()); }
@Test public void testCanSendPsync() { when(redisKeeperServer.getRedisKeeperServerState()).thenReturn(new RedisKeeperServerStateActive(redisKeeperServer)); when(redisMaster.isKeeper()).thenReturn(true); when(resourceManager.getLeakyBucket()).thenReturn(new DefaultLeakyBucket(1)); Assert.assertTrue(listener.canSendPsync()); verify(resourceManager, times(1)).getLeakyBucket(); }
@Test public void testCanSendPsyncWithFirstTimeFail() { when(redisKeeperServer.getRedisKeeperServerState()).thenReturn(new RedisKeeperServerStateActive(redisKeeperServer)); when(redisMaster.isKeeper()).thenReturn(true); LeakyBucket leakyBucket = new DefaultLeakyBucket(1); when(resourceManager.getLeakyBucket()).thenReturn(leakyBucket); leakyBucket.tryAcquire(); Assert.assertFalse(listener.canSendPsync()); Assert.assertFalse(listener.canSendPsync()); verify(resourceManager, times(2)).getLeakyBucket(); }
@Test public void testDynamicAdjustTokenSize() { when(keeperConfig.getLeakyBucketInitSize()).thenReturn(3); CompositeLeakyBucket leakyBucket = new CompositeLeakyBucket(keeperConfig, metaServerKeeperService, keeperContainerService); when(resourceManager.getLeakyBucket()).thenReturn(leakyBucket); listener = new LeakyBucketBasedMasterReplicationListener(redisMasterReplication, redisKeeperServer, resourceManager, scheduled); 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()); when(redisKeeperServer.getRedisKeeperServerState()).thenReturn(new RedisKeeperServerStateActive(redisKeeperServer)); when(redisMaster.isKeeper()).thenReturn(true); for(int i = 0; i < keeperConfig.getLeakyBucketInitSize() + 1; i++) { Assert.assertTrue(listener.canSendPsync()); } Assert.assertFalse(listener.canSendPsync()); }
@Test public void testDynamicAdjustTokenSize2() { when(redisKeeperServer.getRedisKeeperServerState()).thenReturn(new RedisKeeperServerStateActive(redisKeeperServer)); when(redisMaster.isKeeper()).thenReturn(true); when(keeperConfig.getLeakyBucketInitSize()).thenReturn(3); CompositeLeakyBucket leakyBucket = new CompositeLeakyBucket(keeperConfig, metaServerKeeperService, keeperContainerService); leakyBucket.setScheduled(scheduled); 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()); when(metaServerKeeperService.refreshKeeperContainerTokenStatus(any())).thenReturn(response); int originSize = keeperConfig.getLeakyBucketInitSize(); when(keeperConfig.getLeakyBucketInitSize()).thenReturn(originSize - 2); leakyBucket.checkKeeperConfigChange(); sleep(110); for(int i = 0; i < originSize; i++) { leakyBucket.release(); } Assert.assertEquals(originSize -2 , leakyBucket.getTotalSize()); for(int i = 0; i < originSize - 2; i++) { Assert.assertTrue(listener.canSendPsync()); } Assert.assertFalse(listener.canSendPsync()); }
@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()); }
@Test public void testDynamicCloseIt() { 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.isKeeperRateLimitOpen()).thenReturn(false); leakyBucket.setScheduled(scheduled); leakyBucket.checkKeeperConfigChange(); sleep(110); int INFINITY = 1024; for(int i = 0; i < INFINITY; i ++) { Assert.assertTrue(listener.canSendPsync()); } }
@Test public void testWhenDeployRateLimitWillNotBlock() { when(redisKeeperServer.getRedisKeeperServerState()).thenReturn(new RedisKeeperServerStateActive(redisKeeperServer)); when(redisMaster.isKeeper()).thenReturn(true); when(resourceManager.getLeakyBucket()).thenReturn(leakyBucket); when(leakyBucket.tryAcquire()).thenReturn(true); logger.info("[{}]", redisKeeperServer.getKeeperConfig().getReplDownSafeIntervalMilli()); replicationStoreStats.refreshReplDownSince(System.currentTimeMillis()); Assert.assertTrue(listener.canSendPsync()); verify(leakyBucket, never()).tryAcquire(); }
@Test public void testWhenRestart() { when(redisKeeperServer.getRedisKeeperServerState()).thenReturn(new RedisKeeperServerStateActive(redisKeeperServer)); when(redisMaster.isKeeper()).thenReturn(true); when(resourceManager.getLeakyBucket()).thenReturn(leakyBucket); when(leakyBucket.tryAcquire()).thenReturn(true); replicationStoreStats.refreshReplDownSince(0); Assert.assertTrue(listener.canSendPsync()); verify(leakyBucket, times(1)).tryAcquire(); }
@Test public void testWhenNetworkBlockTooLong() { when(redisKeeperServer.getRedisKeeperServerState()).thenReturn(new RedisKeeperServerStateActive(redisKeeperServer)); when(redisMaster.isKeeper()).thenReturn(true); when(resourceManager.getLeakyBucket()).thenReturn(leakyBucket); when(leakyBucket.tryAcquire()).thenReturn(false); logger.info("[{}]", redisKeeperServer.getKeeperConfig().getReplDownSafeIntervalMilli()); replicationStoreStats.refreshReplDownSince(System.currentTimeMillis() - TimeUnit.MINUTES.toMillis(10)); Assert.assertFalse(listener.canSendPsync()); verify(leakyBucket, times(1)).tryAcquire(); }