Flux<DataEntityList> export(KafkaCluster cluster) {
    String clusterOddrn = Oddrn.clusterOddrn(cluster);
    Statistics stats = statisticsCache.get(cluster);
    return Flux.fromIterable(stats.getTopicDescriptions().keySet())
        .filter(topicFilter)
        .flatMap(topic -> createTopicDataEntity(cluster, topic, stats))
        .onErrorContinue(
            (th, topic) -> log.warn("Error exporting data for topic {}, cluster {}", topic, cluster.getName(), th))
        .buffer(100)
        .map(topicsEntities ->
            new DataEntityList()
                .dataSourceOddrn(clusterOddrn)
                .items(topicsEntities));
  }
@Test
  void doesNotExportTopicsWhichDontFitFiltrationRule() {
    when(schemaRegistryClientMock.getSubjectVersion(anyString(), anyString(), anyBoolean()))
        .thenReturn(Mono.error(WebClientResponseException.create(404, "NF", new HttpHeaders(), null, null, null)));
    stats = Statistics.empty()
        .toBuilder()
        .topicDescriptions(
            Map.of(
                "_hidden", new TopicDescription("_hidden", false, List.of(
                    new TopicPartitionInfo(0, null, List.of(), List.of())
                )),
                "visible", new TopicDescription("visible", false, List.of(
                    new TopicPartitionInfo(0, null, List.of(), List.of())
                ))
            )
        )
        .build();

    StepVerifier.create(topicsExporter.export(cluster))
        .assertNext(entityList -> {
          assertThat(entityList.getDataSourceOddrn())
              .isNotEmpty();

          assertThat(entityList.getItems())
              .hasSize(1)
              .allSatisfy(e -> e.getOddrn().contains("visible"));
        })
        .verifyComplete();
  }