public String getRuntimeAndMaybeAddRuntime(final QueryId queryId,
                                             final Collection<String> sourceTopics,
                                             final KsqlConfig config) {
    if (idToRuntime.containsKey(queryId)) {
      return idToRuntime.get(queryId);
    }
    final List<String> possibleRuntimes = runtimesToSourceTopics.entrySet()
        .stream()
        .filter(t -> t.getValue().stream().noneMatch(sourceTopics::contains))
        .map(Map.Entry::getKey)
        .collect(Collectors.toList());
    final String runtime;
    if (possibleRuntimes.isEmpty()) {
      runtime = makeNewRuntime(config);
    } else {
      runtime = possibleRuntimes.get(Math.abs(queryId.hashCode() % possibleRuntimes.size()));
    }
    runtimesToSourceTopics.get(runtime).addAll(sourceTopics);
    idToRuntime.put(queryId, runtime);
    log.info("Assigning query {} to runtime {}", queryId, runtime);
    return runtime;
  }
@Test
  public void shouldGetSameRuntimeForSameQueryId() {
    final String runtime = runtimeAssignor.getRuntimeAndMaybeAddRuntime(
        query1,
        sourceTopics1,
        KSQL_CONFIG
    );
    assertThat(runtime, equalTo(firstRuntime));
  }