When working with the Kafka Bridge, I found that we could not do the following:

• subscribe to a Kafka topic once the Kafka Bridge service was already started
• there were no warning logs when listening a kafka topic and it was failing because the consumer didn't exist
• creating a consumer might take a while to reconcile in Kafka, and we're creating the consumers using the "latest" offset strategy (which means that when there is no offset, it will only consume the new messages after the consumer creation)
• when using the wait for kafka messages, we need to provide the exact message validator with the message we expect. If no messages are matched, then the framework will keep checking for future messages until the timeout happens.

However, if we might want to expect for a specific message with some identifiers, and then assert the content. In order to do this, we need the waitForKafkaMessage to return the list of messages that were matched.

Example:

UtilizationSummary utilizationSummary = kafkaBridge.waitForKafkaMessage(
        UTILIZATION,
        isUtilizationSummaryByTallyIds(...);
    assertNotNull(utilizationSummary, "Utilization summary should be produced");
    assertEquals(
        subscription.getOrgId(), utilizationSummary.getOrgId(), "Should have correct org ID");
    assertEquals(
        subscription.getProductId(),
        utilizationSummary.getProductId(),
        "Should have correct product ID");
    assertTrue(utilizationSummary.getSubscriptionFound(), "Should find matching subscription");