projects
/
demos
/
kafka
/
training
/ blobdiff
commit
grep
author
committer
pickaxe
?
search:
re
summary
|
shortlog
|
log
|
commit
|
commitdiff
|
tree
raw
|
inline
| side by side
Fehler in Logging-Ausgabe 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
2a3445c
..
6e460b4
100644
(file)
--- a/
src/main/java/de/juplo/kafka/EndlessConsumer.java
+++ b/
src/main/java/de/juplo/kafka/EndlessConsumer.java
@@
-35,7
+35,7
@@
public class EndlessConsumer<K, V> implements ConsumerRebalanceListener, Runnabl
private long consumed = 0;
private final Map<Integer, Map<String, Long>> seen = new HashMap<>();
private long consumed = 0;
private final Map<Integer, Map<String, Long>> seen = new HashMap<>();
- private final Map<Integer, Long>
o
ffsets = new HashMap<>();
+ private final Map<Integer, Long>
lastO
ffsets = new HashMap<>();
@Override
@Override
@@
-45,7
+45,7
@@
public class EndlessConsumer<K, V> implements ConsumerRebalanceListener, Runnabl
{
Integer partition = tp.partition();
Long newOffset = consumer.position(tp);
{
Integer partition = tp.partition();
Long newOffset = consumer.position(tp);
- Long oldOffset =
o
ffsets.remove(partition);
+ Long oldOffset =
lastO
ffsets.remove(partition);
log.info(
"{} - removing partition: {}, consumed {} records (offset {} -> {})",
id,
log.info(
"{} - removing partition: {}, consumed {} records (offset {} -> {})",
id,
@@
-80,6
+80,7
@@
public class EndlessConsumer<K, V> implements ConsumerRebalanceListener, Runnabl
.findById(Integer.toString(partition))
.orElse(new StatisticsDocument(partition));
consumer.seek(tp, document.offset);
.findById(Integer.toString(partition))
.orElse(new StatisticsDocument(partition));
consumer.seek(tp, document.offset);
+ lastOffsets.put(partition, document.offset);
seen.put(partition, document.statistics);
});
}
seen.put(partition, document.statistics);
});
}