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:
15d1ea6
)
Log-Meldungen bei Assign/Revoke verbessert und angeglichen
author
Kai Moritz
<kai@juplo.de>
Sun, 21 Aug 2022 16:10:09 +0000
(18:10 +0200)
committer
Kai Moritz
<kai@juplo.de>
Sun, 28 Aug 2022 15:51:35 +0000
(17:51 +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
cd9da64
..
109b205
100644
(file)
--- a/
src/main/java/de/juplo/kafka/ApplicationRebalanceListener.java
+++ b/
src/main/java/de/juplo/kafka/ApplicationRebalanceListener.java
@@
-35,19
+35,33
@@
public class ApplicationRebalanceListener implements PollIntervalAwareConsumerRe
partitions.forEach(tp ->
{
Integer partition = tp.partition();
partitions.forEach(tp ->
{
Integer partition = tp.partition();
+ log.info("{} - adding partition: {}", id, partition);
this.partitions.add(partition);
StateDocument document =
stateRepository
.findById(Integer.toString(partition))
.orElse(new StateDocument(partition));
this.partitions.add(partition);
StateDocument document =
stateRepository
.findById(Integer.toString(partition))
.orElse(new StateDocument(partition));
- log.info("{} - adding partition: {}, offset={}", id, partition, document.offset);
if (document.offset >= 0)
{
// Only seek, if a stored offset was found
// Otherwise: Use initial offset, generated by Kafka
consumer.seek(tp, document.offset);
if (document.offset >= 0)
{
// Only seek, if a stored offset was found
// Otherwise: Use initial offset, generated by Kafka
consumer.seek(tp, document.offset);
+ log.info(
+ "{} - Seeking to offset {} for partition {}",
+ id,
+ document.offset,
+ partition);
}
recordHandler.addPartition(partition, document.state);
}
recordHandler.addPartition(partition, document.state);
+ for (String user : document.state.keySet())
+ {
+ log.info(
+ "{} - Restored state for partition={}|user={}: {}",
+ id,
+ partition,
+ user,
+ document.state.get(user));
+ }
adderResults.addPartition(partition, document.results);
});
}
adderResults.addPartition(partition, document.results);
});
}
@@
-58,16
+72,26
@@
public class ApplicationRebalanceListener implements PollIntervalAwareConsumerRe
partitions.forEach(tp ->
{
Integer partition = tp.partition();
partitions.forEach(tp ->
{
Integer partition = tp.partition();
+ log.info("{} - removing partition: {}", id, partition);
this.partitions.remove(partition);
Long offset = consumer.position(tp);
this.partitions.remove(partition);
Long offset = consumer.position(tp);
- log.info(
- "{} - removing partition: {}, offset of next message {})",
- id,
- partition,
- offset);
if (commitsEnabled)
{
if (commitsEnabled)
{
+ log.info(
+ "{} - Storing {} as offset of next message for partition {}",
+ id,
+ offset,
+ partition);
Map<String, AdderResult> state = recordHandler.removePartition(partition);
Map<String, AdderResult> state = recordHandler.removePartition(partition);
+ for (String user : state.keySet())
+ {
+ log.info(
+ "{} - Saved state for partition={}|user={}: {}",
+ id,
+ partition,
+ user,
+ state.get(user));
+ }
Map<String, List<AdderResult>> results = adderResults.removePartition(partition);
stateRepository.save(new StateDocument(partition, state, results, offset));
}
Map<String, List<AdderResult>> results = adderResults.removePartition(partition);
stateRepository.save(new StateDocument(partition, state, results, offset));
}