LeakyBucketBasedMasterReplicationListener implements RedisMasterReplication.RedisMasterReplicationObserver { @VisibleForTesting protected boolean isTokenReadyToRelease(final long deadline) { if (System.currentTimeMillis() >= deadline) { return true; } if(trafficSafeCounter == null) { trafficSafeCounter = new AtomicInteger(); } KeeperStats keeperStats = redisKeeperServer.getKeeperMonitor().getKeeperStats(); long peakBPS = keeperStats.getPeakInputInstantaneousBPS(); long curBPS = keeperStats.getInputInstantaneousBPS(); long lowWaterMark = redisKeeperServer.getKeeperConfig().getReplicationTrafficLowWaterMark(); if(curBPS < peakBPS / 2 || curBPS < lowWaterMark) { logger.debug("[isTokenReadyToRelease] current BPS {} peak BPS {} lowWaterMark {}", curBPS, peakBPS, lowWaterMark); trafficSafeCounter.incrementAndGet(); } else { logger.debug("[isTokenReadyToRelease] current BPS {} peak BPS {} lowWaterMark {}", curBPS, peakBPS, lowWaterMark); trafficSafeCounter.set(0); } return trafficSafeCounter.get() > 2; } 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 testIsTokenReadyToReleaseWhenDeadline() { long deadline = System.currentTimeMillis() - 10; Assert.assertTrue(listener.isTokenReadyToRelease(deadline)); deadline = System.currentTimeMillis() + TimeUnit.SECONDS.toMillis(30); Assert.assertFalse(listener.isTokenReadyToRelease(deadline)); }
@Test public void testIsTokenReadyToReleaseWhenPeakBPSHuge() { long deadline = System.currentTimeMillis() + TimeUnit.SECONDS.toMillis(30); when(keeperConfig.getReplicationTrafficLowWaterMark()).thenReturn(1024 * 1024 * 20L); when(keeperStats.getInputInstantaneousBPS()).thenReturn(1024 * 1024 * 21L); when(keeperStats.getPeakInputInstantaneousBPS()).thenReturn(1024 * 1024 * 50L); Assert.assertFalse(listener.isTokenReadyToRelease(deadline)); Assert.assertFalse(listener.isTokenReadyToRelease(deadline)); Assert.assertTrue(listener.isTokenReadyToRelease(deadline)); }
@Test public void testIsTokenReadyToReleaseWhenPeakBPSTiny() { long deadline = System.currentTimeMillis() + TimeUnit.SECONDS.toMillis(30); when(keeperConfig.getReplicationTrafficLowWaterMark()).thenReturn(1024 * 1024 * 20L); when(keeperStats.getInputInstantaneousBPS()).thenReturn(1024 * 1024 * 21L); when(keeperStats.getPeakInputInstantaneousBPS()).thenReturn(1024 * 1024 * 30L); Assert.assertFalse(listener.isTokenReadyToRelease(deadline)); Assert.assertFalse(listener.isTokenReadyToRelease(deadline)); Assert.assertFalse(listener.isTokenReadyToRelease(deadline)); }
@Test public void testIsTokenReadyToReleaseWhenPeakBPSChange() { long deadline = System.currentTimeMillis() + TimeUnit.SECONDS.toMillis(30); when(keeperConfig.getReplicationTrafficLowWaterMark()).thenReturn(1024 * 1024 * 20L); when(keeperStats.getInputInstantaneousBPS()).thenReturn(1024 * 1024 * 21L); when(keeperStats.getPeakInputInstantaneousBPS()).thenReturn(1024 * 1024 * 50L); Assert.assertFalse(listener.isTokenReadyToRelease(deadline)); Assert.assertFalse(listener.isTokenReadyToRelease(deadline)); when(keeperStats.getPeakInputInstantaneousBPS()).thenReturn(1024 * 1024 * 30L); Assert.assertFalse(listener.isTokenReadyToRelease(deadline)); when(keeperStats.getPeakInputInstantaneousBPS()).thenReturn(1024 * 1024 * 50L); Assert.assertFalse(listener.isTokenReadyToRelease(deadline)); Assert.assertFalse(listener.isTokenReadyToRelease(deadline)); Assert.assertTrue(listener.isTokenReadyToRelease(deadline)); }
@Test public void testIsTokenReadyToReleaseWhenLessThan20MB() { long deadline = System.currentTimeMillis() + TimeUnit.SECONDS.toMillis(30); when(keeperStats.getInputInstantaneousBPS()).thenReturn(1024 * 1024 * 2L); Assert.assertFalse(listener.isTokenReadyToRelease(deadline)); Assert.assertFalse(listener.isTokenReadyToRelease(deadline)); Assert.assertTrue(listener.isTokenReadyToRelease(deadline)); }
@Test public void testIsTokenReadyToReleaseIfNotContinuously() { long deadline = System.currentTimeMillis() + TimeUnit.SECONDS.toMillis(30); when(keeperStats.getInputInstantaneousBPS()).thenReturn(1024 * 1024 * 2L); Assert.assertFalse(listener.isTokenReadyToRelease(deadline)); Assert.assertFalse(listener.isTokenReadyToRelease(deadline)); when(keeperStats.getInputInstantaneousBPS()).thenReturn(1024 * 1024 * 21L); Assert.assertFalse(listener.isTokenReadyToRelease(deadline)); Assert.assertFalse(listener.isTokenReadyToRelease(deadline)); when(keeperStats.getInputInstantaneousBPS()).thenReturn(1024 * 1024 * 2L); Assert.assertFalse(listener.isTokenReadyToRelease(deadline)); Assert.assertFalse(listener.isTokenReadyToRelease(deadline)); Assert.assertTrue(listener.isTokenReadyToRelease(deadline)); }