Add Watch Files source with persistent state and optional runtime

assemblies/debug/pom.xml

@@ -571,6 +571,12 @@
             <version>${project.version}</version>
             <scope>provided</scope>
         </dependency>
+        <dependency>
+            <groupId>org.apache.hop</groupId>
+            <artifactId>hop-transform-watchfiles</artifactId>
+            <version>${project.version}</version>
+            <scope>provided</scope>
+        </dependency>
         <dependency>
             <groupId>org.apache.hop</groupId>
             <artifactId>hop-transform-writetolog</artifactId>

assemblies/plugins/pom.xml

@@ -1596,6 +1596,12 @@
             <version>${project.version}</version>
             <type>zip</type>
         </dependency>
+        <dependency>
+            <groupId>org.apache.hop</groupId>
+            <artifactId>hop-transform-watchfiles</artifactId>
+            <version>${project.version}</version>
+            <type>zip</type>
+        </dependency>
         <dependency>
             <groupId>org.apache.hop</groupId>
             <artifactId>hop-transform-webserviceavailable</artifactId>

docs/hop-user-manual/modules/ROOT/nav.adoc

@@ -334,6 +334,7 @@ under the License.
 *** xref:pipeline/transforms/vcardinput.adoc[vCard Input]
 *** xref:pipeline/transforms/vcardoutput.adoc[vCard Output]
 *** xref:pipeline/transforms/verticabulkloader.adoc[Vertica Bulk Loader]
+*** xref:pipeline/transforms/watchfiles.adoc[Watch Files]
 *** xref:pipeline/transforms/webservices.adoc[Web services lookup]
 *** xref:pipeline/transforms/workflow-executor.adoc[Workflow Executor]
 *** xref:pipeline/transforms/workflow-logging.adoc[Workflow logging]

plugins/transforms/pom.xml

@@ -172,6 +172,7 @@
         <module>validator</module>
         <module>valuemapper</module>
         <module>vcard</module>
+        <module>watchfiles</module>
         <module>webserviceavailable</module>
         <module>webservices</module>
         <module>workflowexecutor</module>

plugins/transforms/watchfiles/TESTING.md

@@ -0,0 +1,59 @@
+<!--
+Licensed to the Apache Software Foundation (ASF) under one or more
+contributor license agreements. See the NOTICE file distributed with
+this work for additional information regarding copyright ownership.
+The ASF licenses this file to You under the Apache License, Version 2.0
+(the "License"); you may not use this file except in compliance with
+the License. You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+-->
+
+# Testing Watch Files
+
+Use Java 21 and the repository Maven wrapper:
+
+```sh
+./mvnw -B -pl plugins/transforms/watchfiles -Pskip-uitest clean install apache-rat:check spotless:check
+tools/with-isolated-display.sh ./mvnw -B -pl plugins/transforms/watchfiles -Puitest -Dwatchfiles.gui.capture=true test
+```
+
+The first command runs portable headless tests and Linux filesystem tests when
+supported. The second runs SWT GUI tests on a separate Linux display and optionally
+captures screenshots under the module's `target` directory.
+
+Coverage includes stable changes, recursive registration, overflow reconciliation,
+exclusive state ownership, corrupt or incompatible checkpoints, atomic replacement
+and interrupted fallback commits, separate-JVM termination, monotonic scheduling,
+VFS paths (including Unicode, literal percent signs, Windows drive/UNC roots and
+unusual valid Unix names), invalid native-key cleanup after capped drains,
+directory recreation, source restarts, timeout completion and downstream draining, plus GUI
+save/cancel, conditional fields and recovery actions.
+
+Environment-specific tests are opt-in:
+
+* SFTP: set `WATCHFILES_SFTP_HOST`, `WATCHFILES_SFTP_USER`,
+  `WATCHFILES_SFTP_PASSWORD`, `WATCHFILES_SFTP_ROOT`, and optionally
+  `WATCHFILES_SFTP_PORT`. Use a disposable test root; credentials belong in the
+  process environment, not checked-in configuration.
+* Disk full: set `WATCHFILES_FULL_FILESYSTEM` to a disposable, already-full local
+  filesystem for the ENOSPC test.
+* Sustained load: use `-Dtest=WatchFilesLoadTest -Dwatchfiles.load.seconds=86400`.
+  The native/polling test uses concurrent writers, bounded rowsets, a slow sink
+  and restarts. Its JSONL report defaults to `target/watchfiles-load.jsonl`.
+
+The original implementation passed a dedicated Linux soak lasting 86,444 seconds,
+with 94 cycles, 187,000 consumed events, zero duplicate and zero unexpected rows.
+Later filename-pattern, GUI and timeout changes have separate regression checks;
+this historical soak does not constitute a new 24-hour run of those changes.
+
+State records observations handed to Hop, independently of downstream completion.
+Tests do not establish end-to-end exactly-once delivery, power-loss durability,
+macOS support or coverage of every remote VFS provider. Full reactor CI remains
+the merge gate.

plugins/transforms/watchfiles/pom.xml

@@ -0,0 +1,49 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!-- SPDX-License-Identifier: Apache-2.0 -->
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <modelVersion>4.0.0</modelVersion>
+    <parent>
+        <groupId>org.apache.hop</groupId>
+        <artifactId>hop-plugins-transforms</artifactId>
+        <version>2.20.0-SNAPSHOT</version>
+    </parent>
+    <artifactId>hop-transform-watchfiles</artifactId>
+    <packaging>jar</packaging>
+    <name>Hop Plugins Transforms Watch Files</name>
+    <dependencyManagement>
+        <dependencies>
+            <dependency>
+                <groupId>org.apache.hop</groupId>
+                <artifactId>hop-libs</artifactId>
+                <version>${project.version}</version>
+                <type>pom</type>
+                <scope>import</scope>
+            </dependency>
+        </dependencies>
+    </dependencyManagement>
+    <dependencies>
+        <dependency>
+            <groupId>com.github.mwiede</groupId>
+            <artifactId>jsch</artifactId>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.hop</groupId>
+            <artifactId>hop-tech-sftp</artifactId>
+            <version>${project.version}</version>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.hop</groupId>
+            <artifactId>hop-transform-writetolog</artifactId>
+            <version>${project.version}</version>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.junit.platform</groupId>
+            <artifactId>junit-platform-launcher</artifactId>
+            <version>${junit.version}</version>
+            <scope>test</scope>
+        </dependency>
+    </dependencies>
+</project>

plugins/transforms/watchfiles/src/assembly/assembly.xml

@@ -0,0 +1,46 @@
+<!--
+  ~ Licensed to the Apache Software Foundation (ASF) under one or more
+  ~ contributor license agreements.  See the NOTICE file distributed with
+  ~ this work for additional information regarding copyright ownership.
+  ~ The ASF licenses this file to You under the Apache License, Version 2.0
+  ~ (the "License"); you may not use this file except in compliance with
+  ~ the License.  You may obtain a copy of the License at
+  ~
+  ~       http://www.apache.org/licenses/LICENSE-2.0
+  ~
+  ~ Unless required by applicable law or agreed to in writing, software
+  ~ distributed under the License is distributed on an "AS IS" BASIS,
+  ~ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+  ~ See the License for the specific language governing permissions and
+  ~ limitations under the License.
+  ~
+  -->
+
+<assembly xmlns="http://maven.apache.org/ASSEMBLY/2.2.0"
+    xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+    xsi:schemaLocation="http://maven.apache.org/ASSEMBLY/2.2.0 http://maven.apache.org/xsd/assembly-2.2.0.xsd">
+    <id>hop-transform-watchfiles</id>
+    <formats>
+        <format>zip</format>
+    </formats>
+    <baseDirectory>.</baseDirectory>
+    <files>
+        <file>
+            <source>${project.basedir}/src/main/resources/version.xml</source>
+            <outputDirectory>${hop.plugin.libdir}</outputDirectory>
+            <filtered>true</filtered>
+        </file>
+    </files>
+
+    <fileSets>
+        <fileSet>
+            <directory>${project.basedir}/src/main/samples</directory>
+            <outputDirectory>config/projects/samples/</outputDirectory>
+        </fileSet>
+    </fileSets>
+
+
+    <componentDescriptors>
+        <componentDescriptor>${maven.multiModuleProjectDirectory}/assemblies/shared/hop-plugin-libs.xml</componentDescriptor>
+    </componentDescriptors>
+</assembly>

plugins/transforms/watchfiles/src/main/java/org/apache/hop/pipeline/transforms/watchfiles/DetectionStrategy.java

@@ -0,0 +1,24 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hop.pipeline.transforms.watchfiles;
+
+public enum DetectionStrategy {
+  AUTO,
+  NATIVE,
+  POLLING
+}

plugins/transforms/watchfiles/src/main/java/org/apache/hop/pipeline/transforms/watchfiles/FileChangeEvent.java

@@ -0,0 +1,32 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hop.pipeline.transforms.watchfiles;
+
+import lombok.Value;
+
+@Value
+public class FileChangeEvent {
+  FileEventType type;
+  FileState current;
+  FileState previous;
+  long detectedAt;
+
+  public FileState file() {
+    return current == null ? previous : current;
+  }
+}

plugins/transforms/watchfiles/src/main/java/org/apache/hop/pipeline/transforms/watchfiles/FileEventType.java

@@ -0,0 +1,24 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hop.pipeline.transforms.watchfiles;
+
+public enum FileEventType {
+  CREATED,
+  MODIFIED,
+  DELETED
+}

plugins/transforms/watchfiles/src/main/java/org/apache/hop/pipeline/transforms/watchfiles/FileReconciler.java

@@ -0,0 +1,44 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hop.pipeline.transforms.watchfiles;
+
+import java.util.Map;
+import java.util.function.Consumer;
+
+/** Only complete successful snapshots may be passed here: a failed listing is not a deletion. */
+public class FileReconciler {
+  public void compare(
+      Map<String, FileState> observed,
+      Map<String, FileState> current,
+      long now,
+      Consumer<FileChangeEvent> changes) {
+    for (FileState file : current.values()) {
+      FileState previous = observed.get(file.getUri());
+      if (previous == null) {
+        changes.accept(new FileChangeEvent(FileEventType.CREATED, file, null, now));
+      } else if (!file.sameVersion(previous)) {
+        changes.accept(new FileChangeEvent(FileEventType.MODIFIED, file, previous, now));
+      }
+    }
+    for (FileState file : observed.values()) {
+      if (!current.containsKey(file.getUri())) {
+        changes.accept(new FileChangeEvent(FileEventType.DELETED, null, file, now));
+      }
+    }
+  }
+}