projects
/
demos
/
kafka
/
training
/ commitdiff
commit
grep
author
committer
pickaxe
?
search:
re
summary
|
shortlog
|
log
|
commit
| commitdiff |
tree
raw
|
patch
|
inline
| side by side (parent:
31530b8
)
`auto.offset.reset` konfigurierbar gemacht
author
Kai Moritz
<kai@juplo.de>
Fri, 1 Apr 2022 09:44:22 +0000
(11:44 +0200)
committer
Kai Moritz
<kai@juplo.de>
Fri, 1 Apr 2022 11:54:41 +0000
(13:54 +0200)
src/main/java/de/juplo/kafka/Application.java
patch
|
blob
|
history
src/main/java/de/juplo/kafka/ApplicationProperties.java
patch
|
blob
|
history
src/main/java/de/juplo/kafka/EndlessConsumer.java
patch
|
blob
|
history
src/main/resources/application.yml
patch
|
blob
|
history
diff --git
a/src/main/java/de/juplo/kafka/Application.java
b/src/main/java/de/juplo/kafka/Application.java
index
85d0e07
..
dd4b20a
100644
(file)
--- a/
src/main/java/de/juplo/kafka/Application.java
+++ b/
src/main/java/de/juplo/kafka/Application.java
@@
-32,7
+32,8
@@
public class Application
properties.getBootstrapServer(),
properties.getGroupId(),
properties.getClientId(),
properties.getBootstrapServer(),
properties.getGroupId(),
properties.getClientId(),
- properties.getTopic());
+ properties.getTopic(),
+ properties.getAutoOffsetReset());
consumer.start();
consumer.start();
diff --git
a/src/main/java/de/juplo/kafka/ApplicationProperties.java
b/src/main/java/de/juplo/kafka/ApplicationProperties.java
index
fdbb2bd
..
dab3380
100644
(file)
--- a/
src/main/java/de/juplo/kafka/ApplicationProperties.java
+++ b/
src/main/java/de/juplo/kafka/ApplicationProperties.java
@@
-14,4
+14,5
@@
public class ApplicationProperties
private String groupId;
private String clientId;
private String topic;
private String groupId;
private String clientId;
private String topic;
+ private String autoOffsetReset;
}
}
diff --git
a/src/main/java/de/juplo/kafka/EndlessConsumer.java
b/src/main/java/de/juplo/kafka/EndlessConsumer.java
index
da2f8f0
..
b3dd446
100644
(file)
--- a/
src/main/java/de/juplo/kafka/EndlessConsumer.java
+++ b/
src/main/java/de/juplo/kafka/EndlessConsumer.java
@@
-25,6
+25,7
@@
public class EndlessConsumer implements Runnable
private final String groupId;
private final String id;
private final String topic;
private final String groupId;
private final String id;
private final String topic;
+ private final String autoOffsetReset;
private AtomicBoolean running = new AtomicBoolean();
private long consumed = 0;
private AtomicBoolean running = new AtomicBoolean();
private long consumed = 0;
@@
-36,13
+37,15
@@
public class EndlessConsumer implements Runnable
String bootstrapServer,
String groupId,
String clientId,
String bootstrapServer,
String groupId,
String clientId,
- String topic)
+ String topic,
+ String autoOffsetReset)
{
this.executor = executor;
this.bootstrapServer = bootstrapServer;
this.groupId = groupId;
this.id = clientId;
this.topic = topic;
{
this.executor = executor;
this.bootstrapServer = bootstrapServer;
this.groupId = groupId;
this.id = clientId;
this.topic = topic;
+ this.autoOffsetReset = autoOffsetReset;
}
@Override
}
@Override
@@
-54,7
+57,7
@@
public class EndlessConsumer implements Runnable
props.put("bootstrap.servers", bootstrapServer);
props.put("group.id", groupId);
props.put("client.id", id);
props.put("bootstrap.servers", bootstrapServer);
props.put("group.id", groupId);
props.put("client.id", id);
- props.put("auto.offset.reset",
"earliest"
);
+ props.put("auto.offset.reset",
autoOffsetReset
);
props.put("key.deserializer", StringDeserializer.class.getName());
props.put("value.deserializer", StringDeserializer.class.getName());
props.put("key.deserializer", StringDeserializer.class.getName());
props.put("value.deserializer", StringDeserializer.class.getName());
diff --git
a/src/main/resources/application.yml
b/src/main/resources/application.yml
index
763880d
..
db37822
100644
(file)
--- a/
src/main/resources/application.yml
+++ b/
src/main/resources/application.yml
@@
-3,6
+3,7
@@
consumer:
group-id: my-consumer
client-id: peter
topic: test
group-id: my-consumer
client-id: peter
topic: test
+ auto-offset-reset: earliest
management:
endpoints:
web:
management:
endpoints:
web: