Messager companions - method for getting all available messages

[GregJohnStewart]

First off, apologies if this functionality exists and couldn't find it, but it seems like the KafkaCompanion (or presumably the other companions) doesn't support awaiting for all available messages, only the specified number of messages. This makes it hard to perform certain test cases.

Of note is I have tried simply awaiting for single messages in a loop, but that ended up with never actually exiting and always returning the same message. This seems infeasible without additional support from the companion.

[ozangunalp]

Sorry for the late reply.

I assume "all available messages" means messages from the beginning of a topic-partition until the tail of a topic-partition.

You can do this by

• First, get the latest offset of all partitions at a point in time, using topics().describe() and offsets().list()
• Then start consuming from the topic by checking that the received records attained captured the latest offset (you should do probably -1 of the latest offset you've captured previously), for all topic-partitions. You can do this using https://javadoc.io/doc/io.smallrye.reactive/smallrye-reactive-messaging-kafka-test-companion/latest/io/smallrye/reactive/messaging/kafka/companion/ConsumerBuilder.html#fromTopics(java.lang.String,java.util.function.Function) and https://javadoc.io/doc/io.smallrye.reactive/smallrye-reactive-messaging-kafka-test-companion/latest/io/smallrye/reactive/messaging/kafka/companion/RecordQualifiers.html#until(java.util.function.Predicate)
  like

companion.consumeIntegers().fromTopics(topic, RecordQualifiers.until(cr -> {
// return true when to stop consuming
        })).awaitCompletion();

Hope this helps

[ozangunalp]

I realize that getting record offsets before the last stable offset is not always easy.

Another practical approach would be to use Mutiny API to look out for the absence of new polled records:

companion.consumeStrings()
    .fromTopics(topic, m -> m.ifNoItem().after(Duration.ofSeconds(3)).recoverWithCompletion())
    .awaitCompletion()