@Test public void getOldestWithSingleStampReturnsSame() { HashMap<Integer, UpdateStamp> stamps = new HashMap<Integer, UpdateStamp>(); long time = 1234L; UpdateStamp stamp = new UpdateStamp(time, UpdateStamp.State.Active, false); stamps.put(0, stamp); when(context.getSlotStamps(Matchers.<Granularity>any(), anyInt())) .thenReturn(stamps) .thenReturn(null); SlotState slotState = new SlotState(Granularity.MIN_5, 0, stamp.getState()) .withTimestamp(time); String expected = slotState.toString(); Collection<String> result = service.getOldestUnrolledSlotPerGranularity(0); assertNotNull(result); assertEquals(1, result.size()); assertTrue(result.contains(expected)); }
public synchronized Collection<String> getOldestUnrolledSlotPerGranularity(int shard) { final Set<String> results = new HashSet<String>(); for (Granularity g : Granularity.rollupGranularities()) { final Map<Integer, UpdateStamp> stateTimestamps = context.getSlotStamps(g, shard); if (stateTimestamps == null || stateTimestamps.isEmpty()) { continue; } SlotState minSlot = new SlotState().withTimestamp(System.currentTimeMillis()); boolean add = false; for (Map.Entry<Integer, UpdateStamp> entry : stateTimestamps.entrySet()) { final UpdateStamp stamp = entry.getValue(); if (stamp.getState() != UpdateStamp.State.Rolled && stamp.getTimestamp() < minSlot.getTimestamp()) { minSlot = new SlotState(g, entry.getKey(), stamp.getState()).withTimestamp(stamp.getTimestamp()); add = true; } } if (add) { results.add(minSlot.toString()); } } return results; }
RollupService implements Runnable, RollupServiceMBean { public synchronized Collection<String> getOldestUnrolledSlotPerGranularity(int shard) { final Set<String> results = new HashSet<String>(); for (Granularity g : Granularity.rollupGranularities()) { final Map<Integer, UpdateStamp> stateTimestamps = context.getSlotStamps(g, shard); if (stateTimestamps == null || stateTimestamps.isEmpty()) { continue; } SlotState minSlot = new SlotState().withTimestamp(System.currentTimeMillis()); boolean add = false; for (Map.Entry<Integer, UpdateStamp> entry : stateTimestamps.entrySet()) { final UpdateStamp stamp = entry.getValue(); if (stamp.getState() != UpdateStamp.State.Rolled && stamp.getTimestamp() < minSlot.getTimestamp()) { minSlot = new SlotState(g, entry.getKey(), stamp.getState()).withTimestamp(stamp.getTimestamp()); add = true; } } if (add) { results.add(minSlot.toString()); } } return results; } }
RollupService implements Runnable, RollupServiceMBean { public synchronized Collection<String> getOldestUnrolledSlotPerGranularity(int shard) { final Set<String> results = new HashSet<String>(); for (Granularity g : Granularity.rollupGranularities()) { final Map<Integer, UpdateStamp> stateTimestamps = context.getSlotStamps(g, shard); if (stateTimestamps == null || stateTimestamps.isEmpty()) { continue; } SlotState minSlot = new SlotState().withTimestamp(System.currentTimeMillis()); boolean add = false; for (Map.Entry<Integer, UpdateStamp> entry : stateTimestamps.entrySet()) { final UpdateStamp stamp = entry.getValue(); if (stamp.getState() != UpdateStamp.State.Rolled && stamp.getTimestamp() < minSlot.getTimestamp()) { minSlot = new SlotState(g, entry.getKey(), stamp.getState()).withTimestamp(stamp.getTimestamp()); add = true; } } if (add) { results.add(minSlot.toString()); } } return results; } RollupService(ScheduleContext context); @VisibleForTesting  RollupService(ScheduleContext context,
                         ShardStateManager shardStateManager,
                         ThreadPoolExecutor locatorFetchExecutors,
                         ThreadPoolExecutor rollupReadExecutors,
                         ThreadPoolExecutor rollupWriteExecutors,
                         long rollupDelayMillis,
                         long rollupDelayForMetricsWithShortDelay,
                         long rollupWaitForMetricsWithLongDelay,
                         long pollerPeriod,
                         long configRefreshInterval); }
RollupService implements Runnable, RollupServiceMBean { public synchronized Collection<String> getOldestUnrolledSlotPerGranularity(int shard) { final Set<String> results = new HashSet<String>(); for (Granularity g : Granularity.rollupGranularities()) { final Map<Integer, UpdateStamp> stateTimestamps = context.getSlotStamps(g, shard); if (stateTimestamps == null || stateTimestamps.isEmpty()) { continue; } SlotState minSlot = new SlotState().withTimestamp(System.currentTimeMillis()); boolean add = false; for (Map.Entry<Integer, UpdateStamp> entry : stateTimestamps.entrySet()) { final UpdateStamp stamp = entry.getValue(); if (stamp.getState() != UpdateStamp.State.Rolled && stamp.getTimestamp() < minSlot.getTimestamp()) { minSlot = new SlotState(g, entry.getKey(), stamp.getState()).withTimestamp(stamp.getTimestamp()); add = true; } } if (add) { results.add(minSlot.toString()); } } return results; } RollupService(ScheduleContext context); @VisibleForTesting  RollupService(ScheduleContext context,
                         ShardStateManager shardStateManager,
                         ThreadPoolExecutor locatorFetchExecutors,
                         ThreadPoolExecutor rollupReadExecutors,
                         ThreadPoolExecutor rollupWriteExecutors,
                         long rollupDelayMillis,
                         long rollupDelayForMetricsWithShortDelay,
                         long rollupWaitForMetricsWithLongDelay,
                         long pollerPeriod,
                         long configRefreshInterval); void initializeGauges(); void forcePoll(); void run(); synchronized void setServerTime(long millis); synchronized long getServerTime(); synchronized void setKeepingServerTime(boolean b); synchronized boolean getKeepingServerTime(); synchronized void setPollerPeriod(long l); synchronized long getPollerPeriod(); synchronized int getScheduledSlotCheckCount(); synchronized int getSecondsSinceLastSlotCheck(); synchronized int getSlotCheckConcurrency(); synchronized void setSlotCheckConcurrency(int i); synchronized int getRollupConcurrency(); synchronized void setRollupConcurrency(int i); synchronized int getQueuedRollupCount(); synchronized int getInFlightRollupCount(); synchronized boolean getActive(); synchronized void setActive(boolean b); void addShard(Integer shard); void removeShard(Integer shard); Collection<Integer> getManagedShards(); synchronized Collection<Integer> getRecentlyScheduledShards(); synchronized Collection<String> getOldestUnrolledSlotPerGranularity(int shard); }
RollupService implements Runnable, RollupServiceMBean { public synchronized Collection<String> getOldestUnrolledSlotPerGranularity(int shard) { final Set<String> results = new HashSet<String>(); for (Granularity g : Granularity.rollupGranularities()) { final Map<Integer, UpdateStamp> stateTimestamps = context.getSlotStamps(g, shard); if (stateTimestamps == null || stateTimestamps.isEmpty()) { continue; } SlotState minSlot = new SlotState().withTimestamp(System.currentTimeMillis()); boolean add = false; for (Map.Entry<Integer, UpdateStamp> entry : stateTimestamps.entrySet()) { final UpdateStamp stamp = entry.getValue(); if (stamp.getState() != UpdateStamp.State.Rolled && stamp.getTimestamp() < minSlot.getTimestamp()) { minSlot = new SlotState(g, entry.getKey(), stamp.getState()).withTimestamp(stamp.getTimestamp()); add = true; } } if (add) { results.add(minSlot.toString()); } } return results; } RollupService(ScheduleContext context); @VisibleForTesting  RollupService(ScheduleContext context,
                         ShardStateManager shardStateManager,
                         ThreadPoolExecutor locatorFetchExecutors,
                         ThreadPoolExecutor rollupReadExecutors,
                         ThreadPoolExecutor rollupWriteExecutors,
                         long rollupDelayMillis,
                         long rollupDelayForMetricsWithShortDelay,
                         long rollupWaitForMetricsWithLongDelay,
                         long pollerPeriod,
                         long configRefreshInterval); void initializeGauges(); void forcePoll(); void run(); synchronized void setServerTime(long millis); synchronized long getServerTime(); synchronized void setKeepingServerTime(boolean b); synchronized boolean getKeepingServerTime(); synchronized void setPollerPeriod(long l); synchronized long getPollerPeriod(); synchronized int getScheduledSlotCheckCount(); synchronized int getSecondsSinceLastSlotCheck(); synchronized int getSlotCheckConcurrency(); synchronized void setSlotCheckConcurrency(int i); synchronized int getRollupConcurrency(); synchronized void setRollupConcurrency(int i); synchronized int getQueuedRollupCount(); synchronized int getInFlightRollupCount(); synchronized boolean getActive(); synchronized void setActive(boolean b); void addShard(Integer shard); void removeShard(Integer shard); Collection<Integer> getManagedShards(); synchronized Collection<Integer> getRecentlyScheduledShards(); synchronized Collection<String> getOldestUnrolledSlotPerGranularity(int shard);  }