public void containsInAnyOrder(final Collection<M> expected) {
    LATCHES.putIfAbsent(id, PLACE_HOLDER);
    final MessageStream<M> streamToCheck = checkEachTask
        ? messageStream
        : messageStream
            .partitionBy(m -> null, m -> m, KVSerde.of(new StringSerde(), serde), null)
            .map(kv -> kv.value);

    streamToCheck.sink(new CheckAgainstExpected<M>(id, expected, checkEachTask));
  }
Constructors a MessageStreamAssert with an id and serde
@param id unique id
@param messageStream represents messageStream that you want to assert on
@param serde serde used to desialize messageStream
@param <M> represents type of Message
@return MessageStreamAssert that returns the the messages in the stream
samza-test/src/main/java/org/apache/samza/test/framework/MessageStreamAssert.java