Skip to content

Commit 45b5391

Browse files
authored
[Fix #1738] Replace setMetadata by fine grain manipulation. (#1742)
* [Fix #1738] Replace setMetadata by fine grain manipulation. Signed-off-by: Francisco Javier Tirado Sarti <ftirados@ibm.com> * [Fix #1738] Adding safeguard to avoid overriding keys set by user Signed-off-by: Francisco Javier Tirado Sarti <ftirados@ibm.com> --------- Signed-off-by: Francisco Javier Tirado Sarti <ftirados@ibm.com>
1 parent bfdffd0 commit 45b5391

4 files changed

Lines changed: 51 additions & 10 deletions

File tree

‎impl/core/src/main/java/io/serverlessworkflow/impl/WorkflowMutableInstance.java‎

Lines changed: 1 addition & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -53,7 +53,7 @@ public class WorkflowMutableInstance implements WorkflowInstance {
5353
protected AtomicReference<CompletableFuture<WorkflowModel>> futureRef = new AtomicReference<>();
5454
protected Instant completedAt;
5555

56-
private Map<String, Object> additionalObjects = new ConcurrentHashMap<>();
56+
protected Map<String, Object> additionalObjects = new ConcurrentHashMap<>();
5757

5858
protected final Map<String, Integer> iterationsMap = new ConcurrentHashMap<>();
5959

@@ -448,10 +448,6 @@ public void addCancelable(CompletableFuture<?> cancelable) {
448448
}
449449
}
450450

451-
protected void setMetadata(Map<String, Object> metadata) {
452-
this.additionalObjects = new ConcurrentHashMap<>(metadata);
453-
}
454-
455451
@Override
456452
public <T> T addMetadataIfAbsent(String key, Supplier<T> supplier) {
457453
return (T) additionalObjects.computeIfAbsent(key, k -> supplier.get());

‎impl/persistence/api/src/main/java/io/serverlessworkflow/impl/persistence/PersistenceTaskInfo.java‎

Lines changed: 6 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -15,4 +15,9 @@
1515
*/
1616
package io.serverlessworkflow.impl.persistence;
1717

18-
public interface PersistenceTaskInfo {}
18+
import java.util.Map;
19+
20+
public interface PersistenceTaskInfo {
21+
22+
Map<String, Object> additionalObjects();
23+
}

‎impl/persistence/api/src/main/java/io/serverlessworkflow/impl/persistence/RetriedTaskInfo.java‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -17,7 +17,7 @@
1717

1818
import java.util.Map;
1919

20-
public record RetriedTaskInfo(int retryAttempt, Map<String, Object> metadata)
20+
public record RetriedTaskInfo(int retryAttempt, Map<String, Object> additionalObjects)
2121
implements PersistenceTaskInfo {
2222
public RetriedTaskInfo(int retryAttempt) {
2323
this(retryAttempt, Map.of());

‎impl/persistence/api/src/main/java/io/serverlessworkflow/impl/persistence/WorkflowPersistenceInstance.java‎

Lines changed: 43 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -24,12 +24,16 @@
2424
import io.serverlessworkflow.impl.WorkflowMutableInstance;
2525
import io.serverlessworkflow.impl.WorkflowStatus;
2626
import io.serverlessworkflow.impl.executors.TransitionInfo;
27+
import java.util.HashSet;
2728
import java.util.Optional;
29+
import java.util.Set;
2830
import java.util.concurrent.CompletableFuture;
31+
import java.util.function.Supplier;
2932

3033
public class WorkflowPersistenceInstance extends WorkflowMutableInstance {
3134

3235
private final PersistenceWorkflowInfo info;
36+
private Set<String> userMetaKeys = new HashSet<>();
3337

3438
public static WorkflowInstance of(WorkflowDefinition definition, PersistenceWorkflowInfo info) {
3539
return definition
@@ -48,7 +52,7 @@ private WorkflowPersistenceInstance(WorkflowDefinition definition, PersistenceWo
4852
}
4953
});
5054
this.startedAt = info.startedAt();
51-
setMetadata(info.metadata());
55+
additionalObjects.putAll(info.metadata());
5256
}
5357

5458
@Override
@@ -84,7 +88,6 @@ public void restoreContext(WorkflowContext workflow, TaskContext context) {
8488
: workflow.definition().taskExecutor(completedTaskInfo.nextPosition()),
8589
completedTaskInfo.isEndNode()));
8690
workflow.context(completedTaskInfo.context());
87-
setMetadata(completedTaskInfo.additionalObjects());
8891
} else if (taskInfo instanceof RetriedTaskInfo retriedTaskInfo) {
8992
if (context.retryAttempt() == 0) {
9093
context.retryAttempt(retriedTaskInfo.retryAttempt());
@@ -100,7 +103,44 @@ public void restoreContext(WorkflowContext workflow, TaskContext context) {
100103
}
101104
searchContext = tryContext.parent();
102105
}
103-
setMetadata(retriedTaskInfo.metadata());
106+
}
107+
Set<String> retainKeys;
108+
synchronized (userMetaKeys) {
109+
retainKeys = new HashSet<>(userMetaKeys);
110+
taskInfo
111+
.additionalObjects()
112+
.forEach(
113+
(k, v) -> {
114+
if (!retainKeys.contains(k)) {
115+
additionalObjects.put(k, v);
116+
retainKeys.add(k);
117+
}
118+
});
119+
additionalObjects.keySet().retainAll(retainKeys);
120+
}
121+
}
122+
123+
@Override
124+
public <T> T addMetadataIfAbsent(String key, Supplier<T> supplier) {
125+
synchronized (userMetaKeys) {
126+
userMetaKeys.add(key);
127+
return super.addMetadataIfAbsent(key, supplier);
128+
}
129+
}
130+
131+
@Override
132+
public void removeMetadata(String key) {
133+
synchronized (userMetaKeys) {
134+
userMetaKeys.add(key);
135+
super.removeMetadata(key);
136+
}
137+
}
138+
139+
@Override
140+
public <T> Optional<T> removeMetadata(String key, Class<T> clazz) {
141+
synchronized (userMetaKeys) {
142+
userMetaKeys.add(key);
143+
return super.removeMetadata(key, clazz);
104144
}
105145
}
106146
}

0 commit comments

Comments
 (0)