feat: use normal mode when offset can't be found

telekom · Nov 27, 2024 · 40df23f · 40df23f
1 parent 63bf6a1
commit 40df23f
Show file tree

Hide file tree

Showing 2 changed files with 29 additions and 21 deletions.
diff --git a/src/main/java/de/telekom/horizon/pulsar/api/SseController.java b/src/main/java/de/telekom/horizon/pulsar/api/SseController.java
@@ -8,6 +8,7 @@
 import de.telekom.horizon.pulsar.helper.StreamLimit;
 import de.telekom.horizon.pulsar.service.SseService;
 import lombok.extern.slf4j.Slf4j;
+import org.apache.commons.lang3.StringUtils;
 import org.springframework.http.HttpHeaders;
 import org.springframework.http.HttpStatus;
 import org.springframework.http.MediaType;
@@ -84,7 +85,7 @@ public ResponseEntity<ResponseBodyEmitter> getSseStream(@PathVariable String env
             accept = APPLICATION_STREAM_JSON_VALUE;
         }
 
-        var responseContainer = sseService.startEmittingEvents(environment, subscriptionId, accept, offset != null || includeHttpHeaders, offset, StreamLimit.of(maxNumber, maxMinutes, maxBytes));
+        var responseContainer = sseService.startEmittingEvents(environment, subscriptionId, accept, StringUtils.isNotEmpty(offset) || includeHttpHeaders, offset, StreamLimit.of(maxNumber, maxMinutes, maxBytes));
 
         var responseHeaders = new HttpHeaders();
         responseHeaders.add(HttpHeaders.CONTENT_TYPE, accept);

diff --git a/src/main/java/de/telekom/horizon/pulsar/service/EventMessageSupplier.java b/src/main/java/de/telekom/horizon/pulsar/service/EventMessageSupplier.java
@@ -14,6 +14,7 @@
 import de.telekom.eni.pandora.horizon.model.event.StatusMessage;
 import de.telekom.eni.pandora.horizon.model.event.SubscriptionEventMessage;
 import de.telekom.eni.pandora.horizon.model.meta.EventRetentionTime;
+import de.telekom.eni.pandora.horizon.mongo.model.MessageStateMongoDocument;
 import de.telekom.eni.pandora.horizon.mongo.repository.MessageStateMongoRepo;
 import de.telekom.eni.pandora.horizon.tracing.HorizonTracer;
 import de.telekom.horizon.pulsar.config.PulsarConfig;
@@ -27,14 +28,16 @@
 import lombok.extern.slf4j.Slf4j;
 import org.apache.commons.lang3.StringUtils;
 import org.apache.commons.lang3.tuple.Pair;
-import org.bson.types.ObjectId;
 import org.springframework.data.domain.PageRequest;
 import org.springframework.data.domain.Pageable;
 import org.springframework.data.domain.Sort;
 
 import java.time.Instant;
 import java.time.temporal.ChronoUnit;
-import java.util.*;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Objects;
+import java.util.Optional;
 import java.util.concurrent.ConcurrentLinkedQueue;
 import java.util.function.Supplier;
 
@@ -177,27 +180,31 @@ private void pollMessageStates() {
 
             Pageable pageable = PageRequest.of(0, pulsarConfig.getSseBatchSize(), Sort.by(Sort.Direction.ASC, "timestamp"));
 
+            Optional<MessageStateMongoDocument> offsetMsg = Optional.empty();
             if (StringUtils.isNoneEmpty(currentOffset)) {
-                var offsetMsg = messageStateMongoRepo.findById(currentOffset);
-                if (offsetMsg.isPresent()) {
-                    var offsetTimestamp = offsetMsg.get().getTimestamp();
-
-                    var list = messageStateMongoRepo.findByDeliveryTypeAndSubscriptionIdAndTimestampGreaterThanAsc(
-                                    DeliveryType.SERVER_SENT_EVENT,
-                                    subscriptionId,
-                                    offsetTimestamp,
-                                    pageable
-                            ).stream()
-                            .filter(m -> m.getCoordinates() != null) // we skip messages that refer to -1 partitions and offsets
-                            .toList();
-
-                    messageStates.addAll(list);
-
-                    if (!list.isEmpty()) {
-                        currentOffset = list.getLast().getUuid();
-                    }
+                offsetMsg = messageStateMongoRepo.findById(currentOffset);
+            }
+
+            if (offsetMsg.isPresent()) {
+                var offsetTimestamp = offsetMsg.get().getTimestamp();
+
+                var list = messageStateMongoRepo.findByDeliveryTypeAndSubscriptionIdAndTimestampGreaterThanAsc(
+                                DeliveryType.SERVER_SENT_EVENT,
+                                subscriptionId,
+                                offsetTimestamp,
+                                pageable
+                        ).stream()
+                        .filter(m -> m.getCoordinates() != null) // we skip messages that refer to -1 partitions and offsets
+                        .toList();
+
+                messageStates.addAll(list);
+
+                if (!list.isEmpty()) {
+                    currentOffset = list.getLast().getUuid();
                 }
             } else {
+                currentOffset = null;
+
                 var list = messageStateMongoRepo.findByStatusInAndDeliveryTypeAndSubscriptionIdAsc(
                                 List.of(Status.PROCESSED),
                                 DeliveryType.SERVER_SENT_EVENT,