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 doesExportTopicData() {
    when(schemaRegistryClientMock.getSubjectVersion("testTopic-value", "latest", false))
        .thenReturn(Mono.just(
            new SchemaSubject()
                .schema("\"string\"")
                .schemaType(SchemaType.AVRO)
        ));

    when(schemaRegistryClientMock.getSubjectVersion("testTopic-key", "latest", false))
        .thenReturn(Mono.just(
            new SchemaSubject()
                .schema("\"int\"")
                .schemaType(SchemaType.AVRO)
        ));

    stats = Statistics.empty()
        .toBuilder()
        .topicDescriptions(
            Map.of(
                "testTopic",
                new TopicDescription(
                    "testTopic",
                    false,
                    List.of(
                        new TopicPartitionInfo(
                            0,
                            null,
                            List.of(
                                new Node(1, "host1", 9092),
                                new Node(2, "host2", 9092)
                            ),
                            List.of())
                    ))
            )
        )
        .topicConfigs(
            Map.of(
                "testTopic", List.of(
                    new ConfigEntry(
                        "custom.config",
                        "100500",
                        ConfigEntry.ConfigSource.DYNAMIC_TOPIC_CONFIG,
                        false,
                        false,
                        List.of(),
                        ConfigEntry.ConfigType.INT,
                        null
                    )
                )
            )
        )
        .build();

    StepVerifier.create(topicsExporter.export(cluster))
        .assertNext(entityList -> {
          assertThat(entityList.getItems())
              .hasSize(1);

          DataEntity topicEntity = entityList.getItems().get(0);
          assertThat(topicEntity.getName()).isNotEmpty();
          assertThat(topicEntity.getOddrn())
              .isEqualTo("//kafka/cluster/localhost:19092,localhost:9092/topics/testTopic");
          assertThat(topicEntity.getType()).isEqualTo(DataEntityType.KAFKA_TOPIC);
          assertThat(topicEntity.getMetadata())
              .hasSize(1)
              .singleElement()
              .satisfies(e ->
                  assertThat(e.getMetadata())
                      .containsExactlyInAnyOrderEntriesOf(
                          Map.of(
                              "partitions", 1,
                              "replication_factor", 2,
                              "custom.config", "100500")));

          assertThat(topicEntity.getDataset()).isNotNull();
          assertThat(topicEntity.getDataset().getFieldList())
              .hasSize(4); // 2 field for key, 2 for value
        })
        .verifyComplete();
  }