treshenkov/clearing
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

offset window for transactional services

Browse source
This commit is contained in:
ialbert 2022-09-16 19:15:45 +03:00
parent b0bff28194
commit 4a844dcc43
1 changed files with 6 additions and 1 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

7
platform-parent/platform-messaging/src/main/java/ru/spcex/clearing/platform/messaging/service/OffsetChanger.java
View file

@ -30,6 +30,11 @@ public class OffsetChanger implements ConsumerRebalanceListener {
.entrySet() .entrySet()
.stream() .stream()
.filter(entr -> entr.getValue() != null) .filter(entr -> entr.getValue() != null)
.forEach(entry -> consumer.seek(entry.getKey(), entry.getValue().offset())); .forEach(entry -> {
long offsetAssignedByKafka = consumer.position(entry.getKey());
if (offsetAssignedByKafka > entry.getValue().offset()) {
consumer.seek(entry.getKey(), entry.getValue().offset());
}
});
} }
} }