public static SourceDescription create(
      final DataSource dataSource,
      final boolean extended,
      final List<RunningQuery> readQueries,
      final List<RunningQuery> writeQueries,
      final Optional<TopicDescription> topicDescription,
      final List<QueryOffsetSummary> queryOffsetSummaries,
      final List<String> sourceConstraints,
      final MetricCollectors metricCollectors
  ) {
    return create(
        dataSource,
        extended,
        readQueries,
        writeQueries,
        topicDescription,
        queryOffsetSummaries,
        sourceConstraints,
        Stream.empty(),
        Stream.empty(),
        new KsqlHostInfo("", 0),
        metricCollectors
    );
  }
@Test
  public void testShouldIncludeRemoteStatsIfProvided() {
    final List<QueryHostStat> remoteStats = IntStream.range(0, 5)
        .boxed()
        .map(x -> new QueryHostStat(
            new KsqlHostInfoEntity("otherhost:1090"),
            ConsumerCollector.CONSUMER_MESSAGES_PER_SEC,
            x,
            x)
        ).collect(Collectors.toList());
    final List<QueryHostStat> remoteErrors = IntStream.range(0, 5)
        .boxed()
        .map(x -> new QueryHostStat(
            new KsqlHostInfoEntity("otherhost:1090"),
            StreamsErrorCollector.CONSUMER_FAILED_MESSAGES_PER_SEC,
            x,
            x)
        ).collect(Collectors.toList());

    // Given:
    final String kafkaTopicName = "kafka";
    final DataSource dataSource = buildDataSource(kafkaTopicName, Optional.empty());
    final MetricCollectors mock = Mockito.mock(MetricCollectors.class);

    Mockito.when(mock.getAndFormatStatsFor(anyString(), anyBoolean())).thenReturn(mockStringStat);
    Mockito.when(mock.getStatsFor(dataSource.getKafkaTopicName(), true)).thenReturn(errorStats);
    Mockito.when(mock.getStatsFor(dataSource.getKafkaTopicName(), false)).thenReturn(stats);

    // When
    final SourceDescription sourceDescription = SourceDescriptionFactory.create(
        dataSource,
        true,
        Collections.emptyList(),
        Collections.emptyList(),
        Optional.empty(),
        Collections.emptyList(),
        Collections.emptyList(),
        remoteStats.stream(),
        remoteErrors.stream(),
        new KsqlHostInfo("myhost", 10),
        mock
    );

    // Then:
    assertThat(
        remoteStats,
        everyItem(isIn(sourceDescription.getClusterStatistics()))
    );
    assertThat(
        remoteErrors,
        everyItem(isIn(sourceDescription.getClusterErrorStats()))
    );
  }