### Question:
HubRegistry implements PartitionListener { protected void timeout() { logger.info("Checking for expired hubs"); long expirationTime = clock.millis() - offlineTimeoutMs; for(HubState state: hubs.values()) { if(state.lastHeartbeat < expirationTime) { if(hubs.remove(state.getHubId(), state)) { try { onTimeout(state.getHubId()); } catch(Exception e) { logger.warn("Error sending timeout for [{}]", state.getHubId(), e); } } } } } @Inject  HubRegistry(
         Clock clock,
         HubDAO hubDao,
         Partitioner partitioner,
         PlatformMessageBus platformBus,
         HubRegistryConfig config,
         PlacePopulationCacheManager populationCacheMgr
   ); @WarmUp void start(); @PreDestroy void stop(); int getOnlineHubs(); boolean isOnline(String hubId); void online(String hubId, int partitionId, String hubBridge); void offline(String hubId, String hubBridge); boolean remove(String hubId); @Override void onPartitionsChanged(PartitionChangedEvent event);  }### Answer:
@Test public void testTimeout() throws Exception { EasyMock.expect(mockClock.millis()).andReturn(1000L).once(); EasyMock.expect(mockClock.millis()).andReturn(2000L).once(); EasyMock.expect(mockClock.millis()).andReturn(1000000L).once(); expectStreamPartitionAndReturn(0, ImmutableList.of(hub1, hub2)); expectGetHubByIdAndReturn(hub1); expectDisconnected(hub1.getId()); expectGetHubByIdAndReturn(hub2); expectDisconnected(hub2.getId()); replay(); PartitionChangedEvent event = new PartitionChangedEvent(); event.setAddedPartitions(ImmutableSet.of(0)); event.setRemovedPartitions(ImmutableSet.of()); event.setPartitions(ImmutableSet.of(new DefaultPartition(0))); registry.onPartitionsChanged(event); assertEquals(2, registry.getOnlineHubs()); assertNoMessages(); registry.timeout(); assertEquals(2, registry.getOnlineHubs()); assertNoMessages(); registry.timeout(); assertEquals(0, registry.getOnlineHubs()); assertDisconnected(null); assertValueChangeDown(null); assertDisconnected(null); assertValueChangeDown(null); assertNoMessages(); verify(); }