@Test
void commitsCurrentOffsetsOnSuccess() throws Exception
{
- int numberOfGeneratedMessages =
- recordGenerator.generate(false, false, messageSender);
+ recordGenerator.generate(false, false, messageSender);
+
+ int numberOfGeneratedMessages = recordGenerator.getNumberOfMessages();
await(numberOfGeneratedMessages + " records received")
.atMost(Duration.ofSeconds(30))
@SkipWhenErrorCannotBeGenerated(poisonPill = true)
void commitsOffsetOfErrorForReprocessingOnDeserializationError()
{
- int numberOfGeneratedMessages =
- recordGenerator.generate(true, false, messageSender);
+ recordGenerator.generate(true, false, messageSender);
+
+ int numberOfGeneratedMessages = recordGenerator.getNumberOfMessages();
await("Consumer failed")
.atMost(Duration.ofSeconds(30))
@SkipWhenErrorCannotBeGenerated(logicError = true)
void commitsOffsetsOfUnseenRecordsOnLogicError()
{
- int numberOfGeneratedMessages =
- recordGenerator.generate(false, true, messageSender);
+ recordGenerator.generate(false, true, messageSender);
+
+ int numberOfGeneratedMessages = recordGenerator.getNumberOfMessages();
await("Consumer failed")
.atMost(Duration.ofSeconds(30))
public interface RecordGenerator
{
- int generate(
+ void generate(
boolean poisonPills,
boolean logicErrors,
Consumer<ProducerRecord<Bytes, Bytes>> messageSender);
+ int getNumberOfMessages();
+ int getNumberOfPoisonPills();
+ int getNumberOfLogicErrors();
+
default boolean canGeneratePoisonPill()
{
return true;