@Override
  public void start(final QueryId queryId) {
    log.info("Attempting to start query {} in runtime {}", queryId, getApplicationId());
    if (collocatedQueries.containsKey(queryId) && !collocatedQueries.get(queryId).everStarted) {
      if (!kafkaStreams.getTopologyByName(queryId.toString()).isPresent()) {
        final KafkaFuture<Void> toAdd = kafkaStreams
            .addNamedTopology(collocatedQueries.get(queryId).getTopology()).all();
        topolgogiesToAdd.add(toAdd);
      } else {
        throw new IllegalArgumentException("Cannot start because Streams is not done terminating"
                                               + " an older version of query : " + queryId);
      }
    } else {
      throw new IllegalArgumentException("Cannot start because query " + queryId + " was not "
                                             + "registered to runtime " + getApplicationId());
    }
    log.info("Query {} was started successfully", queryId);
  }
@Test
    public void shouldStartQuery() {
        //When:
        sharedKafkaStreamsRuntimeImpl.start(queryId);

        //Then:
        assertThat("Query was not added", sharedKafkaStreamsRuntimeImpl.getQueries().contains(queryId));
    }