public Flux<TopicMessageEventDTO> loadMessages(KafkaCluster cluster,
                                                 String topic,
                                                 ConsumerPosition consumerPosition,
                                                 @Nullable String containsStringFilter,
                                                 @Nullable String filterId,
                                                 @Nullable Integer limit,
                                                 @Nullable String keySerde,
                                                 @Nullable String valueSerde) {
    return loadMessages(
        cluster,
        topic,
        deserializationService.deserializerFor(cluster, topic, keySerde, valueSerde),
        consumerPosition,
        getMsgFilter(containsStringFilter, filterId),
        fixPageSize(limit)
    );
  }
@Test
  void loadMessagesReturnsExceptionWhenTopicNotFound() {
    StepVerifier.create(messagesService
            .loadMessages(cluster, NON_EXISTING_TOPIC,
                new ConsumerPosition(PollingModeDTO.TAILING, NON_EXISTING_TOPIC, List.of(), null, null),
                null, null, 1, "String", "String"))
        .expectError(TopicNotFoundException.class)
        .verify();
  }