From f29ec9e37471db81eab9972de60e81564d301c4a Mon Sep 17 00:00:00 2001 From: Kai Moritz Date: Thu, 20 Feb 2025 18:36:13 +0100 Subject: [PATCH] =?utf8?q?`ExampleConsumer`=20=C3=BCber=20Generics=20typis?= =?utf8?q?iert?= MIME-Version: 1.0 Content-Type: text/plain; charset=utf8 Content-Transfer-Encoding: 8bit --- .../de/juplo/kafka/ApplicationConfiguration.java | 4 ++-- src/main/java/de/juplo/kafka/ExampleConsumer.java | 14 +++++++------- 2 files changed, 9 insertions(+), 9 deletions(-) diff --git a/src/main/java/de/juplo/kafka/ApplicationConfiguration.java b/src/main/java/de/juplo/kafka/ApplicationConfiguration.java index b98c401d..d2b8e05c 100644 --- a/src/main/java/de/juplo/kafka/ApplicationConfiguration.java +++ b/src/main/java/de/juplo/kafka/ApplicationConfiguration.java @@ -17,13 +17,13 @@ import java.util.Properties; public class ApplicationConfiguration { @Bean - public ExampleConsumer exampleConsumer( + public ExampleConsumer exampleConsumer( Consumer kafkaConsumer, ApplicationProperties properties, ConfigurableApplicationContext applicationContext) { return - new ExampleConsumer( + new ExampleConsumer<>( properties.getClientId(), properties.getConsumerProperties().getTopic(), kafkaConsumer, diff --git a/src/main/java/de/juplo/kafka/ExampleConsumer.java b/src/main/java/de/juplo/kafka/ExampleConsumer.java index 1f5a5706..101abd1c 100644 --- a/src/main/java/de/juplo/kafka/ExampleConsumer.java +++ b/src/main/java/de/juplo/kafka/ExampleConsumer.java @@ -11,11 +11,11 @@ import java.util.Arrays; @Slf4j -public class ExampleConsumer implements Runnable +public class ExampleConsumer implements Runnable { private final String id; private final String topic; - private final Consumer consumer; + private final Consumer consumer; private final Thread workerThread; private final Runnable closeCallback; @@ -25,7 +25,7 @@ public class ExampleConsumer implements Runnable public ExampleConsumer( String clientId, String topic, - Consumer consumer, + Consumer consumer, Runnable closeCallback) { this.id = clientId; @@ -49,10 +49,10 @@ public class ExampleConsumer implements Runnable while (true) { - ConsumerRecords records = consumer.poll(Duration.ofSeconds(1)); + ConsumerRecords records = consumer.poll(Duration.ofSeconds(1)); log.info("{} - Received {} messages", id, records.count()); - for (ConsumerRecord record : records) + for (ConsumerRecord record : records) { handleRecord( record.topic(), @@ -86,8 +86,8 @@ public class ExampleConsumer implements Runnable String topic, Integer partition, Long offset, - String key, - String value) + K key, + V value) { consumed++; log.info("{} - partition={}-{}, offset={}: {}={}", id, topic, partition, offset, key, value); -- 2.20.1