[kie-issues-249] data index improvements

.ci/jenkins/Jenkinsfile

@@ -20,6 +20,7 @@ pipeline {
     stages {
         stage('Initialize') {
             steps {
+                cleanWs()
                 script {
                     // load `pr_check.groovy` file from kogito-pipelines:main
                     dir('kogito-pipelines') {

apps-integration-tests/integration-tests-data-index-service/integration-tests-data-index-service-common/src/test/java/org/kie/kogito/index/AbstractProcessDataIndexIT.java

@@ -274,6 +274,7 @@ public void testProcessInstanceEvents() throws IOException {
 
     public void testProcessGatewayAPI() throws IOException {
         String pId2 = createTestProcessInstance();
+
         await()
                 .atMost(TIMEOUT)
                 .untilAsserted(() -> getProcessInstanceById(pId2, "ACTIVE"));
@@ -343,6 +344,8 @@ public void testProcessGatewayAPI() throws IOException {
                         .statusCode(200)
                         .body("errors", nullValue()));
 
+        System.out.println("PROCESS INSTANCE 2!!! UPDATED " + taskSchema);
+
         await()
                 .atMost(TIMEOUT)
                 .untilAsserted(() -> given().spec(dataIndexSpec()).contentType(ContentType.JSON)
@@ -471,6 +474,18 @@ public void testProcessGatewayAPIComments(String taskId, String processInstanceI
                         .body("$.size()", is(1))
                         .body("[0].content", is(commentContent)));
 
+        await()
+                .atMost(TIMEOUT)
+                .untilAsserted(() -> given().spec(dataIndexSpec()).contentType(ContentType.JSON)
+                        .body("{ \"query\" : \"{ UserTaskInstances (where: { processInstanceId: {equal: \\\"" + processInstanceId + "\\\"}}) { " +
+                                "id description priority potentialGroups comments {id content updatedBy updatedAt} } }\"}")
+                        .when().post("/graphql")
+                        .then()
+                        .statusCode(200)
+                        .body("data.UserTaskInstances[0].comments", notNullValue())
+                        .body("data.UserTaskInstances[0].comments.size()", is(1))
+                        .extract().jsonPath().getMap("data.UserTaskInstances[0].comments[0]"));
+
         Map<String, String> commentMap = given().spec(dataIndexSpec()).contentType(ContentType.JSON)
                 .body("{ \"query\" : \"{ UserTaskInstances (where: { processInstanceId: {equal: \\\"" + processInstanceId + "\\\"}}) { " +
                         "id description priority potentialGroups comments {id content updatedBy updatedAt} } }\"}")
@@ -482,6 +497,7 @@ public void testProcessGatewayAPIComments(String taskId, String processInstanceI
                 .body("data.UserTaskInstances[0].potentialGroups[0]", equalTo("managers"))
                 .body("data.UserTaskInstances[0].comments.size()", is(1))
                 .extract().jsonPath().getMap("data.UserTaskInstances[0].comments[0]");
+
         checkExpectedCreatedItemData(commentCreationResult, commentMap);
         String commentNewContent = "commentNewContent";
         String commentUpdateResult = given().spec(dataIndexSpec()).contentType(ContentType.JSON)

apps-integration-tests/integration-tests-data-index-service/integration-tests-data-index-service-quarkus/src/main/resources/application.properties

@@ -11,10 +11,4 @@ mp.messaging.outgoing.kogito-usertaskinstances-events.connector=smallrye-kafka
 mp.messaging.outgoing.kogito-usertaskinstances-events.topic=kogito-usertaskinstances-events
 mp.messaging.outgoing.kogito-usertaskinstances-events.value.serializer=org.apache.kafka.common.serialization.StringSerializer
 mp.messaging.outgoing.kogito-usertaskinstances-events.group.id=kogito-data-index-it
-mp.messaging.outgoing.kogito-usertaskinstances-events.auto.offset.reset=earliest
-
-mp.messaging.outgoing.kogito-variables-events.connector=smallrye-kafka
-mp.messaging.outgoing.kogito-variables-events.topic=kogito-variables-events
-mp.messaging.outgoing.kogito-variables-events.value.serializer=org.apache.kafka.common.serialization.StringSerializer
-mp.messaging.outgoing.kogito-variables-events.group.id=kogito-data-index-it
-mp.messaging.outgoing.kogito-variables-events.auto.offset.reset=earliest
+mp.messaging.outgoing.kogito-usertaskinstances-events.auto.offset.reset=earliest

apps-integration-tests/integration-tests-data-index-service/integration-tests-data-index-service-quarkus/src/test/java/org/kie/kogito/index/quarkus/kafka/KogitoServiceRandomPortQuarkusKafkaTestResource.java

@@ -48,7 +48,6 @@ protected Map<String, String> getProperties() {
         properties.put(KOGITO_SERVICE_URL, "http://host.testcontainers.internal:" + getTestResource().getMappedPort());
         properties.put("mp.messaging.outgoing.kogito-processinstances-events.connector", "smallrye-kafka");
         properties.put("mp.messaging.outgoing.kogito-usertaskinstances-events.connector", "smallrye-kafka");
-        properties.put("mp.messaging.outgoing.kogito-variables-events.connector", "smallrye-kafka");
 
         return properties;
     }

apps-integration-tests/integration-tests-data-index-service/integration-tests-data-index-service-quarkus/src/test/resources/application.properties

@@ -14,9 +14,5 @@ mp.messaging.outgoing.kogito-usertaskinstances-events.connector=smallrye-kafka
 mp.messaging.outgoing.kogito-usertaskinstances-events.topic=kogito-usertaskinstances-events
 mp.messaging.outgoing.kogito-usertaskinstances-events.value.serializer=org.apache.kafka.common.serialization.StringSerializer
 
-mp.messaging.outgoing.kogito-variables-events.connector=smallrye-kafka
-mp.messaging.outgoing.kogito-variables-events.topic=kogito-variables-events
-mp.messaging.outgoing.kogito-variables-events.value.serializer=org.apache.kafka.common.serialization.StringSerializer
-
 quarkus.http.auth.permission.source-files-addon.paths=/management/*
 quarkus.http.auth.permission.source-files-addon.policy=permit

data-index/data-index-common/src/main/java/org/kie/kogito/index/event/mapper/ProcessInstanceErrorDataEventMerger.java

@@ -0,0 +1,45 @@
+/*
+ * 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.kie.kogito.index.event.mapper;
+
+import javax.enterprise.context.ApplicationScoped;
+
+import org.kie.kogito.event.process.ProcessInstanceDataEvent;
+import org.kie.kogito.event.process.ProcessInstanceErrorDataEvent;
+import org.kie.kogito.index.model.ProcessInstance;
+import org.kie.kogito.index.model.ProcessInstanceError;
+
+@ApplicationScoped
+public class ProcessInstanceErrorDataEventMerger implements ProcessInstanceEventMerger {
+
+    @Override
+    public boolean accept(ProcessInstanceDataEvent<?> event) {
+        return event instanceof ProcessInstanceErrorDataEvent;
+    }
+
+    @Override
+    public void merge(ProcessInstance pi, ProcessInstanceDataEvent<?> data) {
+        ProcessInstanceErrorDataEvent event = (ProcessInstanceErrorDataEvent) data;
+        ProcessInstanceError error = new ProcessInstanceError();
+        error.setMessage(event.getData().getErrorMessage());
+        error.setNodeDefinitionId(event.getData().getNodeDefinitionId());
+        pi.setError(error);
+    }
+
+}

data-index/data-index-common/src/main/java/org/kie/kogito/index/event/mapper/ProcessInstanceEventMerger.java

@@ -0,0 +1,30 @@
+/*
+ * 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.kie.kogito.index.event.mapper;
+
+import org.kie.kogito.event.process.ProcessInstanceDataEvent;
+import org.kie.kogito.index.model.ProcessInstance;
+
+public interface ProcessInstanceEventMerger {
+
+    boolean accept(ProcessInstanceDataEvent<?> event);
+
+    void merge(ProcessInstance pi, ProcessInstanceDataEvent<?> event);
+
+}