- TestRecordHandler<K, V> captureOffsetAndExecuteTestHandler =
- new TestRecordHandler<K, V>(recordHandler)
- {
- @Override
- public void onNewRecord(ConsumerRecord<K, V> record)
- {
- seenOffsets.put(
- new TopicPartition(record.topic(), record.partition()),
- record.offset());
- receivedRecords.add(record);
- }
- };
-
- endlessConsumer =
- new EndlessConsumer<>(
- executor,
- properties.getClientId(),
- properties.getTopic(),
- kafkaConsumer,
- rebalanceListener,
- captureOffsetAndExecuteTestHandler);
-