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:
7a41d0b
)
Fehlerbehandlung in EndlessConsumer.destroy() korrigiert
author
Kai Moritz
<kai@juplo.de>
Sat, 9 Apr 2022 16:50:43 +0000
(18:50 +0200)
committer
Kai Moritz
<kai@juplo.de>
Sun, 10 Apr 2022 17:02:23 +0000
(19:02 +0200)
src/main/java/de/juplo/kafka/EndlessConsumer.java
patch
|
blob
|
history
diff --git
a/src/main/java/de/juplo/kafka/EndlessConsumer.java
b/src/main/java/de/juplo/kafka/EndlessConsumer.java
index
b3dd446
..
f25e93c
100644
(file)
--- a/
src/main/java/de/juplo/kafka/EndlessConsumer.java
+++ b/
src/main/java/de/juplo/kafka/EndlessConsumer.java
@@
-110,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);
@@
-120,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();
@@
-140,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);