projects
/
demos
/
kafka
/
training
/ blobdiff
commit
grep
author
committer
pickaxe
?
search:
re
summary
|
shortlog
|
log
|
commit
|
commitdiff
|
tree
raw
|
inline
| side by side
Fehlerbehandlung in EndlessConsumer.destroy() korrigiert
[demos/kafka/training]
/
src
/
main
/
java
/
de
/
juplo
/
kafka
/
EndlessConsumer.java
diff --git
a/src/main/java/de/juplo/kafka/EndlessConsumer.java
b/src/main/java/de/juplo/kafka/EndlessConsumer.java
index
da2f8f0
..
f25e93c
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());
@@
-107,7
+110,7
@@
public class EndlessConsumer implements Runnable
{
boolean stateChanged = running.compareAndSet(false, true);
if (!stateChanged)
{
boolean stateChanged = running.compareAndSet(false, true);
if (!stateChanged)
- throw new
Runtim
eException("Consumer instance " + id + " is already running!");
+ throw new
IllegalStat
eException("Consumer instance " + id + " is already running!");
log.info("{} - Starting - consumed {} messages before", id, consumed);
future = executor.submit(this);
log.info("{} - Starting - consumed {} messages before", id, consumed);
future = executor.submit(this);
@@
-117,7
+120,7
@@
public class EndlessConsumer implements Runnable
{
boolean stateChanged = running.compareAndSet(true, false);
if (!stateChanged)
{
boolean stateChanged = running.compareAndSet(true, false);
if (!stateChanged)
- throw new
Runtim
eException("Consumer instance " + id + " is not running!");
+ throw new
IllegalStat
eException("Consumer instance " + id + " is not running!");
log.info("{} - Stopping", id);
consumer.wakeup();
log.info("{} - Stopping", id);
consumer.wakeup();
@@
-137,6
+140,10
@@
public class EndlessConsumer implements Runnable
{
log.info("{} - Was already stopped", id);
}
{
log.info("{} - Was already stopped", id);
}
+ catch (Exception e)
+ {
+ log.error("{} - Unexpected exception while trying to stop the consumer", id, e);
+ }
finally
{
log.info("{}: Consumed {} messages in total, exiting!", id, consumed);
finally
{
log.info("{}: Consumed {} messages in total, exiting!", id, consumed);