This is an automated email from the ASF dual-hosted git repository.
mattcasters pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/hop.git
The following commit(s) were added to refs/heads/main by this push:
new a411bcad48 Issue #8952 : Fix the mail, lineage, OpenSearch, and
samples integration failures (#8953)
a411bcad48 is described below
commit a411bcad48f522592fce4d82e9f730f211142817
Author: Matt Casters <[email protected]>
AuthorDate: Sat Oct 17 00:55:46 2026 +0200
Issue #8952 : Fix the mail, lineage, OpenSearch, and samples integration
failures (#8953)
* Issue #8952 : Fix the mail, lineage, OpenSearch, and samples integration
failures
James WebAdmin was still starting when the mail tests created users, so the
fixed 30s delay curled a closed port and the later mail tests had no
accounts.
The tests now wait until port 8000 accepts connections.
Registering a child execution linked both the owner and the launcher with
EXECUTES. shortestPath then skipped the action in the error-lineage path.
Link the launcher when its node exists, and move an earlier owner shortcut
onto the launcher when that node is stored later.
An OpenSearch child location loaded a parent without the document _id and
posted a second document, then could overwrite a finished state. Keep the
document id, re-read before that link write, and refresh so the new state is
visible.
The samples suite was still moving when the 3600s watchdog killed it during
the huge-file sample. That project now has a 5400s cap. The other projects
keep 3600s.
* Issue #8952 : Keep a finished OpenSearch state when an attach-only
location writes
A second location that loads a workflow only to attach actions kept that
snapshot and wrote it back on close, replacing Finished with a stale copy.
Re-read before that write and keep the newer state, including children from
either copy. Drop refresh=wait_for; the re-read is a realtime get by
document id.
Co-authored-by: Matt Casters <[email protected]>
---------
Co-authored-by: Cursor Agent <[email protected]>
---
.../integration-tests-samples.yaml | 27 +++
.../caching/BaseCachingExecutionInfoLocation.java | 157 +++++++++++++++-
.../apache/hop/execution/caching/CacheEntry.java | 6 +
.../BaseCachingExecutionInfoLocationTest.java | 205 ++++++++++++++++++++-
integration-tests/mail/main-0001-james-rest.hwf | 37 ++--
integration-tests/mail/main-0009-mail-with-zip.hwf | 16 +-
.../neo4j/execution/NeoExecutionInfoLocation.java | 75 +++++++-
.../NeoExecutionInfoLocationDeleteTest.java | 20 ++
.../OpenSearchExecutionInfoLocation.java | 105 +++++++++--
...OpenSearchExecutionInfoLocationCleanupTest.java | 45 +++++
10 files changed, 646 insertions(+), 47 deletions(-)
diff --git a/docker/integration-tests/integration-tests-samples.yaml
b/docker/integration-tests/integration-tests-samples.yaml
new file mode 100644
index 0000000000..b6c9be40f7
--- /dev/null
+++ b/docker/integration-tests/integration-tests-samples.yaml
@@ -0,0 +1,27 @@
+# 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.
+
+# The samples suite runs every assembled sample in one JVM. A literal timeout
is required:
+# run-tests-docker.sh exports HOP_IT_TIMEOUT before Compose interpolates the
base file, so
+# ${HOP_IT_TIMEOUT:-3600} there cannot raise the cap for this project alone.
+services:
+ integration_test_samples:
+ extends:
+ file: integration-tests-base.yaml
+ service: integration_test
+ environment:
+ HOP_IT_TIMEOUT: "5400"
diff --git
a/engine/src/main/java/org/apache/hop/execution/caching/BaseCachingExecutionInfoLocation.java
b/engine/src/main/java/org/apache/hop/execution/caching/BaseCachingExecutionInfoLocation.java
index 6db70834f4..06b8c94f8c 100644
---
a/engine/src/main/java/org/apache/hop/execution/caching/BaseCachingExecutionInfoLocation.java
+++
b/engine/src/main/java/org/apache/hop/execution/caching/BaseCachingExecutionInfoLocation.java
@@ -239,7 +239,7 @@ public abstract class BaseCachingExecutionInfoLocation
implements IExecutionInfo
for (CacheEntry cacheEntry : cache.values()) {
if (cacheEntry.needsWriting(delay)) {
try {
- persistCacheEntry(cacheEntry);
+ saveCacheEntry(cacheEntry);
} catch (Exception e) {
LogChannel.GENERAL.logError(
"Error persisting cache entry for " + cacheEntry.getId(), e);
@@ -287,7 +287,7 @@ public abstract class BaseCachingExecutionInfoLocation
implements IExecutionInfo
CacheEntry cacheEntry = entry.getValue();
if (cacheEntry != null && cacheEntry.isDirty()) {
try {
- persistCacheEntry(cacheEntry);
+ saveCacheEntry(cacheEntry);
} catch (Exception e) {
// A failed write must stay in memory. Dropping it here loses the
only copy.
LogChannel.GENERAL.logError(
@@ -312,7 +312,7 @@ public abstract class BaseCachingExecutionInfoLocation
implements IExecutionInfo
CacheEntry cacheEntry = mapEntry.getValue();
if (cacheEntry != null && cacheEntry.isDirty()) {
try {
- persistCacheEntry(cacheEntry);
+ saveCacheEntry(cacheEntry);
} catch (Exception e) {
// Leave this entry. A later close() can retry it. Clearing it drops
unsaved state.
if (failure == null) {
@@ -348,7 +348,7 @@ public abstract class BaseCachingExecutionInfoLocation
implements IExecutionInfo
// CacheEntry from disk before calling registerData / register child
executions.
CacheEntry entry = cache.get(execution.getId());
if (entry != null) {
- persistCacheEntry(entry);
+ saveCacheEntry(entry);
}
rememberLaunchedChild(execution);
} else {
@@ -392,7 +392,15 @@ public abstract class BaseCachingExecutionInfoLocation
implements IExecutionInfo
return;
}
parent.setDirty(true);
- persistCacheEntry(parent);
+ // This location loaded the parent only to attach the link. The parent
may have finished and
+ // written a newer state after that load. Writing the stale copy would
drop logging text and
+ // the end date. Re-read and keep the newer state, then put the link
back on it.
+ if (loadedForLink) {
+ parent = keepNewerParentState(parent, child);
+ // The link re-read just ran. This write is the merged document, then
the entry is dropped.
+ parent.setLoadedCopy(false);
+ }
+ saveCacheEntry(parent);
parent.setDirty(false);
} catch (Exception e) {
LogChannel.GENERAL.logError(
@@ -409,6 +417,139 @@ public abstract class BaseCachingExecutionInfoLocation
implements IExecutionInfo
}
}
+ /**
+ * Write one cache entry. A document this location only loaded is merged
with a fresh read first,
+ * so attaching a child cannot replace a finished execution state on close.
+ */
+ private void saveCacheEntry(CacheEntry cacheEntry) throws HopException {
+ if (cacheEntry == null) {
+ return;
+ }
+ mergeLoadedCopyBeforePersist(cacheEntry);
+ persistCacheEntry(cacheEntry);
+ }
+
+ /**
+ * A location that loaded a document to attach children keeps that snapshot
until it closes. The
+ * execution that owns the document may have finished after the load.
Re-read and keep that state,
+ * and keep the children recorded by either copy.
+ */
+ private void mergeLoadedCopyBeforePersist(CacheEntry loaded) throws
HopException {
+ if (loaded == null || !loaded.isLoadedCopy()) {
+ return;
+ }
+ CacheEntry fresh = loadFreshCacheEntry(loaded);
+ if (fresh == null) {
+ return;
+ }
+ if (StringUtils.isEmpty(loaded.getInternalId())
+ && StringUtils.isNotEmpty(fresh.getInternalId())) {
+ loaded.setInternalId(fresh.getInternalId());
+ }
+ loaded.keepStoredProjectId(fresh);
+ copyMissingChildren(fresh, loaded);
+ if (isNewerParentState(fresh, loaded)) {
+ loaded.setExecutionState(fresh.peekExecutionState());
+ }
+ }
+
+ /**
+ * Read the stored document again before a loaded copy is written. A
location can override this
+ * when a direct read sees a newer write than {@link
#loadCacheEntry(String)}.
+ */
+ protected CacheEntry loadFreshCacheEntry(CacheEntry loaded) throws
HopException {
+ if (loaded == null || StringUtils.isEmpty(loaded.getId())) {
+ return null;
+ }
+ return loadCacheEntry(loaded.getId());
+ }
+
+ /** Children stored by the other writer stay on the document this location
is about to save. */
+ private static void copyMissingChildren(CacheEntry from, CacheEntry into) {
+ if (from == null || into == null || from == into) {
+ return;
+ }
+ copyMissingExecutions(from.getChildExecutions(),
into.getChildExecutions());
+ copyMissing(from.getChildExecutionStates(),
into.getChildExecutionStates());
+ copyMissing(from.getChildExecutionData(), into.getChildExecutionData());
+ }
+
+ private static void copyMissingExecutions(
+ Map<String, Execution> from, Map<String, Execution> into) {
+ if (from == null || from.isEmpty() || into == null) {
+ return;
+ }
+ for (Map.Entry<String, Execution> child : from.entrySet()) {
+ Execution stored = child.getValue();
+ Execution kept = into.putIfAbsent(child.getKey(), stored);
+ if (kept != null && stored != null) {
+ kept.unionChildExecutionIds(stored);
+ }
+ }
+ }
+
+ private static <K, V> void copyMissing(Map<K, V> from, Map<K, V> into) {
+ if (from == null || from.isEmpty() || into == null) {
+ return;
+ }
+ for (Map.Entry<K, V> child : from.entrySet()) {
+ into.putIfAbsent(child.getKey(), child.getValue());
+ }
+ }
+
+ /**
+ * Re-read a parent that was loaded only to record a child. When the stored
copy is newer, move
+ * the launcher and the child id onto that copy so the write does not
replace a finished state.
+ */
+ CacheEntry keepNewerParentState(CacheEntry loaded, Execution child) throws
HopException {
+ CacheEntry fresh = loadFreshCacheEntry(loaded);
+ if (fresh == null || fresh.getExecution() == null ||
!isNewerParentState(fresh, loaded)) {
+ if (fresh != null
+ && StringUtils.isEmpty(loaded.getInternalId())
+ && StringUtils.isNotEmpty(fresh.getInternalId())) {
+ loaded.setInternalId(fresh.getInternalId());
+ }
+ return loaded;
+ }
+ Execution launcher = fresh.peekChildExecution(child.getLaunchedById());
+ if (launcher == null) {
+ launcher = minimalLauncher(child, fresh.getExecution());
+ fresh.addChildExecution(launcher);
+ }
+ launcher.addChildExecutionId(child.getId());
+ fresh.setDirty(true);
+ return fresh;
+ }
+
+ /**
+ * A stored parent is newer when it has finished and the loaded copy has
not, or when its update
+ * time is later. A finished copy is never replaced by an unfinished one. A
child location uses
+ * this so it does not write an older execution state back.
+ */
+ static boolean isNewerParentState(CacheEntry fresh, CacheEntry loaded) {
+ ExecutionState freshState = fresh == null ? null :
fresh.peekExecutionState();
+ ExecutionState loadedState = loaded == null ? null :
loaded.peekExecutionState();
+ if (freshState == null) {
+ return false;
+ }
+ if (loadedState == null) {
+ return true;
+ }
+ if (freshState.getExecutionEndDate() != null &&
loadedState.getExecutionEndDate() == null) {
+ return true;
+ }
+ // An unfinished snapshot must not replace a finished one, even when its
update time is later.
+ if (loadedState.getExecutionEndDate() != null &&
freshState.getExecutionEndDate() == null) {
+ return false;
+ }
+ Date freshUpdate = freshState.getUpdateTime();
+ Date loadedUpdate = loadedState.getUpdateTime();
+ if (freshUpdate == null) {
+ return false;
+ }
+ return loadedUpdate == null || freshUpdate.after(loadedUpdate);
+ }
+
/**
* An action or transform that has not been flushed yet. The id and the
child link are enough for
* drill-down. The parent's next save replaces this with the real launcher
and keeps the child id.
@@ -509,6 +650,7 @@ public abstract class BaseCachingExecutionInfoLocation
implements IExecutionInfo
entry.setId(execution.getId());
}
entry.setExecution(execution);
+ entry.setLoadedCopy(false);
entry.setName(execution.getName());
if (StringUtils.isNotEmpty(execution.getProjectId())) {
entry.setProjectId(execution.getProjectId());
@@ -718,6 +860,9 @@ public abstract class BaseCachingExecutionInfoLocation
implements IExecutionInfo
entry.setLastRead(new Date());
entry.setLastWritten(new Date());
entry.setDirty(false);
+ // Registered executions are created in this cache. A loaded document
belongs to another
+ // location until this one registers it, so a later save has to re-read
the state.
+ entry.setLoadedCopy(true);
cache.put(entry.getId(), entry);
enforceMaxCacheSize();
@@ -769,7 +914,7 @@ public abstract class BaseCachingExecutionInfoLocation
implements IExecutionInfo
// A local single-writer flushes from the cache timer. Another process
(Spark/Beam) still
// needs the samples on disk before it merges its own write.
if (!entry.isSingleWriter()) {
- persistCacheEntry(entry);
+ saveCacheEntry(entry);
}
} else {
LogChannel.GENERAL.logError(
diff --git
a/engine/src/main/java/org/apache/hop/execution/caching/CacheEntry.java
b/engine/src/main/java/org/apache/hop/execution/caching/CacheEntry.java
index eae0067049..0915636233 100644
--- a/engine/src/main/java/org/apache/hop/execution/caching/CacheEntry.java
+++ b/engine/src/main/java/org/apache/hop/execution/caching/CacheEntry.java
@@ -98,6 +98,12 @@ public class CacheEntry {
*/
@JsonIgnore private boolean singleWriter;
+ /**
+ * Set when a location loads this document from storage. A save re-reads so
the loaded execution
+ * state does not replace a newer stored state.
+ */
+ @JsonIgnore private boolean loadedCopy;
+
/** The document with metadata and pipeline XML has been inserted. */
@JsonIgnore private boolean heavyDocumentStored;
diff --git
a/engine/src/test/java/org/apache/hop/execution/caching/BaseCachingExecutionInfoLocationTest.java
b/engine/src/test/java/org/apache/hop/execution/caching/BaseCachingExecutionInfoLocationTest.java
index 6bbba58d8d..0f2f83d906 100644
---
a/engine/src/test/java/org/apache/hop/execution/caching/BaseCachingExecutionInfoLocationTest.java
+++
b/engine/src/test/java/org/apache/hop/execution/caching/BaseCachingExecutionInfoLocationTest.java
@@ -18,13 +18,17 @@
package org.apache.hop.execution.caching;
import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertNotNull;
import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertSame;
import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;
+import java.util.ArrayList;
import java.util.Date;
import java.util.HashSet;
+import java.util.List;
import java.util.Set;
import java.util.UUID;
import org.apache.hop.core.exception.HopException;
@@ -207,6 +211,160 @@ class BaseCachingExecutionInfoLocationTest {
assertTrue(wfChildren.stream().anyMatch(e -> e.getId().equals(actionId)));
}
+ @Test
+ void childLinkKeepsANewerStoredParentState() throws Exception {
+ FakeLocation location = new FakeLocation();
+ CacheEntry running = parentEntry("wf", "running", new Date(1_000L), null);
+ CacheEntry finished = parentEntry("wf", "Finished", new Date(5_000L), new
Date(5_000L));
+ finished.setInternalId("doc-9");
+ location.scriptedLoads.add(running);
+ location.scriptedLoads.add(finished);
+
+ location.registerExecution(launchedPipeline("pipe", "wf", "action"));
+
+ assertNull(location.getCache().get("wf"));
+ assertSame(finished, location.lastPersisted);
+ assertEquals("Finished",
location.lastPersisted.getExecutionState().getStatusDescription());
+
assertNotNull(location.lastPersisted.getExecutionState().getExecutionEndDate());
+ assertEquals("doc-9", location.lastPersisted.getInternalId());
+ Execution launcher = location.lastPersisted.peekChildExecution("action");
+ assertNotNull(launcher);
+ assertEquals(List.of("pipe"), launcher.getChildExecutionIds());
+ }
+
+ @Test
+ void childLinkKeepsTheLoadedParentWhenTheStoredCopyIsNotNewer() throws
Exception {
+ FakeLocation location = new FakeLocation();
+ CacheEntry loaded = parentEntry("wf", "Finished", new Date(5_000L), new
Date(5_000L));
+ CacheEntry stored = parentEntry("wf", "Finished", new Date(5_000L), new
Date(5_000L));
+ stored.setInternalId("doc-1");
+ location.scriptedLoads.add(loaded);
+ location.scriptedLoads.add(stored);
+
+ location.registerExecution(launchedPipeline("pipe", "wf", "action"));
+
+ assertSame(loaded, location.lastPersisted);
+ assertEquals("doc-1", location.lastPersisted.getInternalId());
+ assertEquals(
+ List.of("pipe"),
+
location.lastPersisted.peekChildExecution("action").getChildExecutionIds());
+ }
+
+ @Test
+ void childLinkDoesNotReplaceAParentAlreadyInThisLocation() throws Exception {
+ FakeLocation location = new FakeLocation();
+ Execution workflow = pipeline("wf", "Workflow");
+ workflow.setExecutionType(ExecutionType.Workflow);
+ location.registerExecution(workflow);
+ ExecutionState running = new ExecutionState();
+ running.setId("wf");
+ running.setExecutionType(ExecutionType.Workflow);
+ running.setStatusDescription("running");
+ running.setUpdateTime(new Date(1_000L));
+ location.updateExecutionState(running);
+
+ CacheEntry finished = parentEntry("wf", "Finished", new Date(9_000L), new
Date(9_000L));
+ location.scriptedLoads.add(finished);
+ int loadsBeforeChild = location.loadCount;
+
+ location.registerExecution(launchedPipeline("pipe", "wf", "action"));
+
+ assertEquals(loadsBeforeChild, location.loadCount);
+ assertEquals("running",
location.getExecutionState("wf").getStatusDescription());
+ assertEquals(List.of("pipe"),
location.getExecution("action").getChildExecutionIds());
+ }
+
+ @Test
+ void attachOnlyLocationKeepsAFinishedStateWhenItLoadedARunningCopy() throws
Exception {
+ FakeLocation attach = new FakeLocation();
+ CacheEntry running = parentEntry("wf", "running", new Date(9_000L), null);
+ running.setInternalId("doc-1");
+ CacheEntry finished = parentEntry("wf", "Finished", new Date(1_000L), new
Date(5_000L));
+ finished.setInternalId("doc-1");
+ finished.getExecutionState().setLoggingText("done");
+ finished.addChildExecution(action("kept-action"));
+ attach.scriptedLoads.add(running);
+ attach.scriptedLoads.add(finished);
+
+ attach.registerExecution(action("action"));
+ assertTrue(attach.getCache().get("wf").isLoadedCopy());
+ assertEquals(1, attach.loadCount);
+
+ attach.close();
+
+ assertEquals(2, attach.loadCount);
+ assertEquals("Finished",
attach.lastPersisted.getExecutionState().getStatusDescription());
+
assertNotNull(attach.lastPersisted.getExecutionState().getExecutionEndDate());
+ assertEquals("done",
attach.lastPersisted.getExecutionState().getLoggingText());
+ assertEquals("doc-1", attach.lastPersisted.getInternalId());
+ assertNotNull(attach.lastPersisted.peekChildExecution("action"));
+ assertNotNull(attach.lastPersisted.peekChildExecution("kept-action"));
+ }
+
+ @Test
+ void attachOnlyLocationKeepsAFinishedStateWhenItLoadedBeforeAnyState()
throws Exception {
+ FakeLocation attach = new FakeLocation();
+ CacheEntry loaded = new CacheEntry();
+ loaded.setId("wf");
+ loaded.setName("wf");
+ Execution workflow = new Execution();
+ workflow.setId("wf");
+ workflow.setName("wf");
+ workflow.setExecutionType(ExecutionType.Workflow);
+ workflow.setRegistrationDate(new Date(1_000L));
+ loaded.setExecution(workflow);
+ loaded.setInternalId("doc-2");
+ CacheEntry finished = parentEntry("wf", "Finished", new Date(5_000L), new
Date(5_000L));
+ finished.setInternalId("doc-2");
+ attach.scriptedLoads.add(loaded);
+ attach.scriptedLoads.add(finished);
+
+ attach.registerExecution(action("action"));
+ attach.close();
+
+ assertEquals("Finished",
attach.lastPersisted.getExecutionState().getStatusDescription());
+
assertNotNull(attach.lastPersisted.getExecutionState().getExecutionEndDate());
+ assertNotNull(attach.lastPersisted.peekChildExecution("action"));
+ }
+
+ @Test
+ void registeringLocationWritesItsOwnFinishedStateWithoutReloading() throws
Exception {
+ FakeLocation location = new FakeLocation();
+ Execution workflow = pipeline("wf", "Workflow");
+ workflow.setExecutionType(ExecutionType.Workflow);
+ location.registerExecution(workflow);
+ ExecutionState finished = new ExecutionState();
+ finished.setId("wf");
+ finished.setExecutionType(ExecutionType.Workflow);
+ finished.setStatusDescription("Finished");
+ finished.setUpdateTime(new Date(5_000L));
+ finished.setExecutionEndDate(new Date(5_000L));
+ location.updateExecutionState(finished);
+
+ location.close();
+
+ assertEquals(0, location.loadCount);
+ assertEquals("Finished",
location.lastPersisted.getExecutionState().getStatusDescription());
+
assertNotNull(location.lastPersisted.getExecutionState().getExecutionEndDate());
+ }
+
+ @Test
+ void aFinishedParentStateIsNewerThanARunningOne() {
+ CacheEntry running = parentEntry("wf", "running", new Date(9_000L), null);
+ CacheEntry finished = parentEntry("wf", "Finished", new Date(1_000L), new
Date(1_000L));
+
+ assertTrue(BaseCachingExecutionInfoLocation.isNewerParentState(finished,
running));
+ assertFalse(BaseCachingExecutionInfoLocation.isNewerParentState(running,
finished));
+ assertTrue(
+ BaseCachingExecutionInfoLocation.isNewerParentState(
+ parentEntry("wf", "running", new Date(9_000L), null),
+ parentEntry("wf", "running", new Date(1_000L), null)));
+ assertFalse(
+ BaseCachingExecutionInfoLocation.isNewerParentState(
+ parentEntry("wf", "running", new Date(1_000L), null),
+ parentEntry("wf", "running", new Date(1_000L), null)));
+ }
+
@Test
void testAddExecutionDataRetainsNonEmptySamples() {
CacheEntry entry = new CacheEntry();
@@ -244,6 +402,42 @@ class BaseCachingExecutionInfoLocationTest {
assertEquals(1,
entry.getExecutionData("all-transforms").getDataSets().get("set1").size());
}
+ private static Execution action(String id) {
+ Execution execution = new Execution();
+ execution.setId(id);
+ execution.setName(id);
+ execution.setParentId("wf");
+ execution.setExecutionType(ExecutionType.Action);
+ execution.setRegistrationDate(new Date(1_000L));
+ return execution;
+ }
+
+ private static Execution launchedPipeline(String id, String parentId, String
launchedById) {
+ Execution execution = pipeline(id, id);
+ execution.setParentId(parentId);
+ execution.setLaunchedById(launchedById);
+ return execution;
+ }
+
+ private static CacheEntry parentEntry(String id, String status, Date
updateTime, Date endDate) {
+ Execution execution = new Execution();
+ execution.setId(id);
+ execution.setName(id);
+ execution.setExecutionType(ExecutionType.Workflow);
+ execution.setRegistrationDate(new Date(1_000L));
+ CacheEntry entry = new CacheEntry();
+ entry.setId(id);
+ entry.setExecution(execution);
+ ExecutionState state = new ExecutionState();
+ state.setId(id);
+ state.setExecutionType(ExecutionType.Workflow);
+ state.setStatusDescription(status);
+ state.setUpdateTime(updateTime);
+ state.setExecutionEndDate(endDate);
+ entry.setExecutionState(state);
+ return entry;
+ }
+
private static Execution pipeline(String id, String name) {
Execution execution = new Execution();
execution.setId(id);
@@ -266,6 +460,10 @@ class BaseCachingExecutionInfoLocationTest {
private static final class FakeLocation extends
BaseCachingExecutionInfoLocation {
private boolean failPersist;
private final Set<String> persisted = new HashSet<>();
+ private final List<CacheEntry> scriptedLoads = new ArrayList<>();
+ private int scriptedLoadIndex;
+ private CacheEntry lastPersisted;
+ private int loadCount;
private String pluginId;
private String pluginName;
@@ -300,13 +498,18 @@ class BaseCachingExecutionInfoLocationTest {
throw new HopException("persist failed for " + cacheEntry.getId());
}
persisted.add(cacheEntry.getId());
+ lastPersisted = cacheEntry;
cacheEntry.setDirty(false);
cacheEntry.setLastWritten(new Date());
}
@Override
protected CacheEntry loadCacheEntry(String executionId) {
- return null;
+ loadCount++;
+ if (scriptedLoadIndex >= scriptedLoads.size()) {
+ return null;
+ }
+ return scriptedLoads.get(scriptedLoadIndex++);
}
@Override
diff --git a/integration-tests/mail/main-0001-james-rest.hwf
b/integration-tests/mail/main-0001-james-rest.hwf
index 15dc86dea2..06c582d847 100644
--- a/integration-tests/mail/main-0001-james-rest.hwf
+++ b/integration-tests/mail/main-0001-james-rest.hwf
@@ -63,7 +63,21 @@ limitations under the License.
<loglevel>Basic</loglevel>
<script>#!/bin/bash
-echo "james server: " ${MAIL_SMTP_SERVER}
+echo "james server: ${MAIL_SMTP_SERVER}"
+ready=0
+attempt=0
+while [ "$attempt" -lt 60 ]; do
+ attempt=$((attempt + 1))
+ if curl -sf -o /dev/null --max-time 2
"http://${MAIL_SMTP_SERVER}:8000/domains"; 2>/dev/null; then
+ ready=1
+ break
+ fi
+ sleep 2
+done
+if [ "$ready" != 1 ]; then
+ echo "James WebAdmin did not accept connections on port 8000"
+ exit 1
+fi
curl -XPUT http://${MAIL_SMTP_SERVER}:8000/domains/example.com
curl -XPUT http://${MAIL_SMTP_SERVER}:8000/domains/acme.be
@@ -114,18 +128,6 @@ curl -XPUT
http://${MAIL_SMTP_SERVER}:8000/users/[email protected] -d '{"pass
<yloc>64</yloc>
<attributes_hac/>
</action>
- <action>
- <name>wait for 30s</name>
- <description/>
- <type>DELAY</type>
- <attributes/>
- <maximumTimeout>30</maximumTimeout>
- <scaletime>0</scaletime>
- <parallel>N</parallel>
- <xloc>224</xloc>
- <yloc>64</yloc>
- <attributes_hac/>
- </action>
<action>
<name>get mails</name>
<description/>
@@ -232,17 +234,10 @@ curl -XPUT
http://${MAIL_SMTP_SERVER}:8000/users/[email protected] -d '{"pass
</hop>
<hop>
<from>Start</from>
- <to>wait for 30s</to>
- <enabled>Y</enabled>
- <evaluation>Y</evaluation>
- <unconditional>Y</unconditional>
- </hop>
- <hop>
- <from>wait for 30s</from>
<to>add domains, users</to>
<enabled>Y</enabled>
<evaluation>Y</evaluation>
- <unconditional>N</unconditional>
+ <unconditional>Y</unconditional>
</hop>
<hop>
<from>get mails</from>
diff --git a/integration-tests/mail/main-0009-mail-with-zip.hwf
b/integration-tests/mail/main-0009-mail-with-zip.hwf
index ed51945922..27eb98a477 100644
--- a/integration-tests/mail/main-0009-mail-with-zip.hwf
+++ b/integration-tests/mail/main-0009-mail-with-zip.hwf
@@ -139,7 +139,21 @@ limitations under the License.
<loglevel>Basic</loglevel>
<script>#!/bin/bash
-echo "james server: " ${MAIL_SMTP_SERVER}
+echo "james server: ${MAIL_SMTP_SERVER}"
+ready=0
+attempt=0
+while [ "$attempt" -lt 60 ]; do
+ attempt=$((attempt + 1))
+ if curl -sf -o /dev/null --max-time 2
"http://${MAIL_SMTP_SERVER}:8000/domains"; 2>/dev/null; then
+ ready=1
+ break
+ fi
+ sleep 2
+done
+if [ "$ready" != 1 ]; then
+ echo "James WebAdmin did not accept connections on port 8000"
+ exit 1
+fi
curl -XPUT http://${MAIL_SMTP_SERVER}:8000/domains/example.com
curl -XPUT http://${MAIL_SMTP_SERVER}:8000/domains/acme.be
diff --git
a/plugins/tech/neo4j/src/main/java/org/apache/hop/neo4j/execution/NeoExecutionInfoLocation.java
b/plugins/tech/neo4j/src/main/java/org/apache/hop/neo4j/execution/NeoExecutionInfoLocation.java
index 5b87e0d4d9..abf952a2d0 100644
---
a/plugins/tech/neo4j/src/main/java/org/apache/hop/neo4j/execution/NeoExecutionInfoLocation.java
+++
b/plugins/tech/neo4j/src/main/java/org/apache/hop/neo4j/execution/NeoExecutionInfoLocation.java
@@ -500,16 +500,24 @@ public class NeoExecutionInfoLocation implements
IExecutionInfoLocation {
.withValue(EP_EXECUTION_START_DATE,
execution.getExecutionStartDate());
execute(transaction, builder);
- if (StringUtils.isNotEmpty(execution.getParentId())) {
+ // The action or transform that started this execution. Link that node
when it is already
+ // stored, and do not also link the owner: shortestPath would skip the
launcher. A missing
+ // launcher keeps the owner edge so the child is not orphaned.
+ boolean launcherStored =
+ StringUtils.isNotEmpty(execution.getLaunchedById())
+ && !execution.getLaunchedById().equals(execution.getParentId())
+ && executionNodeExists(transaction, execution.getLaunchedById());
+ if (linkOwnerExecutes(execution.getParentId(),
execution.getLaunchedById(), launcherStored)) {
linkExecutes(transaction, execution.getParentId(), execution.getId());
}
- // The action or transform that started this execution. Link it when
that node is already
- // stored. A missing launcher must not roll back the execution itself.
- if (StringUtils.isNotEmpty(execution.getLaunchedById())
- && !execution.getLaunchedById().equals(execution.getParentId())
- && executionNodeExists(transaction, execution.getLaunchedById())) {
+ if (launcherStored) {
linkExecutes(transaction, execution.getLaunchedById(),
execution.getId());
}
+ // A launcher stored after its child still has only the owner shortcut.
Move EXECUTES onto
+ // the launcher now that its node exists.
+ if (StringUtils.isNotEmpty(execution.getId())) {
+ run(transaction, relinkLaunchedChildrenCypher(), Map.of("id",
execution.getId()));
+ }
// See if we need to cache this
//
@@ -523,6 +531,61 @@ public class NeoExecutionInfoLocation implements
IExecutionInfoLocation {
}
}
+ /**
+ * Link {@code (parent)-[:EXECUTES]->(child)} only when the launcher was not
linked. The owner
+ * shortcut and the launcher edge together make {@code shortestPath} skip
the action or transform.
+ */
+ static boolean linkOwnerExecutes(
+ String parentId, String launchedById, boolean launcherNodeExists) {
+ if (StringUtils.isEmpty(parentId)) {
+ return false;
+ }
+ if (StringUtils.isEmpty(launchedById) || launchedById.equals(parentId)) {
+ return true;
+ }
+ return !launcherNodeExists;
+ }
+
+ /**
+ * Children already stored with {@code launchedById} equal to this execution
get an EXECUTES edge
+ * from this node, and the owner shortcut is removed. Actions have no {@code
launchedById}, so
+ * {@code (workflow)-[:EXECUTES]->(action)} stays.
+ */
+ static String relinkLaunchedChildrenCypher() {
+ return "MATCH (launcher:"
+ + EL_EXECUTION
+ + " {"
+ + EP_ID
+ + ": $id}) "
+ + "MATCH (child:"
+ + EL_EXECUTION
+ + ") "
+ + "WHERE child."
+ + EP_LAUNCHED_BY_ID
+ + " = $id "
+ + "AND child."
+ + EP_PARENT_ID
+ + " IS NOT NULL "
+ + "AND child."
+ + EP_PARENT_ID
+ + " <> $id "
+ + "MERGE (launcher)-[:"
+ + R_EXECUTES
+ + "]->(child) "
+ + "WITH child "
+ + "MATCH (owner:"
+ + EL_EXECUTION
+ + ")-[shortcut:"
+ + R_EXECUTES
+ + "]->(child) "
+ + "WHERE owner."
+ + EP_ID
+ + " = child."
+ + EP_PARENT_ID
+ + " "
+ + "DELETE shortcut";
+ }
+
/** {@code (from)-[:EXECUTES]->(to)} when both nodes are already stored. */
private void linkExecutes(IGraphTransaction transaction, String fromId,
String toId) {
execute(
diff --git
a/plugins/tech/neo4j/src/test/java/org/apache/hop/neo4j/execution/NeoExecutionInfoLocationDeleteTest.java
b/plugins/tech/neo4j/src/test/java/org/apache/hop/neo4j/execution/NeoExecutionInfoLocationDeleteTest.java
index 2be8c7f0db..8de8d31092 100644
---
a/plugins/tech/neo4j/src/test/java/org/apache/hop/neo4j/execution/NeoExecutionInfoLocationDeleteTest.java
+++
b/plugins/tech/neo4j/src/test/java/org/apache/hop/neo4j/execution/NeoExecutionInfoLocationDeleteTest.java
@@ -58,6 +58,26 @@ class NeoExecutionInfoLocationDeleteTest {
assertEquals("action-1", query.parameters().get("executionId"));
}
+ @Test
+ void ownerEdgeIsSkippedWhenTheLauncherNodeIsLinked() {
+ assertFalse(NeoExecutionInfoLocation.linkOwnerExecutes("workflow",
"action", true));
+ assertTrue(NeoExecutionInfoLocation.linkOwnerExecutes("workflow",
"action", false));
+ assertTrue(NeoExecutionInfoLocation.linkOwnerExecutes("workflow", null,
false));
+ assertTrue(NeoExecutionInfoLocation.linkOwnerExecutes("workflow",
"workflow", true));
+ assertFalse(NeoExecutionInfoLocation.linkOwnerExecutes(null, "action",
false));
+ }
+
+ @Test
+ void relinkMovesExecutesFromTheOwnerToTheLauncher() {
+ String cypher = NeoExecutionInfoLocation.relinkLaunchedChildrenCypher();
+
+ assertTrue(cypher.contains("launchedById = $id"));
+ assertTrue(cypher.contains("MERGE (launcher)-[:EXECUTES]->(child)"));
+ assertTrue(cypher.contains("WHERE owner.id = child.parentId"));
+ assertTrue(cypher.contains("DELETE shortcut"));
+ assertTrue(cypher.contains("child.parentId <> $id"));
+ }
+
@Test
void batchDeleteLimitsEachTransaction() {
String cypher =
diff --git
a/plugins/tech/opensearch/src/main/java/org/apache/hop/execution/opensearch/OpenSearchExecutionInfoLocation.java
b/plugins/tech/opensearch/src/main/java/org/apache/hop/execution/opensearch/OpenSearchExecutionInfoLocation.java
index 0fd1499831..c519e53d84 100644
---
a/plugins/tech/opensearch/src/main/java/org/apache/hop/execution/opensearch/OpenSearchExecutionInfoLocation.java
+++
b/plugins/tech/opensearch/src/main/java/org/apache/hop/execution/opensearch/OpenSearchExecutionInfoLocation.java
@@ -19,6 +19,7 @@
package org.apache.hop.execution.opensearch;
import com.fasterxml.jackson.databind.ObjectMapper;
+import java.io.IOException;
import java.net.URI;
import java.text.SimpleDateFormat;
import java.time.ZoneId;
@@ -183,13 +184,7 @@ public class OpenSearchExecutionInfoLocation extends
BaseCachingExecutionInfoLoc
cacheEntry.calculateSummary();
URI uri = URI.create(actualUrl);
- URI postUri;
-
- if (StringUtils.isEmpty(cacheEntry.getInternalId())) {
- postUri = uri.resolve(actualIndexName + "/_doc");
- } else {
- postUri = uri.resolve(actualIndexName + "/_doc/" +
cacheEntry.getInternalId());
- }
+ URI postUri = uri.resolve(documentPath(actualIndexName,
cacheEntry.getInternalId()));
ObjectMapper mapper = new ObjectMapper();
String body = mapper.writeValueAsString(cacheEntry);
@@ -404,11 +399,7 @@ public class OpenSearchExecutionInfoLocation extends
BaseCachingExecutionInfoLoc
return null;
}
- // Convert this source object to CacheEntry
- //
- String json = jSource.toJSONString();
- ObjectMapper objectMapper = new ObjectMapper(HopJson.newFactory());
- return objectMapper.readValue(json, CacheEntry.class);
+ return cacheEntryFromSearchHit(jHit);
} catch (Exception e) {
throw new HopException(
"Error loading execution information location from OpenSearch using
executionId '"
@@ -418,6 +409,55 @@ public class OpenSearchExecutionInfoLocation extends
BaseCachingExecutionInfoLoc
}
}
+ /**
+ * Re-read by document id. {@code GET index/_doc/id} returns the latest
write. A search can still
+ * return the document from before that write.
+ */
+ @Override
+ protected CacheEntry loadFreshCacheEntry(CacheEntry loaded) throws
HopException {
+ if (loaded != null && StringUtils.isNotEmpty(loaded.getInternalId())) {
+ CacheEntry byDocumentId =
loadCacheEntryByDocumentId(loaded.getInternalId());
+ if (byDocumentId != null) {
+ return byDocumentId;
+ }
+ }
+ return super.loadFreshCacheEntry(loaded);
+ }
+
+ private CacheEntry loadCacheEntryByDocumentId(String internalId) throws
HopException {
+ try {
+ URI uri = URI.create(actualUrl);
+ URI getUri = uri.resolve(documentGetPath(actualIndexName, internalId));
+ RestCaller restCaller =
+ new RestCaller(
+ metadataProvider,
+ getUri.toString(),
+ actualUsername,
+ actualPassword,
+ "GET",
+ "",
+ ignoreSsl,
+ getHeaders());
+ String responseBody = restCaller.execute();
+ Long statusCode = restCaller.getStatusCode();
+ if (statusCode != null && statusCode.longValue() == 404L) {
+ return null;
+ }
+ checkStatusCode(responseBody, statusCode, 200L);
+ JSONParser parser = new JSONParser();
+ JSONObject body = (JSONObject) parser.parse(responseBody);
+ if (body == null || Boolean.FALSE.equals(body.get("found"))) {
+ return null;
+ }
+ return cacheEntryFromSearchHit(body);
+ } catch (HopException e) {
+ throw e;
+ } catch (Exception e) {
+ throw new HopException(
+ "Error loading execution information from OpenSearch document '" +
internalId + "'", e);
+ }
+ }
+
@Override
protected void retrieveIds(
boolean includeChildren, Set<DatedId> ids, int limit, IExecutionSelector
selector)
@@ -564,6 +604,47 @@ public class OpenSearchExecutionInfoLocation extends
BaseCachingExecutionInfoLoc
return indexName + "/_refresh";
}
+ /**
+ * Document path for a create or update. A later re-read uses {@link
#documentGetPath} so the
+ * latest write is visible without waiting for a refresh on this request.
+ */
+ static String documentPath(String indexName, String internalId) {
+ String path = indexName + "/_doc";
+ if (StringUtils.isNotEmpty(internalId)) {
+ path = path + "/" + internalId;
+ }
+ return path;
+ }
+
+ /**
+ * Realtime read of one document. The get-by-id API returns the latest
stored source, including a
+ * write that search has not refreshed yet.
+ */
+ static String documentGetPath(String indexName, String internalId) {
+ return indexName + "/_doc/" + internalId;
+ }
+
+ /**
+ * {@code internalId} is not part of {@code _source}. The search hit {@code
_id} is what a later
+ * update must post to, or the next save creates a second document.
+ */
+ static CacheEntry cacheEntryFromSearchHit(JSONObject hit) throws IOException
{
+ if (hit == null) {
+ return null;
+ }
+ JSONObject source = (JSONObject) hit.get("_source");
+ if (source == null) {
+ return null;
+ }
+ ObjectMapper objectMapper = new ObjectMapper(HopJson.newFactory());
+ CacheEntry entry = objectMapper.readValue(source.toJSONString(),
CacheEntry.class);
+ Object docId = hit.get("_id");
+ if (docId != null) {
+ entry.setInternalId(String.valueOf(docId));
+ }
+ return entry;
+ }
+
@Override
public int deleteExecutions(Date olderThan, IProgressMonitor monitor) throws
HopException {
int deleted = 0;
diff --git
a/plugins/tech/opensearch/src/test/java/org/apache/hop/execution/opensearch/OpenSearchExecutionInfoLocationCleanupTest.java
b/plugins/tech/opensearch/src/test/java/org/apache/hop/execution/opensearch/OpenSearchExecutionInfoLocationCleanupTest.java
index 010cac4033..fae945c94b 100644
---
a/plugins/tech/opensearch/src/test/java/org/apache/hop/execution/opensearch/OpenSearchExecutionInfoLocationCleanupTest.java
+++
b/plugins/tech/opensearch/src/test/java/org/apache/hop/execution/opensearch/OpenSearchExecutionInfoLocationCleanupTest.java
@@ -19,10 +19,13 @@ package org.apache.hop.execution.opensearch;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNull;
import static org.junit.jupiter.api.Assertions.assertTrue;
import java.util.Date;
import java.util.TimeZone;
+import org.apache.hop.execution.caching.CacheEntry;
+import org.json.simple.JSONObject;
import org.junit.jupiter.api.Test;
class OpenSearchExecutionInfoLocationCleanupTest {
@@ -47,6 +50,48 @@ class OpenSearchExecutionInfoLocationCleanupTest {
"hop-executions/_refresh",
OpenSearchExecutionInfoLocation.refreshPath("hop-executions"));
}
+ @Test
+ void documentPathNamesTheDocumentWithoutWaitingForRefresh() {
+ assertEquals("hop/_doc",
OpenSearchExecutionInfoLocation.documentPath("hop", null));
+ assertEquals("hop/_doc",
OpenSearchExecutionInfoLocation.documentPath("hop", ""));
+ assertEquals("hop/_doc/abc",
OpenSearchExecutionInfoLocation.documentPath("hop", "abc"));
+ }
+
+ @Test
+ void documentGetPathReadsOneDocumentById() {
+ assertEquals("hop/_doc/abc",
OpenSearchExecutionInfoLocation.documentGetPath("hop", "abc"));
+ }
+
+ @Test
+ void searchHitKeepsTheOpenSearchDocumentId() throws Exception {
+ JSONObject source = new JSONObject();
+ source.put("id", "exec-1");
+ source.put("name", "run");
+ JSONObject hit = new JSONObject();
+ hit.put("_id", "os-doc-7");
+ hit.put("_source", source);
+
+ CacheEntry entry =
OpenSearchExecutionInfoLocation.cacheEntryFromSearchHit(hit);
+
+ assertEquals("exec-1", entry.getId());
+ assertEquals("run", entry.getName());
+ assertEquals("os-doc-7", entry.getInternalId());
+ }
+
+ @Test
+ void searchHitWithoutADocumentIdLeavesTheInternalIdEmpty() throws Exception {
+ JSONObject source = new JSONObject();
+ source.put("id", "exec-1");
+ JSONObject hit = new JSONObject();
+ hit.put("_source", source);
+
+ CacheEntry entry =
OpenSearchExecutionInfoLocation.cacheEntryFromSearchHit(hit);
+
+ assertEquals("exec-1", entry.getId());
+ assertNull(entry.getInternalId());
+ assertNull(OpenSearchExecutionInfoLocation.cacheEntryFromSearchHit(null));
+ }
+
@Test
void cleanupSqlWithoutACutoffStillPages() {
String sql = OpenSearchExecutionInfoLocation.cleanupSql("hop-executions",
"", null, 200);
