Skip to content

fix(sql): close non-transactional execution connections #960

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
bercianor merged 4 commits into from
Sep 22, 2026
Merged

fix(sql): close non-transactional execution connections #960

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
ad106c9
fix(sql): close non-transactional execution connections
bercianor Sep 18, 2026
2941d47
fix(sql): scope non-transactional connection cleanup
bercianor Sep 19, 2026
32e546b
Merge branch 'master' into fix/sql-connection-lifecycle
bercianor Sep 20, 2026
22b766f
refactor: nonTxWrapper
dieppa Sep 21, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
8 changes: 4 additions & 4 deletions .../flamingock-cloud/src/test/java/io/flamingock/core/cloud/utils/TestCloudTargetSystem.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,7 +20,7 @@
import io.flamingock.internal.core.external.targets.mark.TargetSystemAuditMark;
import io.flamingock.internal.core.external.targets.mark.TargetSystemAuditMarker;
import io.flamingock.internal.core.external.targets.TransactionalTargetSystem;
import io.flamingock.internal.common.core.transaction.TransactionWrapper;
import io.flamingock.internal.common.core.external.ExecutionWrapper;
import org.mockito.Mockito;

import java.util.Arrays;
Expand All @@ -45,7 +45,7 @@ public TargetSystemAuditMarker getAuditMarker() {
}

@Override
public TransactionWrapper getTxWrapper() {
public ExecutionWrapper getTxWrapper() {
return txWrapper;
}

Expand All @@ -60,10 +60,10 @@ protected TestCloudTargetSystem getSelf() {
}


public static class TestCloudTxWrapper implements TransactionWrapper {
public static class TestCloudTxWrapper implements ExecutionWrapper {

@Override
public <CONTEXT extends RuntimeContext, RESULT> RESULT wrapInTransaction(CONTEXT executionContext, Function<CONTEXT, RESULT> operation) {
public <CONTEXT extends RuntimeContext, RESULT> RESULT wrapExecution(CONTEXT executionContext, Function<CONTEXT, RESULT> operation) {
return operation.apply(executionContext);

}
Expand Down
6 changes: 3 additions & 3 deletions cloud/flamingock-cloud/src/test/java/io/flamingock/core/utils/EmptyTransactionWrapper.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -16,11 +16,11 @@
package io.flamingock.core.utils;

import io.flamingock.internal.common.core.context.RuntimeContext;
import io.flamingock.internal.common.core.transaction.TransactionWrapper;
import io.flamingock.internal.common.core.external.ExecutionWrapper;

import java.util.function.Function;

public class EmptyTransactionWrapper implements TransactionWrapper {
public class EmptyTransactionWrapper implements ExecutionWrapper {

private boolean called = false;

Expand All @@ -31,7 +31,7 @@ public boolean isCalled() {


@Override
public <CONTEXT extends RuntimeContext, RESULT> RESULT wrapInTransaction(CONTEXT executionContext, Function<CONTEXT, RESULT> operation) {
public <CONTEXT extends RuntimeContext, RESULT> RESULT wrapExecution(CONTEXT executionContext, Function<CONTEXT, RESULT> operation) {
called = true;
return operation.apply(executionContext);
}
Expand Down
9 changes: 4 additions & 5 deletions ...itstore/src/main/java/io/flamingock/store/dynamodb/internal/DynamoDBAuditPersistence.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,12 +19,11 @@
import io.flamingock.internal.common.core.context.RuntimeContext;
import io.flamingock.internal.common.core.feature.Features;
import io.flamingock.internal.common.core.journal.JournalEvent;
import io.flamingock.internal.common.core.transaction.TransactionWrapper;
import io.flamingock.internal.common.core.external.ExecutionWrapper;
import io.flamingock.internal.core.configuration.community.CommunityConfigurable;
import io.flamingock.internal.core.context.BasicRuntimeContext;
import io.flamingock.internal.core.external.store.audit.community.AbstractCommunityAuditPersistence;
import io.flamingock.internal.core.journal.JournalEventSequencer;
import io.flamingock.internal.core.journal.JournalEventSequencerFactory;
import io.flamingock.internal.util.FeatureFlag;
import io.flamingock.internal.util.Result;
import io.flamingock.internal.util.id.RunnerId;
Expand All @@ -37,7 +36,7 @@ public class DynamoDBAuditPersistence extends AbstractCommunityAuditPersistence
private final DynamoDBAuditRepository auditRepository;
private final DynamoDBJournalEventStore journalEventStore;
private JournalEventSequencer journalEventSequencer;
private final TransactionWrapper txWrapper;
private final ExecutionWrapper txWrapper;
private final boolean autoCreate;

/**
Expand All @@ -54,7 +53,7 @@ public DynamoDBAuditPersistence(CommunityConfigurable localConfiguration,
DynamoDBAuditRepository auditRepository,
DynamoDBJournalEventStore journalEventStore,
JournalEventSequencer journalEventSequencer,
TransactionWrapper txWrapper,
ExecutionWrapper txWrapper,
boolean autoCreate) {
super(localConfiguration);
this.auditRepository = auditRepository;
Expand All @@ -81,7 +80,7 @@ public List<AuditEntry> getAuditHistory() {
public Result writeEntry(AuditEntry auditEntry) {
if (isJournalEventsEnabled()) {
RuntimeContext baseContext = new BasicRuntimeContext("write-changeState-" + auditEntry.getChangeId());
Result result = txWrapper.wrapInTransaction(baseContext, runtimeContext -> {
Result result = txWrapper.wrapExecution(baseContext, runtimeContext -> {
TransactWriteItemsEnhancedRequest.Builder builder = runtimeContext.getContext()
.getRequiredDependencyValue(TransactWriteItemsEnhancedRequest.Builder.class);
JournalEvent<AuditEntry> journalEvent = journalEventSequencer.newEvent(auditEntry);
Expand Down
2 changes: 1 addition & 1 deletion ...tstore/src/test/java/io/flamingock/store/dynamodb/DynamoDBExternalSystemContractTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -16,7 +16,7 @@
package io.flamingock.store.dynamodb;

import io.flamingock.externalsystem.dynamodb.api.DynamoDBExternalSystem;
import io.flamingock.internal.common.core.transaction.TransactionalExternalSystem;
import io.flamingock.internal.common.core.external.TransactionalExternalSystem;
import org.junit.jupiter.api.DisplayName;
import org.junit.jupiter.api.Test;

Expand Down
2 changes: 1 addition & 1 deletion ...re/src/test/java/io/flamingock/store/dynamodb/internal/DynamoDBJournalEventStoreTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -503,7 +503,7 @@ private void commit(List<JournalEvent<AuditEntry>> events) {
DynamoDBTxWrapper txWrapper = new DynamoDBTxWrapper(
client,
new TransactionManager<>(TransactWriteItemsEnhancedRequest::builder));
txWrapper.wrapInTransaction(new BasicRuntimeContext("session-" + UUID.randomUUID()), ctx -> {
txWrapper.wrapExecution(new BasicRuntimeContext("session-" + UUID.randomUUID()), ctx -> {
TransactWriteItemsEnhancedRequest.Builder builder = ctx.getContext()
.getRequiredDependencyValue(TransactWriteItemsEnhancedRequest.Builder.class);
for (JournalEvent<AuditEntry> event : events) {
Expand Down
4 changes: 2 additions & 2 deletions ...itstore/src/main/java/io/flamingock/store/mongodb/reactive/MongoDBReactiveAuditStore.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -41,7 +41,7 @@
import io.flamingock.store.mongodb.reactive.internal.MongoDBReactiveJournalEventStore;
import io.flamingock.store.mongodb.reactive.internal.MongoDBReactiveLockService;

import io.flamingock.internal.common.core.transaction.TransactionWrapper;
import io.flamingock.internal.common.core.external.ExecutionWrapper;

import java.util.Collections;
import java.util.HashSet;
Expand Down Expand Up @@ -162,7 +162,7 @@ public AuditPersistenceFactory<CommunityAuditPersistence> getPersistenceFactory(
}
JournalEventSequencer journalEventSequencer = journalEventSequencerFactory.forStream(stageId);
boolean supportsTransactions = mongoDBTargetSystem.supportsTransactions();
TransactionWrapper txWrapper = supportsTransactions ? mongoDBTargetSystem.getTxWrapper() : null;
ExecutionWrapper txWrapper = supportsTransactions ? mongoDBTargetSystem.getTxWrapper() : null;
MongoDBReactiveAuditPersistence stagePersistence = new MongoDBReactiveAuditPersistence(
communityConfiguration,
auditRepository,
Expand Down
12 changes: 6 additions & 6 deletions ...n/java/io/flamingock/store/mongodb/reactive/internal/MongoDBReactiveAuditPersistence.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,7 +20,7 @@
import io.flamingock.internal.common.core.context.RuntimeContext;
import io.flamingock.internal.common.core.feature.Features;
import io.flamingock.internal.common.core.journal.JournalEvent;
import io.flamingock.internal.common.core.transaction.TransactionWrapper;
import io.flamingock.internal.common.core.external.ExecutionWrapper;
import io.flamingock.internal.core.context.BasicRuntimeContext;
import io.flamingock.internal.core.configuration.community.CommunityConfigurable;
import io.flamingock.internal.core.external.store.audit.community.AbstractCommunityAuditPersistence;
Expand All @@ -38,7 +38,7 @@ public class MongoDBReactiveAuditPersistence extends AbstractCommunityAuditPersi
private final MongoDBReactiveJournalEventStore journalEventStore;
private final JournalEventSequencer journalEventSequencer;
private final boolean supportsTransactions;
private final TransactionWrapper txWrapper;
private final ExecutionWrapper txWrapper;
private final boolean autoCreate;

/**
Expand All @@ -55,7 +55,7 @@ public MongoDBReactiveAuditPersistence(CommunityConfigurable localConfiguration,
MongoDBReactiveJournalEventStore journalEventStore,
JournalEventSequencer journalEventSequencer,
boolean supportsTransactions,
TransactionWrapper txWrapper,
ExecutionWrapper txWrapper,
boolean autoCreate) {
super(localConfiguration);
this.auditRepository = auditRepository;
Expand All @@ -66,8 +66,8 @@ public MongoDBReactiveAuditPersistence(CommunityConfigurable localConfiguration,
this.autoCreate = autoCreate;
}

private static TransactionWrapper validateTransactionWrapper(boolean supportsTransactions,
TransactionWrapper txWrapper) {
private static ExecutionWrapper validateTransactionWrapper(boolean supportsTransactions,
ExecutionWrapper txWrapper) {
if (supportsTransactions) {
return Objects.requireNonNull(
txWrapper,
Expand Down Expand Up @@ -112,7 +112,7 @@ public Result writeEntry(AuditEntry auditEntry) {

private Result writeJournalAndAuditInTransaction(AuditEntry auditEntry) {
RuntimeContext baseContext = new BasicRuntimeContext("write-changeState-" + auditEntry.getChangeId());
Result result = txWrapper.wrapInTransaction(baseContext, runtimeContext -> {
Result result = txWrapper.wrapExecution(baseContext, runtimeContext -> {
ClientSession clientSession = runtimeContext.getContext().getRequiredDependencyValue(ClientSession.class);
JournalEvent<AuditEntry> journalEvent = journalEventSequencer.newEvent(auditEntry);
journalEventStore.append(clientSession, journalEvent);
Expand Down
4 changes: 2 additions & 2 deletions ...va/io/flamingock/store/mongodb/reactive/internal/MongoDBReactiveAuditPersistenceTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -27,7 +27,7 @@
import io.flamingock.internal.common.core.feature.Features;
import io.flamingock.internal.core.configuration.community.CommunityConfigurable;
import io.flamingock.internal.core.journal.JournalEventSequencer;
import io.flamingock.internal.common.core.transaction.TransactionWrapper;
import io.flamingock.internal.common.core.external.ExecutionWrapper;
import io.flamingock.internal.util.FeatureFlag;
import io.flamingock.internal.util.id.RunnerId;
import io.flamingock.reactive.util.PublisherSync;
Expand Down Expand Up @@ -79,7 +79,7 @@ void beforeEach() {
journalEventStore,
mock(JournalEventSequencer.class),
true,
mock(TransactionWrapper.class),
mock(ExecutionWrapper.class),
true);
persistence.initialize(RunnerId.fromString("runner-1"));
}
Expand Down
6 changes: 2 additions & 4 deletions ...sync-auditstore/src/main/java/io/flamingock/store/mongodb/sync/MongoDBSyncAuditStore.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,13 +20,12 @@
import com.mongodb.WriteConcern;
import com.mongodb.client.ClientSession;
import com.mongodb.client.MongoDatabase;
import io.flamingock.internal.common.core.audit.AuditEntry;
import io.flamingock.internal.common.core.audit.AuditPersistenceFactory;
import io.flamingock.internal.common.core.audit.AuditReader;
import io.flamingock.internal.common.core.context.ContextResolver;
import io.flamingock.internal.common.core.error.FlamingockException;
import io.flamingock.internal.common.core.feature.Features;
import io.flamingock.internal.common.core.transaction.TransactionWrapper;
import io.flamingock.internal.common.core.external.ExecutionWrapper;
import io.flamingock.internal.core.configuration.community.CommunityConfigurable;
import io.flamingock.internal.core.external.store.CommunityAuditStore;
import io.flamingock.internal.core.external.store.audit.community.CommunityAuditPersistence;
Expand All @@ -46,7 +45,6 @@

import java.util.Collections;
import java.util.HashSet;
import java.util.List;
import java.util.Set;

import org.slf4j.Logger;
Expand Down Expand Up @@ -163,7 +161,7 @@ public AuditPersistenceFactory<CommunityAuditPersistence> getPersistenceFactory(
return stageId -> {
JournalEventSequencer journalEventSequencer = journalEventSequencerFactory.forStream(stageId);
boolean supportsTransactions = mongoDBTargetSystem.supportsTransactions();
TransactionWrapper txWrapper = supportsTransactions ? mongoDBTargetSystem.getTxWrapper() : null;
ExecutionWrapper txWrapper = supportsTransactions ? mongoDBTargetSystem.getTxWrapper() : null;
persistence = new MongoDBSyncAuditPersistence(
communityConfiguration,
auditRepository,
Expand Down
14 changes: 7 additions & 7 deletions .../src/main/java/io/flamingock/store/mongodb/sync/internal/MongoDBSyncAuditPersistence.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,7 +20,7 @@
import io.flamingock.internal.common.core.context.RuntimeContext;
import io.flamingock.internal.common.core.feature.Features;
import io.flamingock.internal.common.core.journal.JournalEvent;
import io.flamingock.internal.common.core.transaction.TransactionWrapper;
import io.flamingock.internal.common.core.external.ExecutionWrapper;
import io.flamingock.internal.core.configuration.community.CommunityConfigurable;
import io.flamingock.internal.core.context.BasicRuntimeContext;
import io.flamingock.internal.core.external.store.audit.community.AbstractCommunityAuditPersistence;
Expand All @@ -38,7 +38,7 @@ public class MongoDBSyncAuditPersistence extends AbstractCommunityAuditPersisten
private final MongoDBSyncJournalEventStore journalEventStore;
private final JournalEventSequencer journalEventSequencer;
private final boolean supportsTransactions;
private final TransactionWrapper txWrapper;
private final ExecutionWrapper txWrapper;
private final boolean autoCreate;

/**
Expand All @@ -55,7 +55,7 @@ public MongoDBSyncAuditPersistence(CommunityConfigurable localConfiguration,
MongoDBSyncJournalEventStore journalEventStore,
JournalEventSequencer journalEventSequencer,
boolean supportsTransactions,
TransactionWrapper txWrapper,
ExecutionWrapper txWrapper,
boolean autoCreate) {
super(localConfiguration);
this.auditRepository = auditRepository;
Expand All @@ -66,8 +66,8 @@ public MongoDBSyncAuditPersistence(CommunityConfigurable localConfiguration,
this.autoCreate = autoCreate;
}

private static TransactionWrapper validateTransactionWrapper(boolean supportsTransactions,
TransactionWrapper txWrapper) {
private static ExecutionWrapper validateTransactionWrapper(boolean supportsTransactions,
ExecutionWrapper txWrapper) {
if (supportsTransactions) {
return Objects.requireNonNull(
txWrapper,
Expand Down Expand Up @@ -113,14 +113,14 @@ public Result writeEntry(AuditEntry auditEntry) {

private Result writeJournalAndAuditInTransaction(AuditEntry auditEntry) {
RuntimeContext baseContext = new BasicRuntimeContext("write-changeState-" + auditEntry.getChangeId());
Result result = txWrapper.wrapInTransaction(baseContext, runtimeContext -> {
Result result = txWrapper.wrapExecution(baseContext, runtimeContext -> {
ClientSession clientSession = runtimeContext.getContext().getRequiredDependencyValue(ClientSession.class);
JournalEvent<AuditEntry> journalEvent = journalEventSequencer.newEvent(auditEntry);
journalEventStore.write(clientSession, journalEvent);
return auditRepository.save(clientSession, auditEntry);
});
// Spends the stream position, and only a committed transaction may reach this line. In general a
// normal return from wrapInTransaction does NOT mean commit — a FailedStep result is returned
// normal return from wrapExecution does NOT mean commit — a FailedStep result is returned
// after a rollback, without an exception. It is sound here because this operation returns a
// Result, which can never be a FailedStep, so the commit branch is the only graceful path; a
// failing commit is caught and rethrown as DatabaseTransactionException. Keep that true: an
Expand Down
8 changes: 4 additions & 4 deletions ...ck-sql-auditstore/src/main/java/io/flamingock/store/sql/internal/SqlAuditPersistence.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,7 +18,7 @@
import io.flamingock.internal.common.core.audit.AuditEntry;
import io.flamingock.internal.common.core.context.RuntimeContext;
import io.flamingock.internal.common.core.journal.JournalEvent;
import io.flamingock.internal.common.core.transaction.TransactionWrapper;
import io.flamingock.internal.common.core.external.ExecutionWrapper;
import io.flamingock.internal.core.configuration.community.CommunityConfigurable;
import io.flamingock.internal.core.context.BasicRuntimeContext;
import io.flamingock.internal.core.external.store.audit.community.AbstractCommunityAuditPersistence;
Expand All @@ -34,7 +34,7 @@ public class SqlAuditPersistence extends AbstractCommunityAuditPersistence {
private final SqlAuditRepository auditRepository;
private final SqlJournalEventStore journalEventStore;
private final JournalEventSequencer journalEventSequencer;
private final TransactionWrapper txWrapper;
private final ExecutionWrapper txWrapper;
private final boolean journalEventsEnabled;

/**
Expand All @@ -51,7 +51,7 @@ public SqlAuditPersistence(CommunityConfigurable localConfiguration,
SqlAuditRepository auditRepository,
SqlJournalEventStore journalEventStore,
JournalEventSequencer journalEventSequencer,
TransactionWrapper txWrapper,
ExecutionWrapper txWrapper,
boolean journalEventsEnabled) {
super(localConfiguration);
this.auditRepository = auditRepository;
Expand Down Expand Up @@ -86,7 +86,7 @@ public synchronized Result writeEntry(AuditEntry auditEntry) {
}

RuntimeContext baseContext = new BasicRuntimeContext("write-changeState-" + auditEntry.getChangeId());
Result result = txWrapper.wrapInTransaction(baseContext, runtimeContext -> {
Result result = txWrapper.wrapExecution(baseContext, runtimeContext -> {
Connection connection = runtimeContext.getContext().getRequiredDependencyValue(Connection.class);
JournalEvent<AuditEntry> journalEvent = journalEventSequencer.newEvent(auditEntry);
journalEventStore.append(connection, journalEvent);
Expand Down
Loading
Loading