Repository: kestra-io/kestra
Issue: [TESTING] Flaky log assertions
Labels: bug

### Expected Behavior

We should wait for specific logs since the log buffering is asynchronous due to the queue system.
We need an util to assert it properly

### Actual Behaviour

_No response_

### Steps To Reproduce

_No response_

### Environment Information

//

### Example flow

_No response_

Issue comments:

Pull request review comments:
I don't think it's a good idea to add this in the Worker only for testing purpose.
Moreover it's a memory leak as the list is never purged.
I don't really understand why we don't have this mechanism already in place tbh. I think that's wrong to say that the Worker right now is closing himself, looks more like he is shutting down the whole app so imo we should have this and cancel our subscription instead of shutting down the topic, WDYT ?
Here I did this for the tests since some tests need to be retried but their assert is in the receive method. So I have to stop this continuous assertion for the next test to work, but yeah I'm not a big fan either of adding test-specific code in production code.
Still I think this should be a productionwise thing
I didn't want to break anything here that's why I didn't add the cleanup in the worker itself but it should definitely be done I think
Should be package private
please add a comment explaining why it is needed to use it on tests

Patch that resolved the issue:
diff --git a/core/src/main/java/io/kestra/core/runners/Worker.java b/core/src/main/java/io/kestra/core/runners/Worker.java
index 567e5bde7a..881cf91231 100644
--- a/core/src/main/java/io/kestra/core/runners/Worker.java
+++ b/core/src/main/java/io/kestra/core/runners/Worker.java
@@ -7,46 +7,34 @@
 import com.google.common.collect.ImmutableList;
 import com.google.common.collect.ImmutableMap;
 import com.google.common.hash.Hashing;
-import io.kestra.core.exceptions.InternalException;
-import io.kestra.core.models.conditions.ConditionContext;
-import io.kestra.core.models.executions.Execution;
-import io.kestra.core.models.executions.MetricEntry;
-import io.kestra.core.models.executions.TaskRun;
-import io.kestra.core.models.flows.Flow;
-import io.kestra.core.models.tasks.Task;
-import io.kestra.core.tasks.flows.WorkingDirectory;
-import io.kestra.core.services.WorkerGroupService;
-import io.kestra.core.models.triggers.AbstractTrigger;
-import io.kestra.core.models.triggers.PollingTriggerInterface;
-import io.kestra.core.models.triggers.TriggerContext;
-import io.micronaut.context.ApplicationContext;
-import io.micronaut.core.annotation.Introspected;
-import io.micronaut.inject.qualifiers.Qualifiers;
-import lombok.Getter;
-import lombok.NoArgsConstructor;
-import lombok.Synchronized;
-import lombok.experimental.SuperBuilder;
-import lombok.extern.slf4j.Slf4j;
-import net.jodah.failsafe.Failsafe;
-import net.jodah.failsafe.Timeout;
 import io.kestra.core.exceptions.TimeoutExceededException;
 import io.kestra.core.metrics.MetricRegistry;
-import io.kestra.core.models.executions.ExecutionKilled;
-import io.kestra.core.models.executions.TaskRunAttempt;
+import io.kestra.core.models.executions.*;
 import io.kestra.core.models.flows.State;
 import io.kestra.core.models.tasks.Output;
 import io.kestra.core.models.tasks.RunnableTask;
+import io.kestra.core.models.tasks.Task;
 import io.kestra.core.models.tasks.retrys.AbstractRetry;
+import io.kestra.core.models.triggers.PollingTriggerInterface;
 import io.kestra.core.queues.QueueException;
 import io.kestra.core.queues.QueueFactoryInterface;
 import io.kestra.core.queues.QueueInterface;
 import io.kestra.core.queues.WorkerTaskQueueInterface;
 import io.kestra.core.serializers.JacksonMapper;
+import io.kestra.core.services.WorkerGroupService;
+import io.kestra.core.tasks.flows.WorkingDirectory;
 import io.kestra.core.utils.Await;
 import io.kestra.core.utils.ExecutorsUtils;
+import io.micronaut.context.ApplicationContext;
+import io.micronaut.core.annotation.Introspected;
+import io.micronaut.inject.qualifiers.Qualifiers;
+import lombok.Getter;
+import lombok.Synchronized;
+import lombok.extern.slf4j.Slf4j;
+import net.jodah.failsafe.Failsafe;
+import net.jodah.failsafe.Timeout;
 import org.slf4j.Logger;
 
-import java.io.Closeable;
 import java.io.IOException;
 import java.time.Duration;
 import java.time.ZonedDateTime;
@@ -60,7 +48,7 @@
 
 @Slf4j
 @Introspected
-public class Worker implements Runnable, Closeable {
+public class Worker implements Runnable, AutoCloseable {
     private final static ObjectMapper MAPPER = JacksonMapper.ofJson();
 
     private final ApplicationContext applicationContext;
@@ -74,7 +62,9 @@ public class Worker implements Runnable, Closeable {
     private final MetricRegistry metricRegistry;
 
     private final Set<String> killedExecution = ConcurrentHashMap.newKeySet();
-    private final ExecutorService executors;
+
+    // package private to allow its usage within tests
+    final ExecutorService executors;
 
     @Getter
     private final Map<Long, AtomicInteger> metricRunningCount = new ConcurrentHashMap<>();
@@ -113,7 +103,7 @@ public Worker(ApplicationContext applicationContext, int thread, String workerGr
         this.metricRegistry = applicationContext.getBean(MetricRegistry.class);
 
         ExecutorsUtils executorsUtils = applicationContext.getBean(ExecutorsUtils.class);
-        this.executors = executorsUtils.maxCachedThreadPool(thread,"worker");
+        this.executors = executorsUtils.maxCachedThreadPool(thread, "worker");
 
         WorkerGroupService workerGroupService = applicationContext.getBean(WorkerGroupService.class);
         this.workerGroup = workerGroupService.resolveGroupFromKey(workerGroupKey);
@@ -165,8 +155,7 @@ public void run() {
                         } finally {
                             runContext.cleanup();
                         }
-                    }
-                    else {
+                    } else {
                         throw new RuntimeException("Unable to process the task '" + workerTask.getTask().getId() + "' as it's not a runnable task");
                     }
                 });
@@ -412,7 +401,7 @@ private void logTerminated(WorkerTask workerTask) {
         );
     }
 
-    private void  logError(WorkerTrigger workerTrigger, Throwable e) {
+    private void logError(WorkerTrigger workerTrigger, Throwable e) {
         Logger logger = workerTrigger.getConditionContext().getRunContext().logger();
 
         logger.warn(
@@ -437,7 +426,7 @@ private WorkerTask runAttempt(WorkerTask workerTask) {
 
         Logger logger = runContext.logger();
 
-        if(!(workerTask.getTask() instanceof RunnableTask)) {
+        if (!(workerTask.getTask() instanceof RunnableTask)) {
             // This should never happen but better to deal with it than crashing the Worker
             TaskRunAttempt attempt = TaskRunAttempt.builder().state(new State().withState(State.Type.FAILED)).build();
             List<TaskRunAttempt> attempts = this.addAttempt(workerTask, attempt);
@@ -545,18 +534,18 @@ public AtomicInteger getMetricRunningCount(WorkerTask workerTask) {
 
     @SuppressWarnings("ResultOfMethodCallIgnored")
     @Override
-    public void close() throws IOException {
+    public void close() throws Exception {
         workerTaskQueue.pause();
         executionKilledQueue.pause();
         new Thread(
             () -> {
-            try {
-                this.executors.shutdown();
-                this.executors.awaitTermination(5, TimeUnit.MINUTES);
-            } catch (InterruptedException e) {
-                log.error("Failed to shutdown workers executors", e);
-            }
-        },
+                try {
+                    this.executors.shutdown();
+                    this.executors.awaitTermination(5, TimeUnit.MINUTES);
+                } catch (InterruptedException e) {
+                    log.error("Failed to shutdown workers executors", e);
+                }
+            },
             "worker-shutdown"
         ).start();
 
diff --git a/core/src/main/java/io/kestra/core/utils/TestsUtils.java b/core/src/main/java/io/kestra/core/utils/TestsUtils.java
index 2e4f1857aa..9159ed871a 100644
--- a/core/src/main/java/io/kestra/core/utils/TestsUtils.java
+++ b/core/src/main/java/io/kestra/core/utils/TestsUtils.java
@@ -21,11 +21,12 @@
 import java.io.IOException;
 import java.net.URISyntaxException;
 import java.net.URL;
+import java.time.Duration;
 import java.time.ZonedDateTime;
-import java.util.AbstractMap;
-import java.util.List;
-import java.util.Map;
-import java.util.Objects;
+import java.util.*;
+import java.util.concurrent.TimeoutException;
+import java.util.concurrent.atomic.AtomicReference;
+import java.util.function.Predicate;
 import java.util.stream.Collectors;
 
 abstract public class TestsUtils {
@@ -55,6 +56,32 @@ public static List<LogEntry> filterLogs(List<LogEntry> logs, TaskRun taskRun) {
             .collect(Collectors.toList());
     }
 
+    public static LogEntry awaitLog(List<LogEntry> logs, Predicate<LogEntry> logMatcher) {
+        List<LogEntry> matchingLogs = awaitLogs(logs, logMatcher, (Predicate<Integer>) null);
+        return matchingLogs.isEmpty() ? null : matchingLogs.get(0);
+    }
+
+    public static List<LogEntry> awaitLogs(List<LogEntry> logs, Predicate<LogEntry> logMatcher, Integer exactCount) {
+        return awaitLogs(logs, logMatcher, exactCount::equals);
+    }
+
+    public static List<LogEntry> awaitLogs(List<LogEntry> logs, Predicate<LogEntry> logMatcher, Predicate<Integer> countMatcher) {
+        AtomicReference<List<LogEntry>> matchingLogs = new AtomicReference<>();
+        try {
+            Await.until(() -> {
+                matchingLogs.set(logs.stream().filter(logMatcher).collect(Collectors.toList()));
+                if(countMatcher == null){
+                    return !matchingLogs.get().isEmpty();
+                }
+
+                int matchingLogsCount = matchingLogs.get().size();
+                return countMatcher.test(matchingLogsCount);
+            }, Duration.ofMillis(10), Duration.ofMillis(500));
+        } catch (TimeoutException e) {}
+
+        return matchingLogs.get();
+    }
+
     public static Flow mockFlow() {
         return TestsUtils.mockFlow(Thread.currentThread().getStackTrace()[2]);
     }


Test patch:
diff --git a/core/src/test/java/io/kestra/core/runners/RunContextTest.java b/core/src/test/java/io/kestra/core/runners/RunContextTest.java
index 06ee5f74c0..a435faf3db 100644
--- a/core/src/test/java/io/kestra/core/runners/RunContextTest.java
+++ b/core/src/test/java/io/kestra/core/runners/RunContextTest.java
@@ -3,7 +3,6 @@
 import io.kestra.core.metrics.MetricRegistry;
 import io.kestra.core.models.executions.Execution;
 import io.kestra.core.models.executions.LogEntry;
-import io.kestra.core.models.executions.TaskRunAttempt;
 import io.kestra.core.models.executions.metrics.Counter;
 import io.kestra.core.models.executions.metrics.Timer;
 import io.kestra.core.models.flows.State;
@@ -12,6 +11,8 @@
 import io.kestra.core.storages.StorageInterface;
 import io.kestra.core.utils.TestsUtils;
 import io.micronaut.context.annotation.Property;
+import jakarta.inject.Inject;
+import jakarta.inject.Named;
 import org.exparity.hamcrest.date.ZonedDateTimeMatchers;
 import org.junit.jupiter.api.Test;
 import org.slf4j.event.Level;
@@ -26,13 +27,9 @@
 import java.util.*;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
-import java.util.stream.Collectors;
-import jakarta.inject.Inject;
-import jakarta.inject.Named;
 
 import static org.hamcrest.MatcherAssert.assertThat;
 import static org.hamcrest.Matchers.*;
-import static org.hamcrest.number.OrderingComparison.greaterThan;
 
 @Property(name = "kestra.tasks.tmp-dir.path", value = "/tmp/sub/dir/tmp/")
 class RunContextTest extends AbstractMemoryRunnerTest {
@@ -55,36 +52,35 @@ class RunContextTest extends AbstractMemoryRunnerTest {
     @Test
     void logs() throws TimeoutException {
         List<LogEntry> logs = new ArrayList<>();
-        List<LogEntry> filters;
+        LogEntry matchingLog;
         workerTaskLogQueue.receive(logs::add);
 
         Execution execution = runnerUtils.runOne("io.kestra.tests", "logs");
 
         assertThat(execution.getTaskRunList(), hasSize(3));
 
-        filters = TestsUtils.filterLogs(logs, execution.getTaskRunList().get(0));
-        assertThat(filters, hasSize(1));
-        assertThat(filters.get(0).getLevel(), is(Level.TRACE));
-        assertThat(filters.get(0).getMessage(), is("first t1"));
-
-        filters = TestsUtils.filterLogs(logs, execution.getTaskRunList().get(1));
-        assertThat(filters, hasSize(1));
-        assertThat(filters.get(0).getLevel(), is(Level.WARN));
-        assertThat(filters.get(0).getMessage(), is("second io.kestra.core.tasks.log.Log"));
+        matchingLog = TestsUtils.awaitLog(logs, log -> Objects.equals(log.getTaskRunId(), execution.getTaskRunList().get(0).getId()));
+        assertThat(matchingLog, notNullValue());
+        assertThat(matchingLog.getLevel(), is(Level.TRACE));
+        assertThat(matchingLog.getMessage(), is("first t1"));
 
+        matchingLog = TestsUtils.awaitLog(logs, log -> Objects.equals(log.getTaskRunId(), execution.getTaskRunList().get(1).getId()));
+        assertThat(matchingLog, notNullValue());
+        assertThat(matchingLog.getLevel(), is(Level.WARN));
+        assertThat(matchingLog.getMessage(), is("second io.kestra.core.tasks.log.Log"));
 
-        filters = TestsUtils.filterLogs(logs, execution.getTaskRunList().get(2));
-        assertThat(filters, hasSize(1));
-        assertThat(filters.get(0).getLevel(), is(Level.ERROR));
-        assertThat(filters.get(0).getMessage(), is("third logs"));
+        matchingLog = TestsUtils.awaitLog(logs, log -> Objects.equals(log.getTaskRunId(), execution.getTaskRunList().get(2).getId()));
+        assertThat(matchingLog, notNullValue());
+        assertThat(matchingLog.getLevel(), is(Level.ERROR));
+        assertThat(matchingLog.getMessage(), is("third logs"));
     }
 
     @Test
-    void inputsLarge() throws TimeoutException, InterruptedException {
+    void inputsLarge() throws TimeoutException {
         List<LogEntry> logs = new ArrayList<>();
         workerTaskLogQueue.receive(logs::add);
 
-        char[] chars = new char[1024*11];
+        char[] chars = new char[1024 * 11];
         Arrays.fill(chars, 'a');
 
         Map<String, String> inputs = new HashMap<>(InputsTest.inputs);
@@ -101,12 +97,9 @@ void inputsLarge() throws TimeoutException, InterruptedException {
         assertThat(execution.getState().getCurrent(), is(State.Type.SUCCESS));
         assertThat(execution.getTaskRunList().get(0).getState().getCurrent(), is(State.Type.SUCCESS));
 
-        List<LogEntry> logEntries = logs.stream()
-            .filter(logEntry -> logEntry.getTaskRunId() != null && logEntry.getTaskRunId().equals(execution.getTaskRunList().get(1).getId()))
-            .sorted(Comparator.comparingLong(value -> value.getTimestamp().toEpochMilli()))
-            .collect(Collectors.toList());
+        List<LogEntry> logEntries = TestsUtils.awaitLogs(logs, logEntry -> logEntry.getTaskRunId() != null && logEntry.getTaskRunId().equals(execution.getTaskRunList().get(1).getId()), count -> count > 1);
+        logEntries.sort(Comparator.comparingLong(value -> value.getTimestamp().toEpochMilli()));
 
-        Thread.sleep(100);
         assertThat(logEntries.get(0).getTimestamp().toEpochMilli() + 1, is(logEntries.get(1).getTimestamp().toEpochMilli()));
     }
 
@@ -150,7 +143,7 @@ void largeInput() throws IOException, InterruptedException {
         p.destroy();
 
         URI uri = runContext.putTempFile(path.toFile());
-        assertThat(storageInterface.size(uri), is(size+1));
+        assertThat(storageInterface.size(uri), is(size + 1));
     }
 
     @Test
diff --git a/core/src/test/java/io/kestra/core/runners/TaskDefaultsCaseTest.java b/core/src/test/java/io/kestra/core/runners/TaskDefaultsCaseTest.java
index 9755880766..379b54d037 100644
--- a/core/src/test/java/io/kestra/core/runners/TaskDefaultsCaseTest.java
+++ b/core/src/test/java/io/kestra/core/runners/TaskDefaultsCaseTest.java
@@ -13,6 +13,7 @@
 import io.kestra.core.queues.QueueFactoryInterface;
 import io.kestra.core.queues.QueueInterface;
 import io.kestra.core.services.GraphService;
+import io.kestra.core.utils.TestsUtils;
 import lombok.EqualsAndHashCode;
 import lombok.Getter;
 import lombok.NoArgsConstructor;
@@ -22,6 +23,7 @@
 import java.time.Duration;
 import java.util.ArrayList;
 import java.util.List;
+import java.util.Objects;
 import java.util.concurrent.TimeoutException;
 import java.util.stream.Collectors;
 import java.util.stream.Stream;
@@ -70,7 +72,8 @@ public void invalidTaskDefaults() throws TimeoutException {
         Execution execution = runnerUtils.runOne("io.kestra.tests", "invalid-task-defaults", Duration.ofSeconds(60));
 
         assertThat(execution.getTaskRunList(), hasSize(1));
-        assertThat(logs.stream().filter(logEntry -> logEntry.getMessage().contains("Unrecognized field \"invalid\"")).count(), greaterThan(0L));
+        LogEntry matchingLog = TestsUtils.awaitLog(logs, log -> log.getMessage().contains("Unrecognized field \"invalid\""));
+        assertThat(matchingLog, notNullValue());
     }
 
     @SuperBuilder
diff --git a/core/src/test/java/io/kestra/core/runners/TestMethodScopedWorker.java b/core/src/test/java/io/kestra/core/runners/TestMethodScopedWorker.java
new file mode 100644
index 0000000000..9ab405e36b
--- /dev/null
+++ b/core/src/test/java/io/kestra/core/runners/TestMethodScopedWorker.java
@@ -0,0 +1,25 @@
+package io.kestra.core.runners;
+
+import io.micronaut.context.ApplicationContext;
+
+import java.util.concurrent.TimeUnit;
+
+/**
+ * This worker is a special worker which won't close every queue allowing it to be ran and closed within a test without
+ * preventing the Micronaut context to be used for further tests with queues still being up
+ */
+public class TestMethodScopedWorker extends Worker {
+    public TestMethodScopedWorker(ApplicationContext applicationContext, int thread, String workerGroupKey) {
+        super(applicationContext, thread, workerGroupKey);
+    }
+
+    /**
+     * Override is done to prevent closing the queue. However please note that this is not failsafe because we ideally need
+     * to stop worker's subscriptions to every queue before cutting of the executors pool.
+     */
+    @Override
+    public void close() throws InterruptedException {
+        this.executors.shutdown();
+        this.executors.awaitTermination(60, TimeUnit.SECONDS);
+    }
+}
\ No newline at end of file
diff --git a/core/src/test/java/io/kestra/core/schedulers/AbstractSchedulerTest.java b/core/src/test/java/io/kestra/core/schedulers/AbstractSchedulerTest.java
index 05353827cb..9a194332c0 100644
--- a/core/src/test/java/io/kestra/core/schedulers/AbstractSchedulerTest.java
+++ b/core/src/test/java/io/kestra/core/schedulers/AbstractSchedulerTest.java
@@ -2,16 +2,12 @@
 
 import com.google.common.collect.ImmutableMap;
 import io.kestra.core.models.conditions.ConditionContext;
-import io.kestra.core.models.flows.Input;
-import io.kestra.core.models.flows.TaskDefault;
-import io.kestra.core.models.flows.input.StringInput;
-import io.micronaut.context.ApplicationContext;
-import io.micronaut.test.extensions.junit5.annotation.MicronautTest;
-import lombok.*;
-import lombok.experimental.SuperBuilder;
 import io.kestra.core.models.executions.Execution;
 import io.kestra.core.models.flows.Flow;
+import io.kestra.core.models.flows.Input;
 import io.kestra.core.models.flows.State;
+import io.kestra.core.models.flows.TaskDefault;
+import io.kestra.core.models.flows.input.StringInput;
 import io.kestra.core.models.triggers.AbstractTrigger;
 import io.kestra.core.models.triggers.PollingTriggerInterface;
 import io.kestra.core.models.triggers.TriggerContext;
@@ -19,12 +15,19 @@
 import io.kestra.core.queues.QueueInterface;
 import io.kestra.core.tasks.debugs.Return;
 import io.kestra.core.utils.IdUtils;
+import io.micronaut.context.ApplicationContext;
+import io.micronaut.test.extensions.junit5.annotation.MicronautTest;
+import jakarta.inject.Inject;
+import jakarta.inject.Named;
+import lombok.*;
+import lombok.experimental.SuperBuilder;
 
 import java.time.Duration;
 import java.util.Collections;
 import java.util.List;
 import java.util.Map;
 import java.util.Optional;
+
 import jakarta.inject.Inject;
 import jakarta.inject.Named;
 
@@ -37,11 +40,11 @@ abstract public class AbstractSchedulerTest {
     @Named(QueueFactoryInterface.EXECUTION_NAMED)
     protected QueueInterface<Execution> executionQueue;
 
-    protected  static Flow createFlow(List<AbstractTrigger> triggers) {
+    protected static Flow createFlow(List<AbstractTrigger> triggers) {
         return createFlow(triggers, null);
     }
 
-    protected  static Flow createFlow(List<AbstractTrigger> triggers, List<TaskDefault> list) {
+    protected static Flow createFlow(List<AbstractTrigger> triggers, List<TaskDefault> list) {
         Flow.FlowBuilder<?, ?> flow = Flow.builder()
             .id(IdUtils.create())
             .namespace("io.kestra.unittest")
diff --git a/core/src/test/java/io/kestra/core/schedulers/SchedulerThreadTest.java b/core/src/test/java/io/kestra/core/schedulers/SchedulerThreadTest.java
index 1b7e87ba86..ce4412bd2f 100644
--- a/core/src/test/java/io/kestra/core/schedulers/SchedulerThreadTest.java
+++ b/core/src/test/java/io/kestra/core/schedulers/SchedulerThreadTest.java
@@ -5,9 +5,10 @@
 import io.kestra.core.models.flows.State;
 import io.kestra.core.models.flows.TaskDefault;
 import io.kestra.core.runners.FlowListeners;
+import io.kestra.core.runners.TestMethodScopedWorker;
 import io.kestra.core.runners.Worker;
 import jakarta.inject.Inject;
-import org.junit.jupiter.api.Test;
+import org.junitpioneer.jupiter.RetryingTest;
 
 import java.util.Collections;
 import java.util.List;
@@ -46,7 +47,7 @@ public static Flow createThreadFlow() {
         ));
     }
 
-    @Test
+    @RetryingTest(5)
     void thread() throws Exception {
         // mock flow listeners
         FlowListeners flowListenersServiceSpy = spy(this.flowListenersService);
@@ -64,21 +65,20 @@ void thread() throws Exception {
             .when(schedulerExecutionStateSpy)
             .findById(any());
 
-        // start the worker as it execute polling triggers
-        Worker worker = new Worker(applicationContext, 8, null);
-        worker.run();
-
         // scheduler
-        try (AbstractScheduler scheduler = new DefaultScheduler(
-            applicationContext,
-            flowListenersServiceSpy,
-            schedulerExecutionStateSpy,
-            triggerState
-        )) {
+        try (
+            AbstractScheduler scheduler = new DefaultScheduler(
+                applicationContext,
+                flowListenersServiceSpy,
+                schedulerExecutionStateSpy,
+                triggerState
+            );
+            Worker worker = new TestMethodScopedWorker(applicationContext, 8, null);
+        ) {
             AtomicReference<Execution> last = new AtomicReference<>();
 
             // wait for execution
-            executionQueue.receive(SchedulerThreadTest.class, execution -> {
+            Runnable assertionStop = executionQueue.receive(SchedulerThreadTest.class, execution -> {
                 last.set(execution);
 
                 assertThat(execution.getFlowId(), is(flow.getId()));
@@ -89,8 +89,12 @@ void thread() throws Exception {
                 }
             });
 
+            // start the worker as it execute polling triggers
+            worker.run();
             scheduler.run();
             queueCount.await(1, TimeUnit.MINUTES);
+            // needed for RetryingTest to work since there is no context cleaning between method => we have to clear assertion receiver manually
+            assertionStop.run();
 
             assertThat(last.get().getVariables().get("defaultInjected"), is("done"));
             assertThat(last.get().getVariables().get("counter"), is(3));
@@ -99,5 +103,4 @@ void thread() throws Exception {
             AbstractSchedulerTest.COUNTER = 0;
         }
     }
-
 }
diff --git a/core/src/test/java/io/kestra/core/services/ConditionServiceTest.java b/core/src/test/java/io/kestra/core/services/ConditionServiceTest.java
index da6de410fd..ba212b093c 100644
--- a/core/src/test/java/io/kestra/core/services/ConditionServiceTest.java
+++ b/core/src/test/java/io/kestra/core/services/ConditionServiceTest.java
@@ -85,9 +85,7 @@ void exception() throws InterruptedException {
 
         conditionService.valid(flow, conditions, conditionContext);
 
-        Thread.sleep( 250);
-
-        assertThat(logs.stream().filter(logEntry -> logEntry.getNamespace().equals("io.kestra.core.services.ConditionServiceTest")).count(), greaterThan(0L));
-        assertThat(logs.stream().filter(logEntry -> logEntry.getFlowId().equals("exception")).count(), greaterThan(0L));
+        LogEntry matchingLog = TestsUtils.awaitLog(logs, logEntry -> logEntry.getNamespace().equals("io.kestra.core.services.ConditionServiceTest") && logEntry.getFlowId().equals("exception"));
+        assertThat(matchingLog, notNullValue());
     }
 }
diff --git a/core/src/test/java/io/kestra/core/tasks/flows/EachSequentialTest.java b/core/src/test/java/io/kestra/core/tasks/flows/EachSequentialTest.java
index c9dfd78a04..ea00035541 100644
--- a/core/src/test/java/io/kestra/core/tasks/flows/EachSequentialTest.java
+++ b/core/src/test/java/io/kestra/core/tasks/flows/EachSequentialTest.java
@@ -1,5 +1,6 @@
 package io.kestra.core.tasks.flows;
 
+import io.kestra.core.utils.TestsUtils;
 import org.junit.jupiter.api.Test;
 import io.kestra.core.exceptions.InternalException;
 import io.kestra.core.models.executions.Execution;
@@ -95,7 +96,8 @@ public static void eachNullTest(RunnerUtils runnerUtils, QueueInterface<LogEntry
 
         assertThat(execution.getTaskRunList(), hasSize(1));
         assertThat(execution.getState().getCurrent(), is(State.Type.FAILED));
-        assertThat(logs.stream().filter(logEntry -> logEntry.getMessage().contains("Found '1' null values on Each, with values=[1, null, {key=my-key, value=my-value}]")).count(), greaterThan(0L));
+        LogEntry matchingLog = TestsUtils.awaitLog(logs, logEntry -> logEntry.getMessage().contains("Found '1' null values on Each, with values=[1, null, {key=my-key, value=my-value}]"));
+        assertThat(matchingLog, notNullValue());
     }
 
     @Test
diff --git a/core/src/test/java/io/kestra/core/tasks/flows/TemplateTest.java b/core/src/test/java/io/kestra/core/tasks/flows/TemplateTest.java
index 439dbed4ed..664a935780 100644
--- a/core/src/test/java/io/kestra/core/tasks/flows/TemplateTest.java
+++ b/core/src/test/java/io/kestra/core/tasks/flows/TemplateTest.java
@@ -12,6 +12,7 @@
 import io.kestra.core.runners.ListenersTest;
 import io.kestra.core.runners.RunnerUtils;
 import io.kestra.core.tasks.log.Log;
+import io.kestra.core.utils.TestsUtils;
 import jakarta.inject.Inject;
 import jakarta.inject.Named;
 import org.junit.jupiter.api.Test;
@@ -27,8 +28,7 @@
 import java.util.concurrent.TimeoutException;
 
 import static org.hamcrest.MatcherAssert.assertThat;
-import static org.hamcrest.Matchers.hasSize;
-import static org.hamcrest.Matchers.is;
+import static org.hamcrest.Matchers.*;
 
 public class TemplateTest extends AbstractMemoryRunnerTest {
     @Inject
@@ -65,7 +65,8 @@ public static void withTemplate(RunnerUtils runnerUtils, TemplateRepositoryInter
 
         assertThat(execution.getTaskRunList(), hasSize(4));
         assertThat(execution.getState().getCurrent(), is(State.Type.SUCCESS));
-        assertThat(logs.stream().filter(logEntry -> logEntry.getMessage().equals("myString")).findFirst().orElseThrow().getLevel(), is(Level.ERROR));
+        LogEntry matchingLog = TestsUtils.awaitLog(logs, logEntry -> logEntry.getMessage().equals("myString") && logEntry.getLevel() == Level.ERROR);
+        assertThat(matchingLog, notNullValue());
     }
 
     @Test
@@ -84,7 +85,8 @@ public static void withFailedTemplate(RunnerUtils runnerUtils, TemplateRepositor
 
         assertThat(execution.getTaskRunList(), hasSize(1));
         assertThat(execution.getState().getCurrent(), is(State.Type.FAILED));
-        assertThat(logs.stream().filter(logEntry -> logEntry.getMessage().equals("Can't find flow template 'io.kestra.tests.invalid'")).findFirst().orElseThrow().getLevel(), is(Level.ERROR));
+        LogEntry matchingLog = TestsUtils.awaitLog(logs, logEntry -> logEntry.getMessage().equals("Can't find flow template 'io.kestra.tests.invalid'") && logEntry.getLevel() == Level.ERROR);
+        assertThat(matchingLog, notNullValue());
     }
 
     @Test
diff --git a/core/src/test/java/io/kestra/core/tasks/flows/TimeoutTest.java b/core/src/test/java/io/kestra/core/tasks/flows/TimeoutTest.java
index 50fc2c44d0..fe93ceb7ad 100644
--- a/core/src/test/java/io/kestra/core/tasks/flows/TimeoutTest.java
+++ b/core/src/test/java/io/kestra/core/tasks/flows/TimeoutTest.java
@@ -11,6 +11,7 @@
 import io.kestra.core.services.TaskDefaultService;
 import io.kestra.core.tasks.scripts.Bash;
 import io.kestra.core.utils.IdUtils;
+import io.kestra.core.utils.TestsUtils;
 import jakarta.inject.Inject;
 import jakarta.inject.Named;
 import org.junit.jupiter.api.Test;
@@ -22,7 +23,7 @@
 import java.util.concurrent.TimeoutException;
 
 import static org.hamcrest.MatcherAssert.assertThat;
-import static org.hamcrest.Matchers.is;
+import static org.hamcrest.Matchers.*;
 
 class TimeoutTest extends AbstractMemoryRunnerTest {
     @Inject
@@ -57,6 +58,7 @@ void timeout() throws TimeoutException {
         Execution execution = runnerUtils.runOne(flow.getNamespace(), flow.getId());
 
         assertThat(execution.getState().getCurrent(), is(State.Type.FAILED));
-        assertThat(logs.stream().filter(logEntry -> logEntry.getMessage().contains("Timeout")).count(), is(2L));
+        List<LogEntry> matchingLogs = TestsUtils.awaitLogs(logs, logEntry -> logEntry.getMessage().contains("Timeout"), 2);
+        assertThat(matchingLogs.size(), is(2));
     }
 }
diff --git a/core/src/test/java/io/kestra/core/tasks/flows/VariablesTest.java b/core/src/test/java/io/kestra/core/tasks/flows/VariablesTest.java
index 62d9009314..2fde7e8bb2 100644
--- a/core/src/test/java/io/kestra/core/tasks/flows/VariablesTest.java
+++ b/core/src/test/java/io/kestra/core/tasks/flows/VariablesTest.java
@@ -11,6 +11,7 @@
 
 import java.util.ArrayList;
 import java.util.List;
+import java.util.Objects;
 import java.util.concurrent.TimeoutException;
 
 import jakarta.inject.Inject;
@@ -46,11 +47,14 @@ void invalidVars() throws TimeoutException {
 
         Execution execution = runnerUtils.runOne("io.kestra.tests", "variables-invalid");
 
-        List<LogEntry> filters = TestsUtils.filterLogs(logs, execution.getTaskRunList().get(1));
 
         assertThat(execution.getTaskRunList(), hasSize(2));
         assertThat(execution.getTaskRunList().get(1).getState().getCurrent(), is(State.Type.FAILED));
-        assertThat(filters.stream().filter(logEntry -> logEntry.getMessage().contains("Missing variable: 'inputs' on '{{inputs.invalid}}'")).count(), greaterThan(0L));
+        LogEntry matchingLog = TestsUtils.awaitLog(logs, logEntry ->
+            Objects.equals(logEntry.getTaskRunId(), execution.getTaskRunList().get(1).getId()) &&
+                logEntry.getMessage().contains("Missing variable: 'inputs' on '{{inputs.invalid}}'")
+        );
+        assertThat(matchingLog, notNullValue());
         assertThat(execution.getState().getCurrent(), is(State.Type.FAILED));
     }
 
diff --git a/jdbc/src/test/java/io/kestra/jdbc/runner/JdbcRunnerTest.java b/jdbc/src/test/java/io/kestra/jdbc/runner/JdbcRunnerTest.java
index d524b1271b..4c0151ab57 100644
--- a/jdbc/src/test/java/io/kestra/jdbc/runner/JdbcRunnerTest.java
+++ b/jdbc/src/test/java/io/kestra/jdbc/runner/JdbcRunnerTest.java
@@ -15,9 +15,9 @@
 import jakarta.inject.Inject;
 import jakarta.inject.Named;
 import org.junit.jupiter.api.BeforeAll;
-import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.TestInstance;
+import org.junitpioneer.jupiter.RetryingTest;
 
 import java.io.IOException;
 import java.net.URISyntaxException;
@@ -116,7 +116,7 @@ void parallel() throws TimeoutException, QueueException {
         assertThat(execution.getTaskRunList(), hasSize(8));
     }
 
-    @Test
+    @RetryingTest(5)
     void parallelNested() throws TimeoutException, QueueException {
         Execution execution = runnerUtils.runOne("io.kestra.tests", "parallel-nested", null, null, Duration.ofSeconds(60));