* Ein Handler für die Verarbeitung der einzelnen ConsumerRecord's kann
jetzt ein `java.util.function.Consumer` als Bean definiert werden
* Die Nachricht wird erst nach dem Aufruf des Handlers als konsumiert
gezählt, damit der Handler die Nachricht über eine Exception ablehnen
kann.
+import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.serialization.LongDeserializer;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.serialization.LongDeserializer;
import org.apache.kafka.common.serialization.StringDeserializer;
import java.util.Properties;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.Properties;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
+import java.util.function.Consumer;
@Configuration
@EnableConfigurationProperties(ApplicationProperties.class)
public class ApplicationConfiguration
{
@Configuration
@EnableConfigurationProperties(ApplicationProperties.class)
public class ApplicationConfiguration
{
+ @Bean
+ public Consumer<ConsumerRecord<String, String>> consumer()
+ {
+ return (record) ->
+ {
+ // Handle record
+ };
+ }
+
@Bean
public EndlessConsumer endlessConsumer(
KafkaConsumer<String, String> kafkaConsumer,
ExecutorService executor,
@Bean
public EndlessConsumer endlessConsumer(
KafkaConsumer<String, String> kafkaConsumer,
ExecutorService executor,
+ Consumer<ConsumerRecord<String, String>> handler,
ApplicationProperties properties)
{
return
ApplicationProperties properties)
{
return
executor,
properties.getClientId(),
properties.getTopic(),
executor,
properties.getClientId(),
properties.getTopic(),
+ kafkaConsumer,
+ handler);
private final String id;
private final String topic;
private final Consumer<String, String> consumer;
private final String id;
private final String topic;
private final Consumer<String, String> consumer;
+ private final java.util.function.Consumer<ConsumerRecord<String, String>> handler;
private final Lock lock = new ReentrantLock();
private final Condition condition = lock.newCondition();
private final Lock lock = new ReentrantLock();
private final Condition condition = lock.newCondition();
log.info("{} - Received {} messages", id, records.count());
for (ConsumerRecord<String, String> record : records)
{
log.info("{} - Received {} messages", id, records.count());
for (ConsumerRecord<String, String> record : records)
{
log.info(
"{} - {}: {}/{} - {}={}",
id,
log.info(
"{} - {}: {}/{} - {}={}",
id,
+ handler.accept(record);
+
+ consumed++;
+
Integer partition = record.partition();
String key = record.key() == null ? "NULL" : record.key();
Map<String, Long> byKey = seen.get(partition);
Integer partition = record.partition();
String key = record.key() == null ? "NULL" : record.key();
Map<String, Long> byKey = seen.get(partition);