This is an automated email from the ASF dual-hosted git repository.
martinweiler pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/incubator-kie-drools.git
The following commit(s) were added to refs/heads/main by this push:
new 30c6015f64d [incubator-kie-issues#2382] Handle concurrent duplicate
ProcessDefinition insert (#6850)
30c6015f64d is described below
commit 30c6015f64dc60626f8fe7e5e74c26f2710e7a11
Author: Abhiram Gundala <[email protected]>
AuthorDate: Thu Jul 30 15:11:32 2026 -0400
[incubator-kie-issues#2382] Handle concurrent duplicate ProcessDefinition
insert (#6850)
* Handle concurrent duplicate ProcessDefinition insert
* Simplify transaction wrapping
---
.../QuarkusProcessDefinitionEntityStorage.java | 9 ++
...greSQLConcurrentProcessDefinitionStorageIT.java | 49 ++++++++
.../QuarkusProcessDefinitionEntityStorage.java | 9 ++
.../storage/DataIndexStorageProducer.java | 11 --
.../SpringBootProcessDefinitionEntityStorage.java | 58 +++++++++
...greSQLConcurrentProcessDefinitionStorageIT.java | 56 +++++++++
.../storage/ProcessDefinitionEntityStorage.java | 30 ++++-
...stractConcurrentProcessDefinitionStorageIT.java | 129 +++++++++++++++++++++
8 files changed, 339 insertions(+), 12 deletions(-)
diff --git
a/kogito-apps-quarkus/data-index-storage-quarkus/data-index-storage-jpa-quarkus/src/main/java/org/kie/kogito/index/jpa/quarkus/QuarkusProcessDefinitionEntityStorage.java
b/kogito-apps-quarkus/data-index-storage-quarkus/data-index-storage-jpa-quarkus/src/main/java/org/kie/kogito/index/jpa/quarkus/QuarkusProcessDefinitionEntityStorage.java
index 253db8f34b4..254e740f1de 100644
---
a/kogito-apps-quarkus/data-index-storage-quarkus/data-index-storage-jpa-quarkus/src/main/java/org/kie/kogito/index/jpa/quarkus/QuarkusProcessDefinitionEntityStorage.java
+++
b/kogito-apps-quarkus/data-index-storage-quarkus/data-index-storage-jpa-quarkus/src/main/java/org/kie/kogito/index/jpa/quarkus/QuarkusProcessDefinitionEntityStorage.java
@@ -19,16 +19,19 @@
package org.kie.kogito.index.jpa.quarkus;
import java.util.Collections;
+import java.util.function.Supplier;
import org.eclipse.microprofile.config.inject.ConfigProperty;
import org.kie.kogito.index.jpa.storage.JsonPredicateBuilder;
import org.kie.kogito.index.jpa.storage.ProcessDefinitionEntityStorage;
+import org.kie.kogito.index.model.ProcessDefinition;
import org.kie.kogito.process.Processes;
import jakarta.enterprise.context.ApplicationScoped;
import jakarta.enterprise.inject.Instance;
import jakarta.inject.Inject;
import jakarta.persistence.EntityManager;
+import jakarta.transaction.Transactional;
@ApplicationScoped
public class QuarkusProcessDefinitionEntityStorage extends
ProcessDefinitionEntityStorage {
@@ -39,4 +42,10 @@ public class QuarkusProcessDefinitionEntityStorage extends
ProcessDefinitionEnti
super(em, predicateBuilder, dataIsolationEnabled ? processesInstance :
Collections.emptyList());
}
+ @Override
+ @Transactional(Transactional.TxType.REQUIRES_NEW)
+ protected ProcessDefinition wrapInTransaction(Supplier<ProcessDefinition>
supplier) {
+ return supplier.get();
+ }
+
}
diff --git
a/kogito-apps-quarkus/data-index-storage-quarkus/data-index-storage-jpa-quarkus/src/test/java/org/kie/kogito/index/jpa/quarkus/storage/PostgreSQLConcurrentProcessDefinitionStorageIT.java
b/kogito-apps-quarkus/data-index-storage-quarkus/data-index-storage-jpa-quarkus/src/test/java/org/kie/kogito/index/jpa/quarkus/storage/PostgreSQLConcurrentProcessDefinitionStorageIT.java
new file mode 100644
index 00000000000..a9fa5f7a939
--- /dev/null
+++
b/kogito-apps-quarkus/data-index-storage-quarkus/data-index-storage-jpa-quarkus/src/test/java/org/kie/kogito/index/jpa/quarkus/storage/PostgreSQLConcurrentProcessDefinitionStorageIT.java
@@ -0,0 +1,49 @@
+/*
+ * 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.jpa.quarkus.storage;
+
+import javax.sql.DataSource;
+
+import org.kie.kogito.index.jpa.quarkus.PostgreSQLQuarkusTestProfile;
+import
org.kie.kogito.index.jpa.storage.AbstractConcurrentProcessDefinitionStorageIT;
+import org.kie.kogito.index.jpa.storage.ProcessDefinitionEntityStorage;
+import org.kie.kogito.testcontainers.quarkus.PostgreSqlQuarkusTestResource;
+
+import io.quarkus.narayana.jta.QuarkusTransaction;
+import io.quarkus.test.common.QuarkusTestResource;
+import io.quarkus.test.junit.QuarkusTest;
+import io.quarkus.test.junit.TestProfile;
+
+import jakarta.inject.Inject;
+
+@QuarkusTest
+@QuarkusTestResource(value = PostgreSqlQuarkusTestResource.class,
restrictToAnnotatedClass = true)
+@TestProfile(PostgreSQLQuarkusTestProfile.class)
+class PostgreSQLConcurrentProcessDefinitionStorageIT extends
AbstractConcurrentProcessDefinitionStorageIT {
+
+ @Inject
+ public
PostgreSQLConcurrentProcessDefinitionStorageIT(ProcessDefinitionEntityStorage
storage, DataSource dataSource) {
+ super(storage, dataSource);
+ }
+
+ @Override
+ protected void executeInTransaction(Runnable operation) {
+ QuarkusTransaction.requiringNew().run(operation);
+ }
+}
diff --git
a/kogito-apps-quarkus/data-index-storage-quarkus/data-index-storage-postgresql/src/main/java/org/kie/kogito/index/postgresql/QuarkusProcessDefinitionEntityStorage.java
b/kogito-apps-quarkus/data-index-storage-quarkus/data-index-storage-postgresql/src/main/java/org/kie/kogito/index/postgresql/QuarkusProcessDefinitionEntityStorage.java
index af93150da40..27dc4e8b88a 100644
---
a/kogito-apps-quarkus/data-index-storage-quarkus/data-index-storage-postgresql/src/main/java/org/kie/kogito/index/postgresql/QuarkusProcessDefinitionEntityStorage.java
+++
b/kogito-apps-quarkus/data-index-storage-quarkus/data-index-storage-postgresql/src/main/java/org/kie/kogito/index/postgresql/QuarkusProcessDefinitionEntityStorage.java
@@ -19,16 +19,19 @@
package org.kie.kogito.index.postgresql;
import java.util.Collections;
+import java.util.function.Supplier;
import org.eclipse.microprofile.config.inject.ConfigProperty;
import org.kie.kogito.index.jpa.storage.JsonPredicateBuilder;
import org.kie.kogito.index.jpa.storage.ProcessDefinitionEntityStorage;
+import org.kie.kogito.index.model.ProcessDefinition;
import org.kie.kogito.process.Processes;
import jakarta.enterprise.context.ApplicationScoped;
import jakarta.enterprise.inject.Instance;
import jakarta.inject.Inject;
import jakarta.persistence.EntityManager;
+import jakarta.transaction.Transactional;
@ApplicationScoped
public class QuarkusProcessDefinitionEntityStorage extends
ProcessDefinitionEntityStorage {
@@ -39,4 +42,10 @@ public class QuarkusProcessDefinitionEntityStorage extends
ProcessDefinitionEnti
super(em, predicateBuilder, dataIsolationEnabled ? processesInstance :
Collections.emptyList());
}
+ @Override
+ @Transactional(Transactional.TxType.REQUIRES_NEW)
+ protected ProcessDefinition wrapInTransaction(Supplier<ProcessDefinition>
supplier) {
+ return supplier.get();
+ }
+
}
diff --git
a/kogito-apps-springboot/data-index-springboot/data-index-storage-jpa-springboot/src/main/java/org/kie/kogito/index/jpa/springboot/storage/DataIndexStorageProducer.java
b/kogito-apps-springboot/data-index-springboot/data-index-storage-jpa-springboot/src/main/java/org/kie/kogito/index/jpa/springboot/storage/DataIndexStorageProducer.java
index 18b51ef5bf3..537ca431ebd 100644
---
a/kogito-apps-springboot/data-index-springboot/data-index-storage-jpa-springboot/src/main/java/org/kie/kogito/index/jpa/springboot/storage/DataIndexStorageProducer.java
+++
b/kogito-apps-springboot/data-index-springboot/data-index-storage-jpa-springboot/src/main/java/org/kie/kogito/index/jpa/springboot/storage/DataIndexStorageProducer.java
@@ -23,7 +23,6 @@ import java.util.Collections;
import java.util.List;
import org.kie.kogito.index.api.DateTimeCoercing;
-import org.kie.kogito.index.jpa.mapper.ProcessDefinitionEntityMapper;
import org.kie.kogito.index.jpa.mapper.ProcessInstanceEntityMapper;
import org.kie.kogito.index.jpa.storage.*;
import org.kie.kogito.index.storage.DataIndexStorageService;
@@ -45,16 +44,6 @@ public class DataIndexStorageProducer {
return new JobEntityStorage(entityManager, dataIsolationEnabled ?
processes : Collections.emptyList());
}
- @Bean
- public ProcessDefinitionEntityStorage
processDefinitionEntityStorage(EntityManager entityManager,
- @Autowired(required = false) List<JsonPredicateBuilder>
jsonPredicateBuilders,
- @Autowired(required = false) List<Processes> processes) {
- return new ProcessDefinitionEntityStorage(entityManager,
- jsonPredicateBuilders != null ? jsonPredicateBuilders :
Collections.emptyList(),
- ProcessDefinitionEntityMapper.INSTANCE,
- dataIsolationEnabled ? processes : Collections.emptyList());
- }
-
@Bean
public ProcessInstanceEntityStorage
processInstanceEntityStorage(EntityManager entityManager,
@Autowired(required = false) List<JsonPredicateBuilder>
jsonPredicateBuilders,
diff --git
a/kogito-apps-springboot/data-index-springboot/data-index-storage-jpa-springboot/src/main/java/org/kie/kogito/index/jpa/springboot/storage/SpringBootProcessDefinitionEntityStorage.java
b/kogito-apps-springboot/data-index-springboot/data-index-storage-jpa-springboot/src/main/java/org/kie/kogito/index/jpa/springboot/storage/SpringBootProcessDefinitionEntityStorage.java
new file mode 100644
index 00000000000..75900d77d3f
--- /dev/null
+++
b/kogito-apps-springboot/data-index-springboot/data-index-storage-jpa-springboot/src/main/java/org/kie/kogito/index/jpa/springboot/storage/SpringBootProcessDefinitionEntityStorage.java
@@ -0,0 +1,58 @@
+/*
+ * 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.jpa.springboot.storage;
+
+import java.util.Collections;
+import java.util.function.Supplier;
+
+import org.kie.kogito.index.jpa.mapper.ProcessDefinitionEntityMapper;
+import org.kie.kogito.index.jpa.storage.JsonPredicateBuilder;
+import org.kie.kogito.index.jpa.storage.ProcessDefinitionEntityStorage;
+import org.kie.kogito.index.model.ProcessDefinition;
+import org.kie.kogito.process.Processes;
+import org.springframework.beans.factory.ObjectProvider;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.stereotype.Component;
+import org.springframework.transaction.PlatformTransactionManager;
+import org.springframework.transaction.TransactionDefinition;
+import org.springframework.transaction.support.TransactionTemplate;
+
+import jakarta.persistence.EntityManager;
+
+@Component
+public class SpringBootProcessDefinitionEntityStorage extends
ProcessDefinitionEntityStorage {
+
+ private final TransactionTemplate isolatedTransaction;
+
+ public SpringBootProcessDefinitionEntityStorage(EntityManager
entityManager, ObjectProvider<JsonPredicateBuilder> jsonPredicateBuilders,
+ ObjectProvider<Processes> processes, PlatformTransactionManager
transactionManager,
+ @Value("${kogito.persistence.data-isolation.enabled:false}")
Boolean dataIsolationEnabled) {
+ super(entityManager, jsonPredicateBuilders,
ProcessDefinitionEntityMapper.INSTANCE,
+ dataIsolationEnabled ? processes : Collections.emptyList());
+ this.isolatedTransaction = new TransactionTemplate(transactionManager);
+
this.isolatedTransaction.setPropagationBehavior(TransactionDefinition.PROPAGATION_REQUIRES_NEW);
+ }
+
+ @Override
+ protected ProcessDefinition wrapInTransaction(Supplier<ProcessDefinition>
supplier) {
+ return isolatedTransaction.execute(status -> supplier.get());
+ }
+
+}
diff --git
a/kogito-apps-springboot/data-index-springboot/data-index-storage-jpa-springboot/src/test/java/org/kie/kogito/index/jpa/springboot/storage/PostgreSQLConcurrentProcessDefinitionStorageIT.java
b/kogito-apps-springboot/data-index-springboot/data-index-storage-jpa-springboot/src/test/java/org/kie/kogito/index/jpa/springboot/storage/PostgreSQLConcurrentProcessDefinitionStorageIT.java
new file mode 100644
index 00000000000..14690c6956d
--- /dev/null
+++
b/kogito-apps-springboot/data-index-springboot/data-index-storage-jpa-springboot/src/test/java/org/kie/kogito/index/jpa/springboot/storage/PostgreSQLConcurrentProcessDefinitionStorageIT.java
@@ -0,0 +1,56 @@
+/*
+ * 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.jpa.springboot.storage;
+
+import javax.sql.DataSource;
+
+import org.kie.kogito.index.jpa.springboot.KogitoSpringBootApplication;
+import
org.kie.kogito.index.jpa.storage.AbstractConcurrentProcessDefinitionStorageIT;
+import org.kie.kogito.index.jpa.storage.ProcessDefinitionEntityStorage;
+import
org.kie.kogito.testcontainers.springboot.PostgreSqlSpringBootTestResource;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.test.context.SpringBootTest;
+import org.springframework.test.annotation.DirtiesContext;
+import org.springframework.test.context.ActiveProfiles;
+import org.springframework.test.context.ContextConfiguration;
+import org.springframework.transaction.PlatformTransactionManager;
+import org.springframework.transaction.TransactionDefinition;
+import org.springframework.transaction.support.TransactionTemplate;
+
+@SpringBootTest(webEnvironment = SpringBootTest.WebEnvironment.RANDOM_PORT,
classes = KogitoSpringBootApplication.class)
+@ContextConfiguration(initializers = PostgreSqlSpringBootTestResource.class)
+@ActiveProfiles("postgresql")
+@DirtiesContext(classMode = DirtiesContext.ClassMode.AFTER_EACH_TEST_METHOD)
+class PostgreSQLConcurrentProcessDefinitionStorageIT extends
AbstractConcurrentProcessDefinitionStorageIT {
+
+ private final TransactionTemplate transactionTemplate;
+
+ @Autowired
+ public
PostgreSQLConcurrentProcessDefinitionStorageIT(ProcessDefinitionEntityStorage
storage, DataSource dataSource,
+ PlatformTransactionManager transactionManager) {
+ super(storage, dataSource);
+ this.transactionTemplate = new TransactionTemplate(transactionManager);
+
this.transactionTemplate.setPropagationBehavior(TransactionDefinition.PROPAGATION_REQUIRES_NEW);
+ }
+
+ @Override
+ protected void executeInTransaction(Runnable operation) {
+ transactionTemplate.executeWithoutResult(status -> operation.run());
+ }
+}
diff --git
a/kogito-data-index/data-index-storage/data-index-storage-jpa-common/src/main/java/org/kie/kogito/index/jpa/storage/ProcessDefinitionEntityStorage.java
b/kogito-data-index/data-index-storage/data-index-storage-jpa-common/src/main/java/org/kie/kogito/index/jpa/storage/ProcessDefinitionEntityStorage.java
index 68e5de90c7a..bb8a2d5e952 100644
---
a/kogito-data-index/data-index-storage/data-index-storage-jpa-common/src/main/java/org/kie/kogito/index/jpa/storage/ProcessDefinitionEntityStorage.java
+++
b/kogito-data-index/data-index-storage/data-index-storage-jpa-common/src/main/java/org/kie/kogito/index/jpa/storage/ProcessDefinitionEntityStorage.java
@@ -19,18 +19,26 @@
package org.kie.kogito.index.jpa.storage;
import java.util.Optional;
+import java.util.function.Supplier;
+import org.hibernate.exception.ConstraintViolationException;
import org.kie.kogito.index.jpa.mapper.ProcessDefinitionEntityMapper;
import org.kie.kogito.index.jpa.model.ProcessDefinitionEntity;
import org.kie.kogito.index.model.ProcessDefinition;
import org.kie.kogito.index.model.ProcessDefinitionKey;
import org.kie.kogito.process.Processes;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.google.common.base.Throwables;
import jakarta.persistence.EntityManager;
import static org.kie.kogito.index.DependencyInjectionUtils.getInstance;
-public class ProcessDefinitionEntityStorage extends
AbstractStorage<ProcessDefinitionKey, ProcessDefinitionEntity,
ProcessDefinition> {
+public abstract class ProcessDefinitionEntityStorage extends
AbstractStorage<ProcessDefinitionKey, ProcessDefinitionEntity,
ProcessDefinition> {
+
+ private static final Logger LOGGER =
LoggerFactory.getLogger(ProcessDefinitionEntityStorage.class);
protected ProcessDefinitionEntityStorage() {
}
@@ -44,4 +52,24 @@ public class ProcessDefinitionEntityStorage extends
AbstractStorage<ProcessDefin
e.getVersion()),
Optional.ofNullable(getInstance(predicateBuilder)),
Optional.ofNullable(getInstance(processes)));
}
+ @Override
+ 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);
+
}
diff --git
a/kogito-data-index/data-index-storage/data-index-storage-jpa-common/src/test/java/org/kie/kogito/index/jpa/storage/AbstractConcurrentProcessDefinitionStorageIT.java
b/kogito-data-index/data-index-storage/data-index-storage-jpa-common/src/test/java/org/kie/kogito/index/jpa/storage/AbstractConcurrentProcessDefinitionStorageIT.java
new file mode 100644
index 00000000000..70b82a53a80
--- /dev/null
+++
b/kogito-data-index/data-index-storage/data-index-storage-jpa-common/src/test/java/org/kie/kogito/index/jpa/storage/AbstractConcurrentProcessDefinitionStorageIT.java
@@ -0,0 +1,129 @@
+/*
+ * 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.jpa.storage;
+
+import java.sql.Connection;
+import java.sql.PreparedStatement;
+import java.sql.ResultSet;
+import java.sql.SQLException;
+import java.util.Set;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicReference;
+
+import javax.sql.DataSource;
+
+import org.apache.commons.lang3.RandomStringUtils;
+import org.junit.jupiter.api.Test;
+import org.kie.kogito.index.model.ProcessDefinition;
+import org.kie.kogito.index.model.ProcessDefinitionKey;
+import org.kie.kogito.index.test.TestUtils;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatCode;
+
+/**
+ * A competing replica is simulated with a plain JDBC connection that inserts
the same definition and commits
+ * only once the insert of the storage is queued behind it on the primary key
index, which is what PostgreSQL
+ * does and what makes that insert fail with a duplicate key.
+ */
+public abstract class AbstractConcurrentProcessDefinitionStorageIT {
+
+ private static final String INSERT_DEFINITION = "insert into definitions
(id, version, name, type) values (?, ?, ?, ?)";
+ private static final String COUNT_PENDING_LOCKS = "select count(*) from
pg_locks where not granted and pid <> pg_backend_pid()";
+ private static final long LOCK_POLL_TIMEOUT_MILLIS = 15_000;
+ private static final long LOCK_POLL_INTERVAL_MILLIS = 100;
+
+ private final ProcessDefinitionEntityStorage storage;
+ private final DataSource dataSource;
+
+ protected
AbstractConcurrentProcessDefinitionStorageIT(ProcessDefinitionEntityStorage
storage, DataSource dataSource) {
+ this.storage = storage;
+ this.dataSource = dataSource;
+ }
+
+ protected abstract void executeInTransaction(Runnable operation);
+
+ @Test
+ void testConcurrentRegistrationOfTheSameProcessDefinition() throws
Exception {
+ String processId = RandomStringUtils.randomAlphabetic(10);
+ String version = "2.0";
+ ProcessDefinitionKey key = new ProcessDefinitionKey(processId,
version);
+ ProcessDefinition definition =
TestUtils.createProcessDefinition(processId, version, Set.of("admin",
"kogito"));
+
+ AtomicReference<ProcessDefinition> indexed = new AtomicReference<>();
+ AtomicBoolean indexingInsertBlocked = new AtomicBoolean();
+ ExecutorService executor = Executors.newSingleThreadExecutor();
+ CountDownLatch indexingStarted = new CountDownLatch(1);
+ try (Connection otherReplica = dataSource.getConnection()) {
+ otherReplica.setAutoCommit(false);
+ insertDefinition(otherReplica, processId, version);
+
+ Future<?> commitOfTheOtherReplica = executor.submit(() -> {
+ try {
+ indexingStarted.await();
+
indexingInsertBlocked.set(waitForTheIndexingInsertToBlock(otherReplica));
+ } finally {
+ otherReplica.commit();
+ }
+ return null;
+ });
+
+ indexingStarted.countDown();
+ assertThatCode(() -> executeInTransaction(() ->
indexed.set(storage.put(key, definition)))).doesNotThrowAnyException();
+
+ commitOfTheOtherReplica.get(1, TimeUnit.MINUTES);
+ } finally {
+ executor.shutdownNow();
+ }
+
+ assertThat(indexingInsertBlocked).as("the storage never queued its
insert behind the other replica, the race was not reproduced").isTrue();
+
assertThat(indexed.get()).isNotNull().extracting(ProcessDefinition::getId,
ProcessDefinition::getVersion, ProcessDefinition::getName)
+ .containsExactly(processId, version, processId);
+ assertThat(storage.get(key)).isEqualTo(indexed.get());
+ }
+
+ private void insertDefinition(Connection connection, String processId,
String version) throws SQLException {
+ try (PreparedStatement statement =
connection.prepareStatement(INSERT_DEFINITION)) {
+ statement.setString(1, processId);
+ statement.setString(2, version);
+ statement.setString(3, processId);
+ statement.setString(4, "PROCESS");
+ statement.executeUpdate();
+ }
+ }
+
+ private boolean waitForTheIndexingInsertToBlock(Connection connection)
throws Exception {
+ long deadline = System.currentTimeMillis() + LOCK_POLL_TIMEOUT_MILLIS;
+ while (System.currentTimeMillis() < deadline) {
+ Thread.sleep(LOCK_POLL_INTERVAL_MILLIS);
+ try (PreparedStatement statement =
connection.prepareStatement(COUNT_PENDING_LOCKS);
+ ResultSet resultSet = statement.executeQuery()) {
+ if (resultSet.next() && resultSet.getInt(1) > 0) {
+ return true;
+ }
+ }
+ }
+ return false;
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
