public static QueryDescription forQueryMetadata(
      final QueryMetadata queryMetadata,
      final Map<KsqlHostInfoEntity, KsqlQueryStatus> ksqlHostQueryStatus
  ) {
    if (queryMetadata instanceof PersistentQueryMetadata) {
      final PersistentQueryMetadata persistentQuery = (PersistentQueryMetadata) queryMetadata;
      return create(
          persistentQuery,
          persistentQuery.getResultTopic().map(t -> t.getKeyFormat().getWindowType())
              .orElse(Optional.empty()),
          persistentQuery.getSinkName(),
          ksqlHostQueryStatus
      );
    }

    return create(
        queryMetadata,
        Optional.empty(),
        Optional.empty(),
        ksqlHostQueryStatus
    );
  }
@Test
  public void shouldReportPersistentQueriesStatus() {
    assertThat(persistentQueryDescription.getState(), is(Optional.of("RUNNING")));

    final Map<KsqlHostInfoEntity, KsqlQueryStatus> updatedStatusMap = new HashMap<>(STATUS_MAP);
    updatedStatusMap.put(new KsqlHostInfoEntity("anotherhost", 8080), KsqlQueryStatus.ERROR);
    final QueryDescription updatedPersistentQueryDescription =
        QueryDescriptionFactory.forQueryMetadata(persistentQuery, updatedStatusMap);
    assertThat(updatedPersistentQueryDescription.getState(), is(Optional.of("ERROR")));
  }