Add option to wait for N seconds without new docs

internal/testrunner/runners/system/test_config.go

@@ -53,6 +53,9 @@ type testConfig struct {
 
 		// Field present
 		FieldPresent []string `config:"fields_present"`
+
+		// SecondsWithoutChange
+		SecondsWithoutChange time.Duration `config:"seconds_without_change"`
 	} `config:"assert"`
 
 	// NumericKeywordFields holds a list of fields that have keyword

internal/testrunner/runners/system/tester.go

@@ -1365,13 +1365,18 @@ func (r *tester) prepareScenario(ctx context.Context, config *testConfig, stackC
 	logger.Debugf("checking for expected data in data stream (%s)...", waitForDataTimeout)
 	var hits *hits
 	oldHits := 0
+	prevTime := time.Now()
 	passed, waitErr := wait.UntilTrue(ctx, func(ctx context.Context) (bool, error) {
 		var err error
 		hits, err = r.getDocs(ctx, scenario.dataStream)
 		if err != nil {
 			return false, err
 		}
 
+		defer func() {
+			oldHits = hits.size()
+		}()
+
 		if r.checkFailureStore {
 			failureStore, err := r.getFailureStoreDocs(ctx, scenario.dataStream)
 			if err != nil {
@@ -1391,13 +1396,29 @@ func (r *tester) prepareScenario(ctx context.Context, config *testConfig, stackC
 
 			ret := hits.size() == oldHits
 			if !ret {
-				oldHits = hits.size()
 				time.Sleep(4 * time.Second)
 			}
 
 			return ret, nil
 		}
 
+		if config.Assert.SecondsWithoutChange.Seconds() > 0 {
+			if hits.size() == 0 {
+				// At least there should be one document ingested
+				return false, nil
+			}
+			if oldHits != hits.size() {
+				prevTime = time.Now()
+				return false, nil
+			}
+
+			if time.Since(prevTime) > config.Assert.SecondsWithoutChange {
+				logger.Debugf("No new documents ingested in %s", config.Assert.SecondsWithoutChange)
+				return true, nil
+			}
+			return false, nil
+		}
+
 		if config.Assert.MinCount > 0 {
 			return hits.size() >= config.Assert.MinCount, nil
 		}