fix overzealous start of stream.

DefaultStreamSupervisor started stream without waiting for
stream interval.

Signed-off-by: Cai Yufei (INST/ECS1) <yufei.cai@bosch-si.com>

services/utils/akka/src/main/java/org/eclipse/ditto/services/utils/akka/streaming/DefaultStreamSupervisor.java

@@ -41,6 +41,7 @@
 import akka.japi.pf.DeciderBuilder;
 import akka.japi.pf.ReceiveBuilder;
 import akka.pattern.PatternsCS;
+import akka.stream.DelayOverflowStrategy;
 import akka.stream.Materializer;
 import akka.stream.javadsl.Sink;
 import akka.stream.javadsl.Source;
@@ -179,7 +180,13 @@ public final class DefaultStreamSupervisor<E> extends AbstractActor {
         activityCheck = scheduleActivityCheck(streamConsumerSettings);
         lastStreamStartOrStop = Instant.now();
 
-        computeAndScheduleNextStreamTrigger(null);
+        // schedule first stream after delay
+        PatternsCS.pipe(
+                computeNextStreamTriggerSource(null)
+                        .delay(streamConsumerSettings.getStreamInterval(), DelayOverflowStrategy.backpressure())
+                        .runWith(Sink.head(), materializer),
+                getContext().dispatcher()
+        ).to(getSelf());
     }
 
     /**
@@ -328,6 +335,11 @@ private void computeAndScheduleNextStreamTrigger(@Nullable final Instant lastSuc
     }
 
     private CompletionStage<StreamTrigger> computeNextStreamTrigger(@Nullable final Instant lastSuccessfulQueryEnd) {
+        return computeNextStreamTriggerSource(lastSuccessfulQueryEnd).runWith(Sink.head(), materializer);
+    }
+
+    private Source<StreamTrigger, NotUsed> computeNextStreamTriggerSource(
+            @Nullable final Instant lastSuccessfulQueryEnd) {
         final Instant now = Instant.now();
 
         final Source<Instant, NotUsed> queryStartSource;
@@ -359,7 +371,7 @@ private CompletionStage<StreamTrigger> computeNextStreamTrigger(@Nullable final
                     return StreamTrigger.calculateStreamTrigger(now, queryStart, startOffset, streamInterval);
                 });
 
-        return triggerSource.runWith(Sink.head(), materializer);
+        return triggerSource;
     }
 
     private void scheduleStream(final StreamTrigger streamTrigger) {