From 4a844dcc4379d7e1c703badb5442cdcf0fbd00aa Mon Sep 17 00:00:00 2001
From: ialbert <ialbert@mfd.ru>
Date: Fri, 16 Sep 2022 19:15:45 +0300
Subject: [PATCH] offset window for transactional services

---
 .../clearing/platform/messaging/service/OffsetChanger.java | 7 ++++++-
 1 file changed, 6 insertions(+), 1 deletion(-)

diff --git a/platform-parent/platform-messaging/src/main/java/ru/spcex/clearing/platform/messaging/service/OffsetChanger.java b/platform-parent/platform-messaging/src/main/java/ru/spcex/clearing/platform/messaging/service/OffsetChanger.java
index 9995a51d3..e6339a9df 100644
--- a/platform-parent/platform-messaging/src/main/java/ru/spcex/clearing/platform/messaging/service/OffsetChanger.java
+++ b/platform-parent/platform-messaging/src/main/java/ru/spcex/clearing/platform/messaging/service/OffsetChanger.java
@@ -30,6 +30,11 @@ public class OffsetChanger implements ConsumerRebalanceListener {
                 .entrySet()
                 .stream()
                 .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());
+                    }
+                });
     }
 }