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 shouldReturnLocalStatsBasedOnKafkaTopic() {
    // 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);
    KsqlHostInfo localhost = new KsqlHostInfo("myhost", 10);
    // When
    final SourceDescription sourceDescription = SourceDescriptionFactory.create(
        dataSource,
        true,
        Collections.emptyList(),
        Collections.emptyList(),
        Optional.empty(),
        Collections.emptyList(),
        Collections.emptyList(),
        Stream.empty(),
        Stream.empty(),
        localhost,
        mock
    );

    // Then:
    // TODO deprecate and remove
    assertThat(
        sourceDescription.getStatistics(),
        containsString(mockStringStat));
    assertThat(
        sourceDescription.getErrorStats(),
        containsString(mockStringStat));
    // Also check includes its own stats in cluster stats
    final Stream<QueryHostStat> localStats = stats.stream()
        .map((s) -> QueryHostStat.fromStat(s, new KsqlHostInfoEntity(localhost)));
    assertThat(
        localStats.collect(Collectors.toList()),
        everyItem(isIn(sourceDescription.getClusterStatistics()))
    );

    final Stream<QueryHostStat> localErrors = errorStats.stream()
        .map((s) -> QueryHostStat.fromStat(s, new KsqlHostInfoEntity(localhost)));
    assertThat(
        localErrors.collect(Collectors.toList()),
        everyItem(isIn(sourceDescription.getClusterErrorStats()))
    );
  }