You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kie.apache.org by "Abhitocode (via GitHub)" <gi...@apache.org> on 2026/07/29 13:59:03 UTC
[PR] [incubator-kie-issues#2382] Handle concurrent duplicate ProcessDefinition insert [incubator-kie-drools]
Abhitocode opened a new pull request, #6850:
URL: https://github.com/apache/incubator-kie-drools/pull/6850
When a few replicas of the same app start together, they all see an empty state and insert the same `(id, version)` rows into `definitions`. All but the first hit a primary key violation and the replica dies on startup.
The reason is `em.merge()` only queues the INSERT, and `put` joins the transaction that indexing already runs in, so the INSERT is only sent when that outer transaction commits - after any `try`/`catch` around `put` has returned.
**Changes**
- **`ProcessDefinitionEntityStorage`** - new `put` overload with the actual fix: flush the insert right away so the failure happens where we can catch it, run it in a separate transaction so the rollback doesn't kill the caller's work, and on a duplicate just return the definition that's already stored.
- **`QuarkusProcessDefinitionEntityStorage`** (jpa-quarkus *and* postgresql) - overrides `put` to run the insert in a `@Transactional(REQUIRES_NEW)` method. Two copies of the same class, both used by an embedded add-on, so both need it.
- **`SpringBootProcessDefinitionEntityStorage`** (new, springboot) - same thing with a`TransactionTemplate`. `@Transactional` doesn't work here: Spring ignores it when a class calls its own method, so the insert would just run in the caller's transaction.
This is also why the catch sits *outside* the transaction, not inside, here a failed flush marks the transaction as rollback only, so the commit still fails after the `catch` has run. The exception is different too, Quarkus throws `ConstraintViolationException`, Spring Boot throws `DataIntegrityViolationException`.
- **`DataIndexStorageProducer`** (springboot) - removed the `processDefinitionEntityStorage` bean, otherwise there'd be two beans of the same type now that the class above provides it.
**Tests**
`AbstractConcurrentProcessDefinitionStorageIT` plus a PostgreSQL subclass per runtime. A plain JDBC connection acts as the competing replica, it inserts the row and holds it uncommitted, so the storage still sees nothing and queues its own insert, which blocks on the primary key index. The commit is triggered from `pg_locks` once that insert is actually waiting, so there are no sleeps.
Closes: https://github.com/apache/incubator-kie-issues/issues/2382
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@kie.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@kie.apache.org
For additional commands, e-mail: commits-help@kie.apache.org
Re: [PR] [incubator-kie-issues#2382] Handle concurrent duplicate ProcessDefinition insert [incubator-kie-drools]
Posted by "pefernan (via GitHub)" <gi...@apache.org>.
pefernan commented on code in PR #6850:
URL: https://github.com/apache/incubator-kie-drools/pull/6850#discussion_r3681715648
##########
kogito-data-index/data-index-storage/data-index-storage-jpa-common/src/main/java/org/kie/kogito/index/jpa/storage/ProcessDefinitionEntityStorage.java:
##########
@@ -44,4 +53,22 @@ public ProcessDefinitionEntityStorage(EntityManager em, Iterable<JsonPredicateBu
e.getVersion()), Optional.ofNullable(getInstance(predicateBuilder)), Optional.ofNullable(getInstance(processes)));
}
+ protected ProcessDefinition put(ProcessDefinitionKey key, ProcessDefinition value,
+ Function<Supplier<ProcessDefinition>, ProcessDefinition> isolatedTransaction) {
+ try {
+ return isolatedTransaction.apply(() -> {
+ super.put(key, value);
+ em.flush();
+ return value;
+ });
+ } catch (RuntimeException e) {
+ if (Throwables.getCausalChain(e).stream().noneMatch(ConstraintViolationException.class::isInstance)) {
+ throw e;
+ }
+ LOGGER.info("ProcessDefinition with id '{}' and version '{}' is already present, skipping insert.", key.getId(), key.getVersion());
+ LOGGER.debug("Duplicate ProcessDefinition insert suppressed", e);
+ return get(key);
+ }
+ }
+
Review Comment:
I think this code could be simplified to something like:
```suggestion
public ProcessDefinition put(ProcessDefinitionKey key, ProcessDefinition value) {
try {
return wrapInTransaction(() -> {
super.put(key, value);
em.flush();
return value;
});
} catch (RuntimeException e) {
if (Throwables.getCausalChain(e).stream().noneMatch(ConstraintViolationException.class::isInstance)) {
throw e;
}
LOGGER.info("ProcessDefinition with id '{}' and version '{}' is already present, skipping insert.", key.getId(), key.getVersion());
LOGGER.debug("Duplicate ProcessDefinition insert suppressed", e);
return wrapInTransaction(() -> get(key));
}
}
protected abstract ProcessDefinition wrapInTransaction(Supplier<ProcessDefinition> supplier);
```
##########
kogito-data-index/data-index-storage/data-index-storage-jpa-common/src/main/java/org/kie/kogito/index/jpa/storage/ProcessDefinitionEntityStorage.java:
##########
@@ -44,4 +53,22 @@ public ProcessDefinitionEntityStorage(EntityManager em, Iterable<JsonPredicateBu
e.getVersion()), Optional.ofNullable(getInstance(predicateBuilder)), Optional.ofNullable(getInstance(processes)));
}
+ protected ProcessDefinition put(ProcessDefinitionKey key, ProcessDefinition value,
+ Function<Supplier<ProcessDefinition>, ProcessDefinition> isolatedTransaction) {
+ try {
+ return isolatedTransaction.apply(() -> {
+ super.put(key, value);
+ em.flush();
+ return value;
+ });
+ } catch (RuntimeException e) {
+ if (Throwables.getCausalChain(e).stream().noneMatch(ConstraintViolationException.class::isInstance)) {
+ throw e;
+ }
+ LOGGER.info("ProcessDefinition with id '{}' and version '{}' is already present, skipping insert.", key.getId(), key.getVersion());
+ LOGGER.debug("Duplicate ProcessDefinition insert suppressed", e);
+ return get(key);
+ }
+ }
+
Review Comment:
Then each concrete implementation should only implement thet insertInNewTransaction method with the @Transactional annotation like:
```java
@Override
@Transactional(TxType.REQUIRES_NEW)
protected ProcessDefinition wrapInTransaction(Supplier<ProcessDefinition> supplier) {
return supplier.get();
}
// Spring Boot
@Override
@Transactional(Transactional.TxType.REQUIRES_NEW)
protected ProcessDefinition wrapInTransaction(Supplier<ProcessDefinition> supplier) {
return supplier.get();
}
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@kie.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@kie.apache.org
For additional commands, e-mail: commits-help@kie.apache.org
Re: [PR] [incubator-kie-issues#2382] Handle concurrent duplicate ProcessDefinition insert [incubator-kie-drools]
Posted by "martinweiler (via GitHub)" <gi...@apache.org>.
martinweiler merged PR #6850:
URL: https://github.com/apache/incubator-kie-drools/pull/6850
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@kie.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@kie.apache.org
For additional commands, e-mail: commits-help@kie.apache.org