jprdonnelly/kestra
1
0
Fork 0
mirror of https://github.com/kestra-io/kestra.git synced 2025-12-19 18:05:41 -05:00
Code Issues Packages Projects Releases Wiki Activity

fix(scheduler): better handling of locked triggers (#3603)

Browse Source
This commit is contained in:
YannC
2024-04-22 17:01:26 +02:00
committed by GitHub
parent 0b2015cff6
commit 3f3ce1ce6b
4 changed files with 19 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files

5
core/src/main/java/io/kestra/core/schedulers/AbstractScheduler.java
View File

@@ -373,6 +373,11 @@ public abstract class AbstractScheduler implements Scheduler, Service {
.conditionContext(flowWithTriggers.getConditionContext())
.triggerContext(flowWithTriggers.TriggerContext.toBuilder().date(now()).stopAfter(flowWithTriggers.getAbstractTrigger().getStopAfter()).build())
.build())
.peek(f -> {
if (f.getTriggerContext().getEvaluateRunningDate() != null || isExecutionNotRunning(f)) {
this.triggerState.unlock(f.getTriggerContext());
}
})
.filter(f -> f.getTriggerContext().getEvaluateRunningDate() == null)
.filter(this::isExecutionNotRunning)
.map(FlowWithPollingTriggerNextDate::of)

9
core/src/main/java/io/kestra/core/schedulers/SchedulerTriggerStateInterface.java
View File

@@ -26,6 +26,13 @@ public interface SchedulerTriggerStateInterface {
List<Trigger> findByNextExecutionDateReadyForAllTenants(ZonedDateTime now, ScheduleContextInterface scheduleContext);
// Required for Kafka
/**
* Required for Kafka
*/
List<Trigger> findByNextExecutionDateReadyForGivenFlows(List<Flow> flows, ZonedDateTime now, ScheduleContextInterface scheduleContext);
/**
* Required for Kafka
*/
void unlock(Trigger trigger);
}

3
jdbc/src/main/java/io/kestra/jdbc/runner/JdbcSchedulerTriggerState.java
View File

@@ -84,4 +84,7 @@ public class JdbcSchedulerTriggerState implements SchedulerTriggerStateInterface
public List<Trigger> findByNextExecutionDateReadyForGivenFlows(List<Flow> flows, ZonedDateTime now, ScheduleContextInterface scheduleContext) {
throw new NotImplementedException();
}
@Override
public void unlock(Trigger trigger) {}
}

3
runner-memory/src/main/java/io/kestra/runner/memory/MemorySchedulerTriggerState.java
View File

@@ -79,4 +79,7 @@ public class MemorySchedulerTriggerState implements SchedulerTriggerStateInterfa
public List<Trigger> findByNextExecutionDateReadyForGivenFlows(List<Flow> flows, ZonedDateTime now, ScheduleContextInterface scheduleContext) {
throw new NotImplementedException();
}
@Override
public void unlock(Trigger trigger) {}
}