You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kie.apache.org by fj...@apache.org on 2025/06/30 11:12:40 UTC
(incubator-kie-kogito-apps) branch main updated: [Fix_#3966] Bulk update in data-index-service-common for model process instance storage (#2237)
This is an automated email from the ASF dual-hosted git repository.
fjtiradosarti pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/incubator-kie-kogito-apps.git
The following commit(s) were added to refs/heads/main by this push:
new d1c9c4cc8 [Fix_#3966] Bulk update in data-index-service-common for model process instance storage (#2237)
d1c9c4cc8 is described below
commit d1c9c4cc8161cb688edb81ebec544d99e6d1b9c2
Author: dovandat-future-aavn <15...@users.noreply.github.com>
AuthorDate: Mon Jun 30 18:12:36 2025 +0700
[Fix_#3966] Bulk update in data-index-service-common for model process instance storage (#2237)
* support to update bulk into mongodb for process instance events
* support to update bulk into mongodb for process instance events
* refine
* reformat
* fix failed tests
---
.../messaging/AbstractMessagingConsumerIT.java | 23 +++++++++++++---------
.../index/storage/ModelProcessInstanceStorage.java | 23 ++++++++++++++++------
2 files changed, 31 insertions(+), 15 deletions(-)
diff --git a/data-index/data-index-quarkus/data-index-service-quarkus-common/src/test/java/org/kie/kogito/index/service/messaging/AbstractMessagingConsumerIT.java b/data-index/data-index-quarkus/data-index-service-quarkus-common/src/test/java/org/kie/kogito/index/service/messaging/AbstractMessagingConsumerIT.java
index fee277a1c..bf447f9b1 100644
--- a/data-index/data-index-quarkus/data-index-service-quarkus-common/src/test/java/org/kie/kogito/index/service/messaging/AbstractMessagingConsumerIT.java
+++ b/data-index/data-index-quarkus/data-index-service-quarkus-common/src/test/java/org/kie/kogito/index/service/messaging/AbstractMessagingConsumerIT.java
@@ -33,6 +33,7 @@ import jakarta.inject.Inject;
import static io.restassured.RestAssured.given;
import static org.awaitility.Awaitility.await;
import static org.hamcrest.CoreMatchers.is;
+import static org.hamcrest.Matchers.hasItems;
@Timeout(10000)
public abstract class AbstractMessagingConsumerIT {
@@ -130,15 +131,19 @@ public abstract class AbstractMessagingConsumerIT {
await()
.atMost(timeout)
- .untilAsserted(() -> given().contentType(ContentType.JSON)
- .body("{ \"query\" : \"{ ProcessInstances { id, state } }\" }")
- .when().post("/graphql")
- .then().log().ifValidationFails().statusCode(200)
- .body("data.ProcessInstances.size()", is(2))
- .body("data.ProcessInstances[0].id", is(processInstanceId1))
- .body("data.ProcessInstances[0].state", is("ACTIVE"))
- .body("data.ProcessInstances[1].id", is(processInstanceId2))
- .body("data.ProcessInstances[1].state", is("ACTIVE")));
+ .untilAsserted(() -> {
+ var response = given().contentType(ContentType.JSON)
+ .body("{ \"query\" : \"{ ProcessInstances { id, state } }\" }")
+ .when().post("/graphql")
+ .then().extract().response();
+
+ // Continue with assertions
+ response.then().log().ifValidationFails().statusCode(200)
+ .body("data.ProcessInstances.size()", is(2))
+ .body("data.ProcessInstances.id", hasItems(processInstanceId1, processInstanceId2))
+ .body("data.ProcessInstances.findAll { it.id == '" + processInstanceId1 + "' }.state[0]", is("ACTIVE"))
+ .body("data.ProcessInstances.findAll { it.id == '" + processInstanceId2 + "' }.state[0]", is("ACTIVE"));
+ });
}
diff --git a/data-index/data-index-storage/data-index-storage-common/src/main/java/org/kie/kogito/index/storage/ModelProcessInstanceStorage.java b/data-index/data-index-storage/data-index-storage-common/src/main/java/org/kie/kogito/index/storage/ModelProcessInstanceStorage.java
index 07b645bb7..bedac0296 100644
--- a/data-index/data-index-storage/data-index-storage-common/src/main/java/org/kie/kogito/index/storage/ModelProcessInstanceStorage.java
+++ b/data-index/data-index-storage/data-index-storage-common/src/main/java/org/kie/kogito/index/storage/ModelProcessInstanceStorage.java
@@ -18,6 +18,9 @@
*/
package org.kie.kogito.index.storage;
+import java.util.HashMap;
+import java.util.Map;
+
import org.kie.kogito.event.process.MultipleProcessInstanceDataEvent;
import org.kie.kogito.event.process.ProcessInstanceDataEvent;
import org.kie.kogito.event.process.ProcessInstanceErrorDataEvent;
@@ -72,28 +75,36 @@ public class ModelProcessInstanceStorage extends ModelStorageFetcher<String, Pro
@Override
public void indexGroup(MultipleProcessInstanceDataEvent events) {
+ Map<String, ProcessInstance> processInstances = new HashMap<>();
for (ProcessInstanceDataEvent<?> event : events.getData()) {
+ ProcessInstance processInstance = processInstances.computeIfAbsent(event.getKogitoProcessInstanceId(), k -> findProcessInstance(event));
if (event instanceof ProcessInstanceErrorDataEvent) {
- index(event, errorMerger);
+ errorMerger.merge(processInstance, event);
} else if (event instanceof ProcessInstanceNodeDataEvent) {
- index(event, nodeMerger);
+ nodeMerger.merge(processInstance, event);
} else if (event instanceof ProcessInstanceSLADataEvent) {
- index(event, slaMerger);
+ slaMerger.merge(processInstance, event);
} else if (event instanceof ProcessInstanceStateDataEvent) {
- index(event, stateMerger);
+ stateMerger.merge(processInstance, event);
} else if (event instanceof ProcessInstanceVariableDataEvent) {
- index(event, variableMerger);
+ variableMerger.merge(processInstance, event);
}
}
+ processInstances.values().forEach(processInstance -> storage.put(processInstance.getId(), processInstance));
}
private <T extends ProcessInstanceDataEvent<?>> void index(T event, ProcessInstanceEventMerger merger) {
+ ProcessInstance processInstance = findProcessInstance(event);
+ storage.put(event.getKogitoProcessInstanceId(), merger.merge(processInstance, event));
+ }
+
+ private <T extends ProcessInstanceDataEvent<?>> ProcessInstance findProcessInstance(T event) {
ProcessInstance processInstance = storage.get(event.getKogitoProcessInstanceId());
if (processInstance == null) {
processInstance = new ProcessInstance();
processInstance.setId(event.getKogitoProcessInstanceId());
processInstance.setProcessId(event.getKogitoProcessId());
}
- storage.put(event.getKogitoProcessInstanceId(), merger.merge(processInstance, event));
+ return processInstance;
}
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@kie.apache.org
For additional commands, e-mail: commits-help@kie.apache.org