+ void assertSeenOffsetsAreBehindCommittedOffsets(Map<TopicPartition, Long> offsetsToCheck)
+ {
+ List<Boolean> isOffsetBehindSeen = new LinkedList<>();
+
+ doForCurrentOffsets((tp, offset) ->
+ {
+ Long expected = offsetsToCheck.get(tp) + 1;
+ log.debug("Checking, if the offset {} for {} is at most {}", offset, tp, expected);
+ assertThat(offset)
+ .describedAs("Committed offset must be at most equal to the offset of the consumer")
+ .isLessThanOrEqualTo(expected);
+ isOffsetBehindSeen.add(offset < expected);
+ });
+
+ assertThat(isOffsetBehindSeen.stream().reduce(false, (result, next) -> result | next))
+ .describedAs("Committed offsets are behind seen offsets")
+ .isTrue();
+ }
+