@Autowired
KafkaListenerEndpointRegistry kafkaListenerEndpointRegistry;
@Autowired
- TestRecordHandler<K, V> recordHandler;
+ TestRecordHandler recordHandler;
@Autowired
DriverController driverController;
await(numberOfGeneratedMessages + " records received")
.atMost(Duration.ofSeconds(30))
.pollInterval(Duration.ofSeconds(1))
- .until(() -> recordHandler.receivedRecords.size() >= numberOfGeneratedMessages);
+ .until(() -> recordHandler.receivedMessages.size() >= numberOfGeneratedMessages);
await("Offsets committed")
.atMost(Duration.ofSeconds(10))
checkSeenOffsetsForProgress();
assertSeenOffsetsEqualCommittedOffsets(recordHandler.seenOffsets);
- assertThat(recordHandler.receivedRecords.size())
+ assertThat(recordHandler.receivedMessages.size())
.describedAs("Received not all sent events")
.isLessThan(numberOfGeneratedMessages);
oldOffsets = new HashMap<>();
recordHandler.seenOffsets = new HashMap<>();
- recordHandler.receivedRecords = new HashSet<>();
+ recordHandler.receivedMessages = new HashSet<>();
doForCurrentOffsets((tp, offset) ->
{