This is an automated email from the ASF dual-hosted git repository.
pefernan pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/incubator-kie.git
The following commit(s) were added to refs/heads/main by this push:
new 4e3727fab40 [incubator-kie-7106]
fix(`kogito-quarkus-workflow-extension-common`): scope the UnitOfWorkManager
bean to the application, not to the JVM (#7107)
4e3727fab40 is described below
commit 4e3727fab4033c10338619c3110dc2e668fb4f24
Author: Thiago Lugli <[email protected]>
AuthorDate: Wed Sep 16 10:46:45 2026 -0300
[incubator-kie-7106] fix(`kogito-quarkus-workflow-extension-common`): scope
the UnitOfWorkManager bean to the application, not to the JVM (#7107)
* [NO-ISSUE] chore: fix the jobs-service IT `shared/` ignore rule stale
since the repo merge
The four integration-tests-jobs-service-quarkus modules copy the shared
BPMN/SWF
definitions into their own src/main/resources/shared during
generate-resources, so
any working copy that has been built shows 24 generated files as untracked.
An ignore rule for exactly this already existed, but it still pointed at
apps-integration-tests/integration-tests-jobs-service/** -- the path this
content
had before the kogito-apps repository was merged into this one. The rule has
therefore matched nothing since the merge.
Point it at the current layout and record why the path changed, so the next
person
to see these files does not add a second, duplicate rule. No tracked file
lives
under that path, so nothing committed is shadowed.
* [incubator-kie-7106] fix(quarkus): scope the UnitOfWorkManager bean to
the application, not to the JVM
In Quarkus dev mode, the first or second live reload of an application using
kogito-addons-quarkus-data-index-jpa fails to start with
`IllegalStateException: EntityManagerFactory is closed`, raised from
ProcessDefinitionRegistration -> DataIndexEventPublisher -> IndexingService
->
QuarkusProcessDefinitionEntityStorage.get(). Once it starts, it never
recovers: every
subsequent reload fails the same way.
Cause: the CDI producer returned
StaticUnitOfWorkManger.staticUnitOfWorkManager(). That
static lives in a class loaded by the Quarkus *base* runtime ClassLoader,
which a live
reload does not recreate, so the UnitOfWorkManager -- and the
BaseEventManager inside it --
outlives the application it belongs to. On every boot AbstractProcessConfig
runs
eventPublishers.forEach(p ->
unitOfWorkManager().eventManager().addPublisher(p));
unitOfWorkListeners.forEach(l -> unitOfWorkManager().register(l));
and both addPublisher and register are add-only. So each reload appends
another
generation's publishers and listeners to the same manager, and publish()
then fans out to
publishers belonging to applications that were shut down long ago. The
data-index JPA
publisher is the one that synchronously opens a Hibernate session, so it is
the one that
turns a stale reference into a visible exception; the leak itself is not
specific to it.
Measured across five consecutive dev-mode generations: Application is
correctly recreated
every time (@2b15af5c, @41d76578, @266a08bf, @6637bbb9, @72a2ae1f) while
the manager and
its event manager never are (DefaultUnitOfWorkManager@135ba,
BaseEventManager@78f53ca9
throughout), and the publisher set grows 1, 2, 3, 4, 5.
This is not a revert of dcaca447 ([Fix #3869] Using the same work and event
manager).
Before that commit the producer was a bare @Produces, i.e. @Dependent, so
every injection
point received its own manager and only ProcessConfig's carried the
publishers -- that is
what #3869 was. The defect there was the missing scope; dcaca447 obtained
sharing through a
JVM static, and it is that static's lifetime that is fixed here. @Singleton
provides the
same sharing through the container, so dcaca447's invariant is preserved.
@Singleton rather
than @ApplicationScoped so injection points hold the instance directly,
leaving the
unit-of-work path free of a client proxy.
LightProcessRuntimeServiceProvider is
deliberately left as dcaca447 set it.
Adds the regression test #3869 never shipped: `timerResume` parks on an
intermediate timer
so the instance is persisted and resumed by the job scheduler rather than
the request
thread, and the test asserts the data index sees the completion. Verified
failing before
this change (holds ACTIVE until the awaitility ceiling) and passing after,
both with
dcaca447 reverse-applied and with the producer unscoped.
---
.gitignore | 2 +-
.../integration-tests-process/pom.xml | 5 ++
.../src/main/resources/timerresume.bpmn | 45 ++++++++++++
.../it/JPAQuarkusAddonDataIndexTimerResumeIT.java | 80 ++++++++++++++++++++++
.../quarkus/workflow/KogitoBeanProducer.java | 8 ++-
5 files changed, 137 insertions(+), 3 deletions(-)
diff --git a/.gitignore b/.gitignore
index b892e8dd1e9..934cc1aa584 100644
--- a/.gitignore
+++ b/.gitignore
@@ -180,4 +180,4 @@ package-lock.json
!.rat-excludes
# Jobs Service ITs
-apps-integration-tests/integration-tests-jobs-service/**/src/main/resources/shared/
\ No newline at end of file
+kogito-apps-quarkus/integration-tests-jobs-service-quarkus/**/src/main/resources/shared/
\ No newline at end of file
diff --git
a/kogito-apps-quarkus/data-index-quarkus/kogito-addons-quarkus-data-index/kogito-addons-quarkus-data-index-jpa/integration-tests-process/pom.xml
b/kogito-apps-quarkus/data-index-quarkus/kogito-addons-quarkus-data-index/kogito-addons-quarkus-data-index-jpa/integration-tests-process/pom.xml
index da51da23cbf..ca4047d98ed 100644
---
a/kogito-apps-quarkus/data-index-quarkus/kogito-addons-quarkus-data-index/kogito-addons-quarkus-data-index-jpa/integration-tests-process/pom.xml
+++
b/kogito-apps-quarkus/data-index-quarkus/kogito-addons-quarkus-data-index/kogito-addons-quarkus-data-index-jpa/integration-tests-process/pom.xml
@@ -79,6 +79,11 @@
<artifactId>rest-assured</artifactId>
<scope>test</scope>
</dependency>
+ <dependency>
+ <groupId>org.awaitility</groupId>
+ <artifactId>awaitility</artifactId>
+ <scope>test</scope>
+ </dependency>
<dependency>
<groupId>org.kie</groupId>
<artifactId>kogito-addons-quarkus-data-index-postgresql-deployment</artifactId>
diff --git
a/kogito-apps-quarkus/data-index-quarkus/kogito-addons-quarkus-data-index/kogito-addons-quarkus-data-index-jpa/integration-tests-process/src/main/resources/timerresume.bpmn
b/kogito-apps-quarkus/data-index-quarkus/kogito-addons-quarkus-data-index/kogito-addons-quarkus-data-index-jpa/integration-tests-process/src/main/resources/timerresume.bpmn
new file mode 100644
index 00000000000..6729671dc33
--- /dev/null
+++
b/kogito-apps-quarkus/data-index-quarkus/kogito-addons-quarkus-data-index/kogito-addons-quarkus-data-index-jpa/integration-tests-process/src/main/resources/timerresume.bpmn
@@ -0,0 +1,45 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+ 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.
+-->
+<bpmn2:definitions xmlns:bpmn2="http://www.omg.org/spec/BPMN/20100524/MODEL";
+ xmlns:drools="http://www.jboss.org/drools";
+ xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance";
+ id="_timerResumeDefinitions"
+ targetNamespace="http://www.omg.org/bpmn20";>
+ <!-- Parks on a timer so the instance is persisted and resumed by the
scheduler. Guards #3869. -->
+ <bpmn2:process id="timerResume"
drools:packageName="org.kie.kogito.addons.quarkus.data.index.it"
+ drools:version="1.0" drools:adHoc="false" name="timerResume"
+ isExecutable="true" processType="Public">
+ <bpmn2:startEvent id="_timerResumeStart">
+ <bpmn2:outgoing>_timerResumeFlow1</bpmn2:outgoing>
+ </bpmn2:startEvent>
+ <bpmn2:intermediateCatchEvent id="_timerResumeWait" name="Wait">
+ <bpmn2:incoming>_timerResumeFlow1</bpmn2:incoming>
+ <bpmn2:outgoing>_timerResumeFlow2</bpmn2:outgoing>
+ <bpmn2:timerEventDefinition id="_timerResumeTimerDef">
+ <bpmn2:timeDuration
xsi:type="bpmn2:tFormalExpression">PT1S</bpmn2:timeDuration>
+ </bpmn2:timerEventDefinition>
+ </bpmn2:intermediateCatchEvent>
+ <bpmn2:endEvent id="_timerResumeEnd">
+ <bpmn2:incoming>_timerResumeFlow2</bpmn2:incoming>
+ </bpmn2:endEvent>
+ <bpmn2:sequenceFlow id="_timerResumeFlow1" sourceRef="_timerResumeStart"
targetRef="_timerResumeWait"/>
+ <bpmn2:sequenceFlow id="_timerResumeFlow2" sourceRef="_timerResumeWait"
targetRef="_timerResumeEnd"/>
+ </bpmn2:process>
+</bpmn2:definitions>
diff --git
a/kogito-apps-quarkus/data-index-quarkus/kogito-addons-quarkus-data-index/kogito-addons-quarkus-data-index-jpa/integration-tests-process/src/test/java/org/kie/kogito/addons/quarkus/data/index/it/JPAQuarkusAddonDataIndexTimerResumeIT.java
b/kogito-apps-quarkus/data-index-quarkus/kogito-addons-quarkus-data-index/kogito-addons-quarkus-data-index-jpa/integration-tests-process/src/test/java/org/kie/kogito/addons/quarkus/data/index/it/JPAQuarkusAddonDataIndexTimerResumeIT.java
new file mode 100644
index 00000000000..57c1c49d6e0
--- /dev/null
+++
b/kogito-apps-quarkus/data-index-quarkus/kogito-addons-quarkus-data-index/kogito-addons-quarkus-data-index-jpa/integration-tests-process/src/test/java/org/kie/kogito/addons/quarkus/data/index/it/JPAQuarkusAddonDataIndexTimerResumeIT.java
@@ -0,0 +1,80 @@
+/*
+ * 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.addons.quarkus.data.index.it;
+
+import java.time.Duration;
+
+import org.junit.jupiter.api.Test;
+
+import io.quarkus.test.junit.QuarkusIntegrationTest;
+import io.restassured.RestAssured;
+import io.restassured.http.ContentType;
+import io.restassured.response.ValidatableResponse;
+
+import static io.restassured.RestAssured.given;
+import static org.awaitility.Awaitility.await;
+import static org.hamcrest.CoreMatchers.is;
+import static org.hamcrest.CoreMatchers.notNullValue;
+
+/**
+ * Regression test for <a
href="https://github.com/apache/incubator-kie-kogito-runtimes/issues/3869";>#3869</a>:
+ * events emitted after a persisted instance is resumed must reach the data
index.
+ */
+@QuarkusIntegrationTest
+class JPAQuarkusAddonDataIndexTimerResumeIT {
+
+ private static final String PROCESS_ID = "timerResume";
+
+ static {
+ RestAssured.enableLoggingOfRequestAndResponseIfValidationFails();
+ }
+
+ @Test
+ void testEventsAfterTimerResumeReachDataIndex() {
+ String processInstanceId = given()
+ .contentType(ContentType.JSON)
+ .accept(ContentType.JSON)
+ .post("/" + PROCESS_ID)
+ .then()
+ .statusCode(201)
+ .body("id", is(notNullValue()))
+ .extract().path("id");
+
+ // Parked on the timer: indexed, but not finished yet.
+ queryState(processInstanceId).body("data.ProcessInstances.size()",
is(1))
+ .body("data.ProcessInstances[0].state", is("ACTIVE"));
+
+ // The timer fires, the instance is resumed off the request thread and
completes. If the
+ // resuming path publishes into a UnitOfWorkManager that carries no
publishers, the data
+ // index never learns about it and this never becomes COMPLETED.
+ await().atMost(Duration.ofSeconds(30))
+ .pollInterval(Duration.ofSeconds(1))
+ .untilAsserted(() -> queryState(processInstanceId)
+ .body("data.ProcessInstances.size()", is(1))
+ .body("data.ProcessInstances[0].state",
is("COMPLETED")));
+ }
+
+ private static ValidatableResponse queryState(String processInstanceId) {
+ return given().contentType(ContentType.JSON)
+ .body("{ \"query\" : \"{ProcessInstances(where: { id: {equal:
\\\"" + processInstanceId
+ + "\\\"}}){ id, state, processId } }\" }")
+ .when().post("/graphql")
+ .then().statusCode(200);
+ }
+}
diff --git
a/kogito-quarkus/extensions/kogito-quarkus-workflow-extension-common/kogito-quarkus-workflow-common/src/main/java/org/kie/kogito/quarkus/workflow/KogitoBeanProducer.java
b/kogito-quarkus/extensions/kogito-quarkus-workflow-extension-common/kogito-quarkus-workflow-common/src/main/java/org/kie/kogito/quarkus/workflow/KogitoBeanProducer.java
index 98f978b24bf..4807d5bf777 100644
---
a/kogito-quarkus/extensions/kogito-quarkus-workflow-extension-common/kogito-quarkus-workflow-common/src/main/java/org/kie/kogito/quarkus/workflow/KogitoBeanProducer.java
+++
b/kogito-quarkus/extensions/kogito-quarkus-workflow-extension-common/kogito-quarkus-workflow-common/src/main/java/org/kie/kogito/quarkus/workflow/KogitoBeanProducer.java
@@ -31,7 +31,8 @@ import
org.kie.kogito.process.version.ProjectVersionProcessVersionResolver;
import org.kie.kogito.services.jobs.impl.InMemoryJobContext;
import org.kie.kogito.services.jobs.impl.InMemoryJobService;
import org.kie.kogito.services.jobs.impl.InMemoryProcessJobExecutorFactory;
-import org.kie.kogito.services.uow.StaticUnitOfWorkManger;
+import org.kie.kogito.services.uow.CollectingUnitOfWorkFactory;
+import org.kie.kogito.services.uow.DefaultUnitOfWorkManager;
import org.kie.kogito.uow.UnitOfWorkManager;
import org.kie.kogito.usertask.UserTasks;
@@ -41,6 +42,7 @@ import io.quarkus.arc.properties.IfBuildProperty;
import jakarta.enterprise.context.ApplicationScoped;
import jakarta.enterprise.inject.Instance;
import jakarta.enterprise.inject.Produces;
+import jakarta.inject.Singleton;
@ApplicationScoped
public class KogitoBeanProducer {
@@ -57,10 +59,12 @@ public class KogitoBeanProducer {
return new DefaultCorrelationService();
}
+ // Must stay a per-application singleton: a JVM-wide static outlives a
dev-mode reload. See incubator-kie#7106.
@DefaultBean
@Produces
+ @Singleton
UnitOfWorkManager unitOfWorkManager() {
- return StaticUnitOfWorkManger.staticUnitOfWorkManager();
+ return new DefaultUnitOfWorkManager(new CollectingUnitOfWorkFactory());
}
@DefaultBean
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
