From 6ff05112077432d7538d269e48f8bd49b5a394bc Mon Sep 17 00:00:00 2001
From: Kai Moritz <kai@juplo.de>
Date: Tue, 22 Nov 2022 19:29:32 +0100
Subject: [PATCH] Vorlage
---
pom.xml | 4 ---
.../java/de/juplo/kafka/SimpleConsumer.java | 25 +++++--------------
2 files changed, 6 insertions(+), 23 deletions(-)
diff --git a/pom.xml b/pom.xml
index 985797bc..a07cfaaf 100644
--- a/pom.xml
+++ b/pom.xml
@@ -34,10 +34,6 @@
<build>
<plugins>
- <plugin>
- <groupId>pl.project13.maven</groupId>
- <artifactId>git-commit-id-plugin</artifactId>
- </plugin>
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-dependency-plugin</artifactId>
diff --git a/src/main/java/de/juplo/kafka/SimpleConsumer.java b/src/main/java/de/juplo/kafka/SimpleConsumer.java
index 9741e672..880ed787 100644
--- a/src/main/java/de/juplo/kafka/SimpleConsumer.java
+++ b/src/main/java/de/juplo/kafka/SimpleConsumer.java
@@ -45,28 +45,16 @@ public class SimpleConsumer
try
{
log.info("{} - Subscribing to topic {}", id, topic);
- consumer.subscribe(Arrays.asList(topic));
+ // TODO: subscribe!
running = true;
while (true)
{
- ConsumerRecords<String, String> records =
- consumer.poll(Duration.ofSeconds(1));
-
- log.info("{} - Received {} messages", id, records.count());
- for (ConsumerRecord<String, String> record : records)
- {
- consumed++;
- log.info(
- "{} - {}: {}/{} - {}={}",
- id,
- record.offset(),
- record.topic(),
- record.partition(),
- record.key(),
- record.value()
- );
- }
+ // TODO:
+ // * Poll for new records
+ // * Loop through received records
+ // * Count received records in the variable `consumed`
+ // * "Work" with the data: Log partition/offset/key/value
}
}
catch(WakeupException e)
@@ -76,7 +64,6 @@ public class SimpleConsumer
catch(Exception e)
{
log.error("{} - Unexpected error: {}, unsubscribing!", id, e.toString());
- consumer.unsubscribe();
}
finally
{
--
2.20.1