@Autowired
ExecutorService executor;
@Autowired
- KeyCountingRebalanceListener keyCountingRebalanceListener;
+ ApplicationRebalanceListener rebalanceListener;
@Autowired
- KeyCountingRecordHandler keyCountingRecordHandler;
+ ApplicationRecordHandler recordHandler;
EndlessConsumer<String, Long> endlessConsumer;
Map<TopicPartition, Long> oldOffsets;
});
TestRecordHandler<String, Long> captureOffsetAndExecuteTestHandler =
- new TestRecordHandler<String, Long>(keyCountingRecordHandler) {
+ new TestRecordHandler<String, Long>(recordHandler) {
@Override
public void onNewRecord(ConsumerRecord<String, Long> record)
{
properties.getClientId(),
properties.getTopic(),
kafkaConsumer,
- keyCountingRebalanceListener,
+ rebalanceListener,
captureOffsetAndExecuteTestHandler);
endlessConsumer.start();