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 shouldScatterGatherAndMergeShowQueries() {
    // Given
    when(sessionProperties.getInternalRequest()).thenReturn(false);
    final ConfiguredStatement<?> showQueries = engine.configure("SHOW QUERIES;");
    final PersistentQueryMetadata localMetadata = givenPersistentQuery("id", RUNNING_QUERY_STATE);
    final PersistentQueryMetadata remoteMetadata = givenPersistentQuery("id", ERROR_QUERY_STATE);

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

    final List<RunningQuery> remoteRunningQueries = Collections.singletonList(persistentQueryMetadataToRunningQuery(
        remoteMetadata,
            new QueryStatusCount(Collections.singletonMap(KsqlQueryStatus.ERROR, 1))));


    when(remoteQueries.getQueries()).thenReturn(remoteRunningQueries);
    when(ksqlEntityList.get(anyInt())).thenReturn(remoteQueries);
    when(response.getResponse()).thenReturn(ksqlEntityList);

    queryStatusCount.updateStatusCount(RUNNING_QUERY_STATE, 1);
    queryStatusCount.updateStatusCount(ERROR_QUERY_STATE, 1);

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

    // Then
    assertThat(queries.getQueries(), containsInAnyOrder(persistentQueryMetadataToRunningQuery(localMetadata, queryStatusCount)));
  }