@Test public void testDeleteStaleChunks() { assertTrue(chunkManager.getChunkMap().isEmpty()); assertEquals(0, offHeapChunkManagerTask.deleteStaleChunks(startTimeSecs)); assertEquals(0, offHeapChunkManagerTask.deleteStaleChunks(startTimeSecs + 1)); assertEquals(0, offHeapChunkManagerTask.deleteStaleChunks(startTimeSecs + 3600 * 2)); assertEquals(0, offHeapChunkManagerTask.deleteStaleChunks(startTimeSecs + 3600 * 3)); chunkManager.addMetric(MetricUtils.makeMetricString( testMetricName, inputTagString1, startTimeSecs + 1, testValue)); final List<TimeSeries> timeSeries = queryChunkManager(startTimeSecs, startTimeSecs + 3600 * 3); assertEquals(1, timeSeries.size()); assertEquals(1, chunkManager.getChunkMap().size()); assertEquals(0, offHeapChunkManagerTask.deleteStaleChunks(startTimeSecs)); assertEquals(0, offHeapChunkManagerTask.deleteStaleChunks(startTimeSecs + 1)); assertEquals(0, offHeapChunkManagerTask.deleteStaleChunks(startTimeSecs + 3600 - 1)); assertEquals(0, offHeapChunkManagerTask.deleteStaleChunks(startTimeSecs + 3600)); assertEquals(0, offHeapChunkManagerTask.deleteStaleChunks(startTimeSecs + 3600 + 1)); assertEquals(0, offHeapChunkManagerTask.deleteStaleChunks(startTimeSecs + 3600 * 2 - 1)); assertEquals(1, offHeapChunkManagerTask.deleteStaleChunks(startTimeSecs + 3600 * 2)); assertTrue(chunkManager.getChunkMap().isEmpty()); assertEquals(0, offHeapChunkManagerTask.deleteStaleChunks(startTimeSecs)); assertEquals(0, offHeapChunkManagerTask.deleteStaleChunks(startTimeSecs + 1)); assertEquals(0, offHeapChunkManagerTask.deleteStaleChunks(startTimeSecs + 3600 * 2)); assertEquals(0, offHeapChunkManagerTask.deleteStaleChunks(startTimeSecs + 3600 * 3)); final List<TimeSeries> timeSeries2 = queryChunkManager(startTimeSecs, startTimeSecs + 3600 * 3); assertTrue(timeSeries2.isEmpty()); chunkManager.addMetric(MetricUtils.makeMetricString( testMetricName, inputTagString1, startTimeSecs + 1, testValue)); chunkManager.addMetric(MetricUtils.makeMetricString( testMetricName, inputTagString1, startTimePlusTwoHoursSecs + 1, testValue)); assertEquals(2, chunkManager.getChunkMap().size()); final List<TimeSeries> timeSeries3 = queryChunkManager(startTimeSecs, startTimeSecs + 3600 * 3); assertEquals(1, timeSeries3.size()); assertEquals(2, chunkManager.getChunkMap().size()); assertEquals(0, offHeapChunkManagerTask.deleteStaleChunks(startTimeSecs)); assertEquals(0, offHeapChunkManagerTask.deleteStaleChunks(startTimeSecs + 1)); assertEquals(0, offHeapChunkManagerTask.deleteStaleChunks(startTimeSecs + 3600 - 1)); assertEquals(0, offHeapChunkManagerTask.deleteStaleChunks(startTimeSecs + 3600)); assertEquals(0, offHeapChunkManagerTask.deleteStaleChunks(startTimeSecs + 3600 + 1)); assertEquals(0, offHeapChunkManagerTask.deleteStaleChunks(startTimeSecs + 3600 * 2 - 1)); assertEquals(1, offHeapChunkManagerTask.deleteStaleChunks(startTimeSecs + 3600 * 2)); assertEquals(1, chunkManager.getChunkMap().size()); chunkManager.getChunkMap() .forEach((k, v) -> assertEquals(startTimePlusTwoHoursSecs, k.longValue())); assertEquals(0, offHeapChunkManagerTask.deleteStaleChunks(startTimeSecs + 3600 * 2 + 1)); assertEquals(0, offHeapChunkManagerTask.deleteStaleChunks(startTimeSecs + 3600 * 3 - 1)); assertEquals(0, offHeapChunkManagerTask.deleteStaleChunks(startTimeSecs + 3600 * 3)); assertEquals(0, offHeapChunkManagerTask.deleteStaleChunks(startTimeSecs + 3600 * 3 + 1)); assertEquals(0, offHeapChunkManagerTask.deleteStaleChunks(startTimeSecs + 3600 * 4 - 1)); assertEquals(1, offHeapChunkManagerTask.deleteStaleChunks(startTimeSecs + 3600 * 4 + 1)); assertTrue(chunkManager.getChunkMap().isEmpty()); testDeleteMultipleChunks(0); testDeleteMultipleChunks(1); testDeleteMultipleChunks(2); testDeleteMultipleChunks(3); testDeleteMultipleChunks(4); }
int deleteStaleChunks(long staleDataCutoffSecs) { if (staleDataCutoffSecs <= 0) { throw new IllegalArgumentException("staleDateCutOffSecs can't be negative."); } LOG.info("stale data cut off secs is {}.", staleDataCutoffSecs); List<Map.Entry<Long, Chunk>> staleChunks = new ArrayList<>(); for (Map.Entry<Long, Chunk> chunkEntry: chunkManager.getChunkMap().entrySet()) { Chunk chunk = chunkEntry.getValue(); if (chunk.info().endTimeSecs <= staleDataCutoffSecs) { staleChunks.add(chunkEntry); } } LOG.info("Number of stale chunks is: {}.", staleChunks.size()); chunkManager.removeStaleChunks(staleChunks); return staleChunks.size(); }
OffHeapChunkManagerTask implements Runnable { int deleteStaleChunks(long staleDataCutoffSecs) { if (staleDataCutoffSecs <= 0) { throw new IllegalArgumentException("staleDateCutOffSecs can't be negative."); } LOG.info("stale data cut off secs is {}.", staleDataCutoffSecs); List<Map.Entry<Long, Chunk>> staleChunks = new ArrayList<>(); for (Map.Entry<Long, Chunk> chunkEntry: chunkManager.getChunkMap().entrySet()) { Chunk chunk = chunkEntry.getValue(); if (chunk.info().endTimeSecs <= staleDataCutoffSecs) { staleChunks.add(chunkEntry); } } LOG.info("Number of stale chunks is: {}.", staleChunks.size()); chunkManager.removeStaleChunks(staleChunks); return staleChunks.size(); } }
OffHeapChunkManagerTask implements Runnable { int deleteStaleChunks(long staleDataCutoffSecs) { if (staleDataCutoffSecs <= 0) { throw new IllegalArgumentException("staleDateCutOffSecs can't be negative."); } LOG.info("stale data cut off secs is {}.", staleDataCutoffSecs); List<Map.Entry<Long, Chunk>> staleChunks = new ArrayList<>(); for (Map.Entry<Long, Chunk> chunkEntry: chunkManager.getChunkMap().entrySet()) { Chunk chunk = chunkEntry.getValue(); if (chunk.info().endTimeSecs <= staleDataCutoffSecs) { staleChunks.add(chunkEntry); } } LOG.info("Number of stale chunks is: {}.", staleChunks.size()); chunkManager.removeStaleChunks(staleChunks); return staleChunks.size(); } OffHeapChunkManagerTask(ChunkManager chunkManager);  OffHeapChunkManagerTask(ChunkManager chunkManager,
                                 int metricsDelaySecs,
                                 int staleDataDelaySecs); }
OffHeapChunkManagerTask implements Runnable { int deleteStaleChunks(long staleDataCutoffSecs) { if (staleDataCutoffSecs <= 0) { throw new IllegalArgumentException("staleDateCutOffSecs can't be negative."); } LOG.info("stale data cut off secs is {}.", staleDataCutoffSecs); List<Map.Entry<Long, Chunk>> staleChunks = new ArrayList<>(); for (Map.Entry<Long, Chunk> chunkEntry: chunkManager.getChunkMap().entrySet()) { Chunk chunk = chunkEntry.getValue(); if (chunk.info().endTimeSecs <= staleDataCutoffSecs) { staleChunks.add(chunkEntry); } } LOG.info("Number of stale chunks is: {}.", staleChunks.size()); chunkManager.removeStaleChunks(staleChunks); return staleChunks.size(); } OffHeapChunkManagerTask(ChunkManager chunkManager);  OffHeapChunkManagerTask(ChunkManager chunkManager,
                                 int metricsDelaySecs,
                                 int staleDataDelaySecs); @Override void run(); }
OffHeapChunkManagerTask implements Runnable { int deleteStaleChunks(long staleDataCutoffSecs) { if (staleDataCutoffSecs <= 0) { throw new IllegalArgumentException("staleDateCutOffSecs can't be negative."); } LOG.info("stale data cut off secs is {}.", staleDataCutoffSecs); List<Map.Entry<Long, Chunk>> staleChunks = new ArrayList<>(); for (Map.Entry<Long, Chunk> chunkEntry: chunkManager.getChunkMap().entrySet()) { Chunk chunk = chunkEntry.getValue(); if (chunk.info().endTimeSecs <= staleDataCutoffSecs) { staleChunks.add(chunkEntry); } } LOG.info("Number of stale chunks is: {}.", staleChunks.size()); chunkManager.removeStaleChunks(staleChunks); return staleChunks.size(); } OffHeapChunkManagerTask(ChunkManager chunkManager);  OffHeapChunkManagerTask(ChunkManager chunkManager,
                                 int metricsDelaySecs,
                                 int staleDataDelaySecs); @Override void run(); static final int DEFAULT_METRICS_DELAY_SECS; static final int DEFAULT_STALE_DATA_DELAY_SECS; }