@Autowired
MongoProperties mongoProperties;
@Autowired
- ConsumerRebalanceListener rebalanceListener;
- @Autowired
RecordHandler<K, V> recordHandler;
KafkaProducer<Bytes, Bytes> testRecordProducer;
Long expected = offsetsToCheck.get(tp) + 1;
log.debug("Checking, if the offset {} for {} is at most {}", offset, tp, expected);
assertThat(offset)
- .describedAs("Committed offset corresponds to the offset of the consumer")
+ .describedAs("Committed offset must be at most equal to the offset of the consumer")
.isLessThanOrEqualTo(expected);
isOffsetBehindSeen.add(offset < expected);
});
properties.getClientId(),
properties.getTopic(),
kafkaConsumer,
- rebalanceListener,
captureOffsetAndExecuteTestHandler);
endlessConsumer.start();