[8.19](backport #46576) [Flaky Test] adjust timing of TestFilestreamTruncateBlockedOutput

filebeat/input/filestream/environment_test.go

@@ -30,6 +30,7 @@
 	"testing"
 	"time"
 
+	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 
 	loginp "github.com/elastic/beats/v7/filebeat/input/filestream/internal/input-logfile"
@@ -68,7 +69,7 @@
 }
 
 func newInputTestingEnvironment(t *testing.T) *inputTestingEnvironment {
 	logp.DevelopmentSetup(logp.ToObserverOutput())
 
 	t.Cleanup(func() {
 		if t.Failed() {
@@ -229,33 +230,26 @@
 // requireOffsetInRegistry checks if the expected offset is set for a file.
 func (e *inputTestingEnvironment) requireOffsetInRegistry(filename, inputID string, expectedOffset int) {
 	e.t.Helper()
-	var offsetStr strings.Builder
+	require.EventuallyWithT(e.t, func(ct *assert.CollectT) {
+		var offsetStr strings.Builder
 
-	filepath := e.abspath(filename)
-	fi, err := os.Stat(filepath)
-	if err != nil {
-		e.t.Fatalf("cannot stat file when cheking for offset: %+v", err)
-	}
+		filepath := e.abspath(filename)
+		fi, err := os.Stat(filepath)
+		assert.NoError(ct, err, "cannot stat file when checking for offset")
 
-	id := getIDFromPath(filepath, inputID, fi)
-	var entry registryEntry
-	require.Eventuallyf(e.t, func() bool {
+		id := getIDFromPath(filepath, inputID, fi)
+		var entry registryEntry
 		offsetStr.Reset()
 
 		entry, err = e.getRegistryState(id)
-		if err != nil {
-			e.t.Fatalf("could not get state for '%s' from registry, err: %s", id, err)
-		}
+		assert.NoError(ct, err, "error getting state for ID '%s' from the registry", id)
 
 		fmt.Fprint(&offsetStr, entry.Cursor.Offset)
-
-		return expectedOffset == entry.Cursor.Offset
+		assert.Equal(ct, expectedOffset, entry.Cursor.Offset, "expected offset does not match")
 	},
 		10*time.Second,
 		100*time.Millisecond,
-		"expected offset: '%d', cursor offset: '%s'",
-		expectedOffset,
-		&offsetStr)
+		"failed to get expected registry offset")
 }
 
 // requireMetaInRegistry checks if the expected metadata is saved to the registry.
@@ -518,7 +512,7 @@
 	messages := make([]string, 0)
 	for _, c := range e.pipeline.clients {
 		for _, evt := range c.GetEvents() {
 			messages = append(messages, evt.Fields["message"].(string))
 		}
 	}
 	return messages