diff --git a/CHANGELOG.md b/CHANGELOG.md index 41afc2dee..20551e5c3 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -3,9 +3,14 @@ **Highlights** - Use Spring Boot 4 in `nflow-examples` +- Recover workflow instances from dead executors in configurable chunks to avoid too large SQL `IN` clauses **Details** +- `nflow-engine` + - Recover workflow instances from dead executors in chunks controlled by `nflow.db.max_sql_in_parameters`. + This avoids oversized SQL `IN` clauses, especially on SQL Server. + ## 11.0.0 (2026-05-28) **Highlights** diff --git a/nflow-engine/src/main/java/io/nflow/engine/internal/dao/WorkflowInstanceDao.java b/nflow-engine/src/main/java/io/nflow/engine/internal/dao/WorkflowInstanceDao.java index e1631e475..0f1d05b1a 100644 --- a/nflow-engine/src/main/java/io/nflow/engine/internal/dao/WorkflowInstanceDao.java +++ b/nflow-engine/src/main/java/io/nflow/engine/internal/dao/WorkflowInstanceDao.java @@ -111,6 +111,7 @@ public class WorkflowInstanceDao { private final long workflowInstanceQueryMaxActions; private final long workflowInstanceQueryMaxActionsDefault; private final int workflowInstanceTypeCacheSize; + private final int maxSqlInParameters; private final AtomicBoolean disableBatchUpdates = new AtomicBoolean(); AtomicInteger instanceStateTextLength = new AtomicInteger(); AtomicInteger actionStateTextLength = new AtomicInteger(); @@ -145,6 +146,7 @@ public WorkflowInstanceDao(SQLVariants sqlVariants, @NFlow JdbcTemplate nflowJdb logger.info("nFlow DB batch updates are disabled (system property nflow.db.disable_batch_updates=true)"); } workflowInstanceTypeCacheSize = env.getRequiredProperty("nflow.db.workflowInstanceType.cacheSize", Integer.class); + maxSqlInParameters = env.getProperty("nflow.db.max_sql_in_parameters", Integer.class, 1000); instanceStateTextLength.set(env.getProperty("nflow.workflow.instance.state.text.length", Integer.class, -1)); actionStateTextLength.set(env.getProperty("nflow.workflow.action.state.text.length", Integer.class, -1)); stateVariableValueMaxLength.set(env.getProperty("nflow.workflow.state.variable.value.length", Integer.class, -1)); @@ -399,11 +401,16 @@ public void recoverWorkflowInstancesFromDeadNodes() { } WorkflowInstanceAction.Builder builder = new WorkflowInstanceAction.Builder().setExecutionStart(now()).setExecutionEnd(now()) .setType(recovery).setStateText("Recovered"); - for (InstanceInfo instance : getRecoverableWorkflowInstances(recoverableExecutorIds)) { - WorkflowInstanceAction action = builder.setState(instance.state()).setWorkflowInstanceId(instance.id()).build(); - recoverWorkflowInstance(instance.id(), instance.executorId(), action); + List recoverableExecutorIdList = new ArrayList<>(recoverableExecutorIds); + for (int from = 0; from < recoverableExecutorIdList.size(); from += maxSqlInParameters) { + List recoverableExecutorIdChunk = recoverableExecutorIdList.subList(from, + min(from + maxSqlInParameters, recoverableExecutorIdList.size())); + for (InstanceInfo instance : getRecoverableWorkflowInstances(recoverableExecutorIdChunk)) { + WorkflowInstanceAction action = builder.setState(instance.state()).setWorkflowInstanceId(instance.id()).build(); + recoverWorkflowInstance(instance.id(), instance.executorId(), action); + } + recoverableExecutorIdChunk.forEach(executorInfo::markRecovered); } - recoverableExecutorIds.forEach(executorInfo::markRecovered); } private List getRecoverableWorkflowInstances(Collection executorsIds) { diff --git a/nflow-engine/src/main/resources/nflow-engine.properties b/nflow-engine/src/main/resources/nflow-engine.properties index 0599223a7..6791a2276 100644 --- a/nflow-engine/src/main/resources/nflow-engine.properties +++ b/nflow-engine/src/main/resources/nflow-engine.properties @@ -65,6 +65,7 @@ nflow.db.max_pool_size=4 nflow.db.idle_timeout_seconds=600 nflow.db.create_on_startup=true nflow.db.disable_batch_updates=false +nflow.db.max_sql_in_parameters=1000 nflow.db.workflowInstanceType.cacheSize=10000 nflow.db.initialization_fail_timeout_seconds=10 diff --git a/nflow-engine/src/test/java/io/nflow/engine/internal/dao/WorkflowInstanceDaoTest.java b/nflow-engine/src/test/java/io/nflow/engine/internal/dao/WorkflowInstanceDaoTest.java index 0efe26726..1051040fe 100644 --- a/nflow-engine/src/test/java/io/nflow/engine/internal/dao/WorkflowInstanceDaoTest.java +++ b/nflow-engine/src/test/java/io/nflow/engine/internal/dao/WorkflowInstanceDaoTest.java @@ -33,9 +33,13 @@ import static org.junit.jupiter.api.Assertions.assertThrows; import static org.junit.jupiter.api.Assertions.assertTrue; import static org.junit.jupiter.api.Assertions.fail; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyString; import static org.mockito.ArgumentMatchers.eq; import static org.mockito.Mockito.lenient; import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.times; +import static org.mockito.Mockito.verify; import static org.mockito.Mockito.when; import java.sql.ResultSet; @@ -53,17 +57,21 @@ import org.joda.time.DateTime; import org.junit.jupiter.api.Test; import org.mockito.ArgumentCaptor; +import org.mockito.ArgumentMatchers; import org.springframework.core.env.Environment; import org.springframework.dao.EmptyResultDataAccessException; import org.springframework.jdbc.core.JdbcTemplate; import org.springframework.jdbc.core.RowCallbackHandler; +import org.springframework.jdbc.core.RowMapper; import org.springframework.jdbc.core.namedparam.NamedParameterJdbcTemplate; +import org.springframework.mock.env.MockEnvironment; import org.springframework.transaction.TransactionStatus; import org.springframework.transaction.support.TransactionCallback; import org.springframework.transaction.support.TransactionTemplate; import io.nflow.engine.config.db.PgDatabaseConfiguration.PostgreSQLVariants; import io.nflow.engine.internal.dao.WorkflowInstanceDao.WorkflowInstanceActionRowMapper; +import io.nflow.engine.internal.executor.InstanceInfo; import io.nflow.engine.internal.executor.WorkflowInstanceExecutor; import io.nflow.engine.internal.storage.db.SQLVariants; import io.nflow.engine.service.WorkflowInstanceInclude; @@ -748,7 +756,10 @@ public void fakePostgreSQLpollNextWorkflowInstances() { } private WorkflowInstanceDao preparePostgreSQLDao(JdbcTemplate jdbcTemplate) { - ExecutorDao eDao = mock(ExecutorDao.class); + return preparePostgreSQLDao(jdbcTemplate, mock(ExecutorDao.class), env); + } + + private WorkflowInstanceDao preparePostgreSQLDao(JdbcTemplate jdbcTemplate, ExecutorDao eDao, Environment env) { lenient().when(eDao.getExecutorGroupCondition()).thenReturn("group matches"); lenient().when(eDao.getExecutorId()).thenReturn(42); NamedParameterJdbcTemplate namedJdbc = mock(NamedParameterJdbcTemplate.class); @@ -762,6 +773,16 @@ private WorkflowInstanceDao preparePostgreSQLDao(JdbcTemplate jdbcTemplate) { return d; } + private MockEnvironment prepareEnvironment() { + return new MockEnvironment() + .withProperty("nflow.workflow.instance.query.max.results", "1000") + .withProperty("nflow.workflow.instance.query.max.results.default", "1000") + .withProperty("nflow.workflow.instance.query.max.actions", "1000") + .withProperty("nflow.workflow.instance.query.max.actions.default", "1000") + .withProperty("nflow.db.disable_batch_updates", "false") + .withProperty("nflow.db.workflowInstanceType.cacheSize", "10000"); + } + @Test public void pollNextWorkflowInstancesWithPartialRaceCondition() throws InterruptedException { int batchSize = 100; @@ -982,6 +1003,33 @@ public void recoverWorkflowInstancesFromDeadNodesSetsExecutorIdToNullAndStatusTo assertThat(workflowInstanceAction.stateText, is("Recovered")); } + @Test + public void recoverWorkflowInstancesFromDeadNodesSplitsExecutorIdsByConfiguredMaxSqlInParameters() { + JdbcTemplate j = mock(JdbcTemplate.class); + ExecutorDao eDao = mock(ExecutorDao.class); + when(eDao.getRecoverableExecutorIds()).thenReturn(asList(1, 2, 3, 4, 5)); + when(j.query(anyString(), ArgumentMatchers.> any(), any(Object[].class))).thenReturn(emptyList()); + WorkflowInstanceDao d = preparePostgreSQLDao(j, eDao, prepareEnvironment().withProperty("nflow.db.max_sql_in_parameters", "2")); + + d.recoverWorkflowInstancesFromDeadNodes(); + + ArgumentCaptor sql = ArgumentCaptor.forClass(String.class); + ArgumentCaptor args = ArgumentCaptor.forClass(Object[].class); + verify(j, times(3)).query(sql.capture(), ArgumentMatchers.> any(), args.capture()); + assertThat(sql.getAllValues(), contains( + "select id, executor_id, state from nflow_workflow where executor_id in (?,?)", + "select id, executor_id, state from nflow_workflow where executor_id in (?,?)", + "select id, executor_id, state from nflow_workflow where executor_id in (?)")); + assertThat(asList(args.getAllValues().get(0)), contains((Object) 1, (Object) 2)); + assertThat(asList(args.getAllValues().get(1)), contains((Object) 3, (Object) 4)); + assertThat(asList(args.getAllValues().get(2)), contains((Object) 5)); + verify(eDao).markRecovered(1); + verify(eDao).markRecovered(2); + verify(eDao).markRecovered(3); + verify(eDao).markRecovered(4); + verify(eDao).markRecovered(5); + } + @Test public void settingSignalInsertsAction() { WorkflowInstance i = constructWorkflowInstanceBuilder().setBusinessKey("setSignalTest").build();