kestra-io / kestra-io/plugin-fs
Add realtime WatchService-based trigger for local filesystem events
- Dominant language
- Java
- Stars
- 7
- Forks
- 21
- Avg merge
- 4d 4m
- Merged PRs (30d)
- 8
Description
## Summary
`io.kestra.plugin.fs.local.Trigger` is polling-only: it wakes up on a fixed `interval` (default 60s) and re-lists the watched directory every time, even when nothing changed. Add a companion `io.kestra.plugin.fs.local.RealtimeTrigger` that registers a Java NIO `WatchService` on the target directory/directories and fires as soon as the filesystem reports a create/modify/delete event — no polling loop, no wasted list calls, near-instant reaction to file changes.
## Motivation
- Today, users who want fast reaction to new files must set a short `interval` (e.g. `PT5S`) on the polling `Trigger`, which means constant directory listing even when idle — wasted I/O and worker cycles.
- `WatchService` (`java.nio.file.WatchService`, part of the JDK since 7) delivers OS-level filesystem events (`inotify` on Linux, `ReadDirectoryChangesW` on Windows, `FSEvents`/kqueue on macOS) instead of re-scanning the directory.
- This plugin already has the realtime-trigger pattern for other transports (`io.kestra.plugin.fs.tcp.RealtimeTrigger`, `io.kestra.plugin.fs.udp.RealtimeTrigger`), both implementing `RealtimeTriggerInterface` with a reactive `Flux`. This issue extends that same pattern to local filesystem events, complementing the existing polling `io.kestra.plugin.fs.local.Trigger` (kept as-is for users who want interval-based batching/actions).
## Context
- Reference implementation to model the polling behavior after: `io.kestra.plugin.fs.local.Trigger` (`src/main/java/io/kestra/plugin/fs/local/Trigger.java`) — reuse its `on` (`CREATE_OR_UPDATE`/etc.), `regExp`, and `recursive` semantics where they make sense for a realtime source.
- Reference implementation to model the realtime/reactive plumbing after: `io.kestra.plugin.fs.tcp.RealtimeTrigger` and `io.kestra.plugin.fs.udp.RealtimeTrigger` (`RealtimeTriggerInterface`, `Flux`, `AtomicBoolean` stop flag, `Publisher`).
- Related closed issues on the polling trigger (do not need to be re-fixed here, just informative): [#208](https://github.com/kestra-io/plugin-fs/issues/208) (regex handling on MOVE), [#353](https://github.com/kestra-io/plugin-fs/issues/353) (count output), [#354](https://github.com/kestra-io/plugin-fs/issues/354) (MOVE returns pre-move path).
## API Reference
- **API**: `java.nio.file.WatchService` / `java.nio.file.WatchKey` / `java.nio.file.StandardWatchEventKinds` — JDK standard library, no external client.
- **Authentication**: none (local filesystem access only).
- **OS backing**: `inotify` (Linux), `ReadDirectoryChangesW` (Windows), polling-based `WatchService` fallback on macOS (JDK does not use native FSEvents for `WatchService`) — worth calling out as a known macOS caveat in the docs.
## Gradle Dependencies
None. `java.nio.file.WatchService` is part of the JDK standard library — no new dependency required.
## Plugin Structure
- **Repository**: `plugin-fs` (existing)
- **Namespace**: `io.kestra.plugin.fs.local`
- **New class**: `io.kestra.plugin.fs.local.RealtimeTrigger`
- **Sub-plugins**: `local` (existing sub-package, no new categories)
- **Categories**: unchanged — keep whatever `@PluginSubGroup` category `io.kestra.plugin.fs.local`'s `package-info.java` already declares.
> **Task class naming**: `RealtimeTrigger` matches the existing convention used by `fs.tcp` and `fs.udp` — do not suffix/prefix with `Local`.
## Suggested Tasks
1. Implement `io.kestra.plugin.fs.local.RealtimeTrigger` implementing `RealtimeTriggerInterface`, `TriggerOutput` (reuse `local.List.Output`/`local.models.File` shapes for parity with the polling trigger's output).
2. Register a `WatchService` on `from` (single directory); if `recursive` is true, register a `WatchKey` per subdirectory (walk tree on start, and register new subdirectories as `ENTRY_CREATE` events for directories arrive).
3. Map `StandardWatchEventKinds.ENTRY_CREATE` / `ENTRY_MODIFY` / `ENTRY_DELETE` to the same `on` filter (`CREATE`, `UPDATE`, `CREATE_OR_UPDATE`, `DELETE`) already used by `local.Trigger`.
4. Apply `regExp` filtering to matched file names, same semantics as `local.Trigger`.
5. Emit one `Execution` per matched event via the `Flux`/`Publisher` pattern from `tcp.RealtimeTrigger`; stop cleanly on trigger cancel/kill (close the `WatchService`, honor the `AtomicBoolean` stop flag).
6. Write unit + integration tests (create/modify/delete a file in a temp dir, assert an execution fires within a bounded time; test `recursive` picking up a new subdirectory).
7. Add/extend `package-info.java` for `io.kestra.plugin.fs.local` if a new category is needed (none expected — reuse existing).
8. Add YAML examples and plugin documentation; document the macOS `WatchService` polling-fallback caveat.
## YAML Examples
### Example 1 — React immediately when a file is created or updated in a directory
```yaml
id: local_realtime_trigger
namespace: company.team
triggers:
- id: watch
type: io.kestra.plugin.fs.local.RealtimeTrigger
from: /data/incoming
on: CREATE_OR_UPDATE
tasks:
- id: log_file
type: io.kestra.plugin.core.log.Log
message: "New file: {{ trigger.file.path }}"
```
### Example 2 — Watch recursively and filter by extension
```yaml
id: local_realtime_recursive
namespace: company.team
triggers:
- id: watch_csv
type: io.kestra.plugin.fs.local.RealtimeTrigger
from: /data/incoming
recursive: true
regExp: ".*\\.csv$"
tasks:
- id: process
type: io.kestra.plugin.core.log.Log
message: "CSV ready: {{ trigger.file.path }}"
```
### Example 3 — React to deletions for cleanup bookkeeping
```yaml
id: local_realtime_delete
namespace: company.team
triggers:
- id: watch_deletes
type: io.kestra.plugin.fs.local.RealtimeTrigger
from: /data/incoming
on: DELETE
tasks:
- id: log_removed
type: io.kestra.plugin.core.log.Log
message: "File removed: {{ trigger.file.path }}"
```
## Acceptance Criteria
### Functional
- [ ] `RealtimeTrigger` fires an execution as soon as a filesystem event is reported, no polling `interval`
- [ ] `recursive` picks up events from subdirectories, including subdirectories created after the trigger starts
- [ ] `regExp` and `on` filters behave the same as the polling `local.Trigger`
- [ ] `WatchService`/`WatchKey`s are closed and no thread is leaked when the trigger stops/is killed
- [ ] Unit + integration tests pass (`./gradlew test`)
- [ ] Build passes (`./gradlew build`)
### Kestra Plugin Coding Standards
- [ ] All new properties use `Property` — no legacy `@PluginProperty(dynamic = true)` on new code
- [ ] Every property and output has a `@Schema` annotation
- [ ] Trigger class carries the five mandatory Lombok annotations (`@SuperBuilder`, `@ToString`, `@EqualsAndHashCode`, `@Getter`, `@NoArgsConstructor`)
- [ ] Logging via `runContext.logger()` only
- [ ] All `Property` fields support Kestra expression language (template rendering)
### Documentation & Structure
- [ ] `@Plugin(examples = ...)` entries each set `full = true` with a complete runnable flow (id + namespace + triggers/tasks)
- [ ] `package-info.java` category unchanged/consistent for `io.kestra.plugin.fs.local`
- [ ] Docs call out the macOS `WatchService` polling-fallback caveat
---
*[View as Artifact](https://claude.ai/artifact/PWNirRoCWQWTNrPhSyU5oh)*
Contributor guide
No contributing guide indexed for this repository
Research direction
Start with src/main/java/io/kestra/plugin/fs/local/Trigger.java and the tcp/udp RealtimeTrigger implementations to understand filtering, reactive plumbing, and lifecycle handling. Add the local RealtimeTrigger plus the requested unit and integration tests for filesystem events, recursion, and cleanup, then run ./gradlew test and ./gradlew build. Done includes YAML examples and documentation covering the macOS WatchService fallback.
Written by the indexing model from the issue text.
Assessment
- Tech stack
- java
- Domain
- backend, operating-systems
- Issue type
- Feature
- Difficulty
- 5/5
- Estimated time
- Over a week
- Activity status
- Stale
- Clarity
- Clearly specified
- Newbie friendliness
- 42/100