@Override
  public void run() {
    final Instant now = time.get();

    try {
      final Collection<PersistentQueryMetadata> queries = engine.getPersistentQueries();
      final Optional<Double> saturation = queries.stream()
          .collect(Collectors.groupingBy(PersistentQueryMetadata::getQueryApplicationId))
          .entrySet()
          .stream()
          .map(e -> measure(now, e.getKey(), e.getValue()))
          .max(PersistentQuerySaturationMetrics::compareSaturation)
          .orElse(Optional.of(0.0));
      saturation.ifPresent(s -> report(now, s));

      final Set<String> appIds = queries.stream()
          .map(PersistentQueryMetadata::getQueryApplicationId)
          .collect(Collectors.toSet());
      for (final String appId
          : Sets.difference(new HashSet<>(perKafkaStreamsStats.keySet()), appIds)) {
        perKafkaStreamsStats.get(appId).cleanup(reporter);
        perKafkaStreamsStats.remove(appId);
      }
    } catch (final RuntimeException e) {
      LOGGER.error("Error collecting saturation", e);
      throw e;
    }
  }
@Test
  public void shouldCleanupQueryMetricWhenRuntimeRemoved() {
    // Given:
    final Instant start = Instant.now();
    when(clock.get()).thenReturn(start);
    givenMetrics(kafkaStreams1)
        .withThreadStartTime("t1", start)
        .withBlockedTime("t1", Duration.ofMinutes(0));
    collector.run();
    when(engine.getPersistentQueries()).thenReturn(ImmutableList.of(query2));

    // When:
    collector.run();

    // Then:
    verify(reporter).cleanup("node-query-saturation", ImmutableMap.of("query-id", "hootie"));
  }