class QueryMonitor implements Closeable {

  void restartFailedQueries() {
    // Collect a list of new queries in ERROR state
    ksqlEngine.getPersistentQueries().stream()
        .filter(QueryMetadata::isError)
        .filter(query -> !queriesRetries.containsKey(query.getQueryId()))
        .map(QueryMetadata::getQueryId)
        .forEach(queryId -> queriesRetries.put(queryId, newRetryEvent(queryId)));

    maybeRestartQueries();
  }

  public  QueryMonitor(final KsqlConfig ksqlConfig, final KsqlEngine ksqlEngine);
  @VisibleForTesting  QueryMonitor(
      final KsqlEngine ksqlEngine,
      final ExecutorService executor,
      final long retryBackoffInitialMs,
      final long retryBackoffMaxMs,
      final long statusRunningThresholdMs,
      final Ticker ticker
  );

}