public static <StreamMessageType> void containsInAnyOrder(Map<Integer, List<StreamMessageType>> expected,
      InMemoryOutputDescriptor<StreamMessageType> outputDescriptor, Duration timeout) throws InterruptedException {
    Preconditions.checkNotNull(outputDescriptor);
    Map<Integer, List<StreamMessageType>> actual = TestRunner.consumeStream(outputDescriptor, timeout);
    for (Integer paritionId : expected.keySet()) {
      assertThat(actual.get(paritionId),
          IsIterableContainingInAnyOrder.containsInAnyOrder(expected.get(paritionId).toArray()));
    }
  }
Verifies that the {@code expected} messages are present in any order in the multi partition stream
represented by {@code outputDescriptor}

@param expected map of partitionId as key and list of messages in stream as value
@param outputDescriptor describes the stream which will be consumed to compare against expected partition map
@param timeout maximum time to wait for consuming the stream
@param <StreamMessageType>  type of messages in the stream
@throws InterruptedException when {@code consumeStream} is interrupted by another thread during polling messages
samza-test/src/main/java/org/apache/samza/test/framework/StreamAssert.java