[kie-issues-249] Data Index Events changes for allowing historical data

.ci/jenkins/Jenkinsfile

@@ -10,6 +10,7 @@ pipeline {
         timestamps()
         timeout(time: 480, unit: 'MINUTES')
         disableConcurrentBuilds(abortPrevious: true)
+        skipDefaultCheckout()
     }
     environment {
         BUILDCHAIN_PROJECT = 'apache/incubator-kie-kogito-apps'
@@ -20,6 +21,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

@@ -277,6 +277,7 @@ public void testProcessInstanceEvents() throws IOException {
 
     public void testProcessGatewayAPI() throws IOException {
         String pId2 = createTestProcessInstance();
+
         await()
                 .atMost(TIMEOUT)
                 .untilAsserted(() -> getProcessInstanceById(pId2, "ACTIVE"));
@@ -474,6 +475,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} } }\"}")
@@ -485,6 +498,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

@@ -30,10 +30,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

@@ -51,7 +51,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

@@ -33,9 +33,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);
+
+}