@ExecuteOn(TaskExecutors.IO)
    @Post(uri = "/replay/by-ids")
    @Operation(tags = {"Executions"}, summary = "Create new executions from old ones. Keep the flow revision")
    @ApiResponse(responseCode = "200", description = "On success", content = {@Content(schema = @Schema(implementation = BulkResponse.class))})
    @ApiResponse(responseCode = "422", description = "Replayed with errors", content = {@Content(schema = @Schema(implementation = BulkErrorResponse.class))})
    public MutableHttpResponse<?> replayByIds(
        @Parameter(description = "The execution id") @Body List<String> executionsId
    ) throws Exception {
        List<Execution> executions = new ArrayList<>();
        Set<ManualConstraintViolation<String>> invalids = new HashSet<>();

        for (String executionId : executionsId) {
            Optional<Execution> execution = executionRepository.findById(tenantService.resolveTenant(), executionId);
            if (execution.isEmpty()) {
                invalids.add(ManualConstraintViolation.of(
                    "execution not found",
                    executionId,
                    String.class,
                    "execution",
                    executionId
                ));
            } else {
                executions.add(execution.get());
            }
        }

        if (!invalids.isEmpty()) {
            return HttpResponse.badRequest(BulkErrorResponse
                .builder()
                .message("invalid bulk replay")
                .invalids(invalids)
                .build()
            );
        }

        for (Execution execution : executions) {
            Execution replay = executionService.replay(execution, null, null);
            executionQueue.emit(replay);
            eventPublisher.publishEvent(new CrudEvent<>(replay, execution, CrudEventType.CREATE));
        }

        return HttpResponse.ok(BulkResponse.builder().count(executions.size()).build());
    }
@Test
    void replayByIds() throws TimeoutException {
        Execution execution1 = runnerUtils.runOne(null, "io.kestra.tests", "each-sequential-nested");
        Execution execution2 = runnerUtils.runOne(null, "io.kestra.tests", "each-sequential-nested");

        assertThat(execution1.getState().isTerminated(), is(true));
        assertThat(execution2.getState().isTerminated(), is(true));

        PagedResults<?> executions = client.toBlocking().retrieve(
            GET("/api/v1/executions/search"), PagedResults.class
        );
        assertThat(executions.getTotal(), is(2L));

        // replay executions
        BulkResponse replayResponse = client.toBlocking().retrieve(
            HttpRequest.POST(
                "/api/v1/executions/replay/by-ids",
                List.of(execution1.getId(), execution2.getId())
            ),
            BulkResponse.class
        );
        assertThat(replayResponse.getCount(), is(2));

        executions = client.toBlocking().retrieve(
            GET("/api/v1/executions/search"), PagedResults.class
        );
        assertThat(executions.getTotal(), is(4L));
    }