projects
/
demos
/
kafka
/
training
/ commitdiff
commit
grep
author
committer
pickaxe
?
search:
re
summary
|
shortlog
|
log
|
commit
| commitdiff |
tree
raw
|
patch
|
inline
| side by side (from parent 1:
a226274
)
Log-Meldung für durchgeführte Berechnungen bei Revoke korrigiert
author
Kai Moritz
<kai@juplo.de>
Sat, 20 Aug 2022 16:13:50 +0000
(18:13 +0200)
committer
Kai Moritz
<kai@juplo.de>
Sat, 20 Aug 2022 16:13:50 +0000
(18:13 +0200)
src/main/java/de/juplo/kafka/ApplicationRebalanceListener.java
patch
|
blob
|
history
diff --git
a/src/main/java/de/juplo/kafka/ApplicationRebalanceListener.java
b/src/main/java/de/juplo/kafka/ApplicationRebalanceListener.java
index
32e14e8
..
6776c0d
100644
(file)
--- a/
src/main/java/de/juplo/kafka/ApplicationRebalanceListener.java
+++ b/
src/main/java/de/juplo/kafka/ApplicationRebalanceListener.java
@@
-51,14
+51,14
@@
public class ApplicationRebalanceListener implements PollIntervalAwareConsumerRe
log.info("{} - removing partition: {}", id, partition);
this.partitions.remove(partition);
Map<String, AdderResult> state = recordHandler.removePartition(partition);
log.info("{} - removing partition: {}", id, partition);
this.partitions.remove(partition);
Map<String, AdderResult> state = recordHandler.removePartition(partition);
- for (String
key
: state.keySet())
+ for (String
user
: state.keySet())
{
log.info(
{
log.info(
- "{} -
Seen {} messages for partition={}|key=
{}",
+ "{} -
Calculations for partition={}|user={}:
{}",
id,
id,
- state.get(key),
partition,
partition,
- key);
+ user,
+ state.get(user));
}
Map<String, List<AdderResult>> results = adderResults.removePartition(partition);
stateRepository.save(new StateDocument(partition, state, results));
}
Map<String, List<AdderResult>> results = adderResults.removePartition(partition);
stateRepository.save(new StateDocument(partition, state, results));