public static StatementExecutorResponse execute(
      final ConfiguredStatement<ListQueries> statement,
      final SessionProperties sessionProperties,
      final KsqlExecutionContext executionContext,
      final ServiceContext serviceContext
  ) {
    final RemoteHostExecutor remoteHostExecutor = RemoteHostExecutor.create(
        statement,
        sessionProperties,
        executionContext,
        serviceContext.getKsqlClient()
    );
    return statement.getStatement().getShowExtended()
        ? executeExtended(statement, sessionProperties, executionContext, remoteHostExecutor)
        : executeSimple(statement, executionContext, remoteHostExecutor);
  }
@Test
  public void shouldIncludeUnresponsiveIfShowQueriesExtendedFutureThrowsException() {
    // Given
    when(sessionProperties.getInternalRequest()).thenReturn(false);
    final ConfiguredStatement<?> showQueries = engine.configure("SHOW QUERIES EXTENDED;");
    final PersistentQueryMetadata metadata = givenPersistentQuery("id", RUNNING_QUERY_STATE);

    
    when(mockKsqlEngine.getAllLiveQueries()).thenReturn(ImmutableList.of(metadata));
    when(mockKsqlEngine.getPersistentQueries()).thenReturn(ImmutableList.of(metadata));

    when(ksqlClient.makeKsqlRequest(any(), any(), any())).thenThrow(new KsqlRestClientException("error"));
    when(serviceContext.getKsqlClient()).thenReturn(ksqlClient);

    // When
    final Map<KsqlHostInfoEntity, KsqlQueryStatus> map = new HashMap<>();
    map.put(LOCAL_KSQL_HOST_INFO_ENTITY, KsqlQueryStatus.RUNNING);
    map.put(REMOTE_KSQL_HOST_INFO_ENTITY, KsqlQueryStatus.UNRESPONSIVE);

    // When
    final QueryDescriptionList queries = (QueryDescriptionList) CustomExecutors.LIST_QUERIES.execute(
        showQueries,
        sessionProperties,
        mockKsqlEngine,
        serviceContext
    ).getEntity().orElseThrow(IllegalStateException::new);

    // Then
    assertThat(queries.getQueryDescriptions(),
        containsInAnyOrder(QueryDescriptionFactory.forQueryMetadata(metadata, map)));
  }