public class ApplicationConfiguration
{
@Bean
- public Consumer<ConsumerRecord<String, String>> consumer()
- {
- return (record) ->
- {
- // Handle record
- };
- }
-
- @Bean
- public EndlessConsumer<String, String> endlessConsumer(
+ public EndlessConsumer endlessConsumer(
KafkaConsumer<String, String> kafkaConsumer,
ExecutorService executor,
- Consumer<ConsumerRecord<String, String>> handler,
PartitionStatisticsRepository repository,
ApplicationProperties properties)
{
return
- new EndlessConsumer<>(
+ new EndlessConsumer(
executor,
repository,
properties.getClientId(),
properties.getTopic(),
Clock.systemDefaultZone(),
properties.getCommitInterval(),
- kafkaConsumer,
- handler);
+ kafkaConsumer);
}
@Bean