Watch Files

Local watching uses one WatchService and no auxiliary threads. Existing subdirectories are registered before the initial scan. Newly created or moved directory trees trigger priority reconciliation and registration. An overflow, invalid WatchKey or excess native hints requests priority reconciliation. Ordinary file hints read only the affected file’s metadata. The normal Local Hop Engine gives this transform its own execution thread; the Single Threaded engine is excluded.

Remote schemes use Hop’s installed VFS providers; this plugin creates no SFTP, FTP or cloud clients. Root resolution uses HopVfs.getFileObject(resolvedPath, variables), preserving execution variables, metadata-driven connections and VFS namespaces. Subsequent operations resolve relative to that same root FileObject. Named connections are recommended, particularly on Hop Server. Relative paths follow the existing HopVfs working-directory rules; use an explicit variable such as ${PROJECT_HOME} for a project-relative path.

A candidate holds the latest size and last-modified timestamp and the previously acknowledged version. Repeated operating-system hints for an unchanged version do not restart the stability interval. A changed size or timestamp resets consecutive successful checks to zero. The configured number of unchanged observations, separated by at least the stability interval, is required before emission. Minimum age is measured from last modification and applies even if stability checking is disabled. A fresh metadata read is performed before each emission. An unacknowledged new file remains CREATED while it grows; an acknowledged file becomes MODIFIED. Files which disappear before their first emission are removed from the candidate set.

This is a readiness heuristic, not a writer lock: a writer can pause longer than the configured interval or resume after emission. For a strict producer protocol, write to an excluded temporary name and rename to a matching final filename. Files with a future last-modified timestamp wait until they meet minimum age. Keep host clocks synchronized.

Polling, reconciliation, checkpoints and stability intervals use elapsed monotonic time. Wall-clock adjustments do not advance these intervals. Minimum age and output dates still use wall time and filesystem timestamps. The polling/reconciliation interval starts after the previous scan completes, leaving time for readiness checks and emission even when a scan exceeds its interval.

Keep the generated or existing Watch ID stable, independent of the transform name. It is not regenerated at execution time. The state directory must be local and outside the watched tree. One process holds an exclusive OS file lock per Watch ID for the entire run. Another transform, pipeline or JVM using the same state directory and ID fails instead of racing.

Each versioned checkpoint contains Watch ID, resolved root, recursion/filter scope and metadata for each acknowledged URI. Schema 2 additionally records semantics=OBSERVED, a checkpoint generation and wall-clock save time. Schema 1 is read without losing observations; the first subsequent save archives its original artifacts and upgrades it to schema 2. Unknown future schemas are rejected. Changing the root, recursion or resolved patterns while reusing an ID fails safely; choose a new ID for a new scope. For example, to expand the CSV sample to all files, clear both filename filters and use a new Watch ID such as manual-all-files. An empty include filter includes all filenames. With Pattern syntax = WILDCARD, also includes all files, *test selects names containing test, .csv selects CSV files and file?.txt selects a single character after file. With *Pattern syntax = REGEXP, . includes all files; alone is invalid. Existing pipelines that do not contain a patternSyntax setting retain their regular-expression interpretation and checkpoint scope. Variables are resolved before pattern conversion. Checkpoint scope uses the resolved effective regular expressions, so a syntax change that changes the matched scope requires a new ID or an explicit reset. The Reprocess files pattern always uses a regular expression regardless of filename pattern syntax. If all files already present must also be emitted, select EMIT_EXISTING before the first run with the new ID. Alternatively, stop the pipeline and use Maintenance → Advanced recovery → Clear saved history to archive and reset the existing ID explicitly before changing its scope. The startup error distinguishes a changed Directory from changed filters or recursion; neither mismatch means the saved JSON is corrupt. Event enablement can be changed without changing scope. Disabled events still advance observed state.

All initial-scan modes compare against an existing valid checkpoint. Without a checkpoint:

Example: baseline A and B; emit C; stop; create D offline; restart. The initial comparison emits D and retains A, B and C without repeating them.

Rows are acknowledged in memory only after putRow returns successfully and the pipeline remains running. Checkpoints run periodically and during disposal. Pending unstable candidates are never checkpointed as emitted. The checkpoint describes the source’s observed version handed to Hop, not completion by downstream transforms. An abrupt crash before the next checkpoint can repeat rows. A downstream failure after a successful checkpoint can require explicit replay; there is no downstream acknowledgement transaction. This is recoverable filesystem-state detection with possible redelivery, not end-to-end exactly-once or a guaranteed durable event log.

The checkpoint is written to a temporary sibling file, fully closed and forced to storage, then replaced using ATOMIC_MOVE. NIO is intentionally used for this local persistence operation: a generic VFS copy/delete move does not provide checkpoint atomicity.

If atomic replacement is unsupported, the previous checkpoint is copied to a forced .bak file and a forced .commit intent is created before replacement. A completed fallback removes the intent. An interrupted fallback leaves the backup and intent; initialization and further saves refuse to proceed. Use Restore backup under Advanced recovery in the Maintenance tab to validate a .json.bak or archived .json, archive current artifacts and restore it before clearing the intent. If no backup exists (first checkpoint), explicitly decide whether to start over with a new ID. The fallback protects the last known-good checkpoint and fails closed; it cannot promise atomic replacement on an unsupported filesystem.

Corrupt JSON, unknown/missing versions, invalid entries, an incompatible scope or an interrupted commit never become an empty baseline. Initialization fails with an actionable message and leaves the artifacts intact. Restore a verified checkpoint or use a new Watch ID intentionally. A stale .tmp file from an interrupted atomic write is ignored while the intact main checkpoint is loaded.

Directory forcing is best effort because Windows and some filesystems cannot force directory handles. Local OS locks and fsync semantics vary on network-mounted drives; use a reliable durable local state volume. Choose a durable state directory explicitly. Temporary directories may be cleaned by the OS and are unsuitable for production checkpoints. Checkpoint metadata contains filenames and URIs; protect it with filesystem permissions and use named VFS connections rather than embedding credentials in URLs.

The Maintenance tab shows three everyday actions: View status, Create backup and Reprocess files. Select Advanced recovery to reveal Clear saved history, Restore backup and Update saved-state format. The results area is hidden until an action produces a result or error. Each action has hover help. Actions resolve the state directory and Watch ID from the current form. They operate immediately, independently of OK/Cancel; cancelling the configuration dialog does not undo an explicit recovery action. Mutations require the watcher to be stopped and acquire the same OS lock as a running source. Each destructive action asks for confirmation and first archives the reserved artifacts for that Watch ID in <watch-id>.history/<timestamp>-<uuid>. Backups are retained until an operator removes them; monitor disk usage and apply your own retention policy.

Periodic diagnostics include observed and pending counts, successful scans/checkpoints, retries, native overflows, emitted/suppressed events, last successful scan time and most recent scan/checkpoint duration. Slow-operation warnings identify completed scans, stats and checkpoints; they do not cancel blocked VFS calls. Configure the provider’s connection/session timeouts as well.

Complete scans compare actual metadata with acknowledged state and recover missed create/modify/delete changes. Failed, incomplete or cancelled scans never infer deletions; the last valid state is retained and reconciliation retries at the polling interval. Root disappearance or permission failure during execution is treated as an unavailable scan, rather than mass deletion. Stability probe failures retain candidates for retry. Invalid startup roots, regexes, state directories and checkpoints fail initialization. Checkpoint write failures fail the transform. Configured file/directory/candidate limits fail safely and require a larger limit; they are not transient retries.

Native hints are capped by Native hint capacity. Excess hints request reconciliation rather than accumulating without bound. Acknowledged files and pending candidates are bounded by Maximum entries, as are visited/registered directories. Hop’s bounded rowsets provide downstream backpressure. No unbounded output-event queue or scheduler is created. A slow downstream can delay watching, scans and checkpoints; WatchService overflow then requests a scan when execution resumes. A VFS provider may materialize a full directory listing internally; configure provider connection/read timeouts and size the process for the monitored tree. Closing the WatchService wakes the native poll, and polling waits are notified on Stop. In-flight remote calls depend on the provider’s timeout support. Configure finite connection and session/read timeouts on named VFS connections. Providers such as SFTP may retry a timed-out read, so stop latency can exceed a single configured timeout.

Symbolic links are skipped and recursive native traversal does not follow links. Providers must report symbolic links accurately; visited-directory identities also prevent simple loops. Windows drive paths, file URIs and UNC paths are resolved through VFS; native paths use NIO Path normalization rather than string prefix matching. Local file URIs are normalized using UTF-8 decoding. Literal percent signs in native paths are escaped for VFS resolution, and filters use decoded base filenames. Use POLLING for network mounts with unreliable native events.

Reconciliation recovers persisting state differences, not every transient operation. A file created and removed entirely between observations cannot be reconstructed. Changes which restore the previous size and timestamp cannot be detected without a content/identity strategy. Renames are modeled as DELETED old URI and CREATED new URI. No content hashes or content reads are performed.