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..be196eabb 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 maxInParameters; 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); + maxInParameters = env.getProperty("nflow.db.max_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)); @@ -407,12 +409,18 @@ public void recoverWorkflowInstancesFromDeadNodes() { } private List getRecoverableWorkflowInstances(Collection executorsIds) { - StringBuilder sql = new StringBuilder(128); - sql.append("select id, executor_id, state from nflow_workflow where executor_id in ("); - executorsIds.forEach(id -> sql.append("?,")); - sql.setCharAt(sql.length() - 1, ')'); - return jdbc.query(sql.toString(), (rs, rowNum) -> new InstanceInfo(rs.getLong(1), rs.getInt(2), rs.getString(3)), - (Object[]) executorsIds.toArray(new Integer[0])); + List ids = new ArrayList<>(executorsIds); + List result = new ArrayList<>(); + for (int i = 0; i < ids.size(); i += maxInParameters) { + List batch = ids.subList(i, min(i + maxInParameters, ids.size())); + StringBuilder sql = new StringBuilder(64); + sql.append("select id, executor_id, state from nflow_workflow where executor_id in ("); + batch.forEach(id -> sql.append("?,")); + sql.setCharAt(sql.length() - 1, ')'); + result.addAll(jdbc.query(sql.toString(), (rs, rowNum) -> new InstanceInfo(rs.getLong(1), rs.getInt(2), rs.getString(3)), + (Object[]) batch.toArray(new Integer[0]))); + } + return result; } private void recoverWorkflowInstance(final long instanceId, int expectedExecutorId, final WorkflowInstanceAction action) { diff --git a/nflow-engine/src/main/resources/nflow-engine.properties b/nflow-engine/src/main/resources/nflow-engine.properties index 0599223a7..697be3540 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_in_parameters=1000 nflow.db.workflowInstanceType.cacheSize=10000 nflow.db.initialization_fail_timeout_seconds=10 diff --git a/nflow-tests/src/test/java/io/nflow/tests/dao/BulkRecoveryDaoTest.java b/nflow-tests/src/test/java/io/nflow/tests/dao/BulkRecoveryDaoTest.java new file mode 100644 index 000000000..34b7ea97f --- /dev/null +++ b/nflow-tests/src/test/java/io/nflow/tests/dao/BulkRecoveryDaoTest.java @@ -0,0 +1,99 @@ +package io.nflow.tests.dao; + +import static io.nflow.engine.workflow.instance.WorkflowInstance.WorkflowInstanceStatus.executing; +import static org.hamcrest.MatcherAssert.assertThat; +import static org.hamcrest.Matchers.is; +import static org.joda.time.DateTime.now; + +import java.sql.Timestamp; +import java.util.ArrayList; +import java.util.List; + +import jakarta.inject.Inject; + +import org.junit.jupiter.api.MethodOrderer; +import org.junit.jupiter.api.Order; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.TestMethodOrder; +import org.springframework.context.annotation.Configuration; +import org.springframework.jdbc.core.JdbcTemplate; + +import io.nflow.engine.config.NFlow; +import io.nflow.engine.internal.dao.ExecutorDao; +import io.nflow.engine.internal.dao.WorkflowInstanceDao; +import io.nflow.tests.AbstractNflowTest; +import io.nflow.tests.extension.NflowServerConfig; +import io.nflow.tests.extension.NflowServerExtension.BeforeServerStop; + +@TestMethodOrder(MethodOrderer.OrderAnnotation.class) +public class BulkRecoveryDaoTest extends AbstractNflowTest { + + public static NflowServerConfig server = new NflowServerConfig.Builder() + .springContextClass(ServerContext.class) + .build(); + + private static JdbcTemplate jdbc; + private static WorkflowInstanceDao workflowInstanceDao; + private static ExecutorDao executorDao; + + public BulkRecoveryDaoTest() { + super(server); + } + + @Configuration + static class ServerContext { + @Inject + public void init(@NFlow JdbcTemplate nflowJdbc, WorkflowInstanceDao dao, ExecutorDao executor) { + jdbc = nflowJdbc; + workflowInstanceDao = dao; + executorDao = executor; + } + } + + @Test + @Order(1) + public void insertNineThousandDeadExecutorsAndOneWorkflow() { + int count = 9000; + Timestamp crash = new Timestamp(now().minusDays(1).getMillis()); + Timestamp active = new Timestamp(now().minusDays(1).plusSeconds(1).getMillis()); + Timestamp expires = new Timestamp(now().minusDays(1).plusHours(1).getMillis()); + List args = new ArrayList<>(count); + for (int i = 0; i < count; i++) { + args.add(new Object[] { "localhost", 666 + i, executorDao.getExecutorGroup(), crash, active, expires }); + } + jdbc.batchUpdate( + "insert into nflow_executor (host, pid, executor_group, started, active, expires) values (?, ?, ?, ?, ?, ?)", + args); + + int firstCrashedExecutorId = jdbc.queryForObject( + "select min(id) from nflow_executor where executor_group = ? and expires < current_timestamp and recovered is null", + Integer.class, executorDao.getExecutorGroup()); + jdbc.update( + "insert into nflow_workflow (status, type, external_id, state, executor_id, executor_group, priority) values (?, ?, ?, ?, ?, ?, 0)", + executing.name(), "bulkTest", "extId0", "processing", firstCrashedExecutorId, executorDao.getExecutorGroup()); + } + + @Test + @Order(2) + public void recoverWorkflowInstancesFromNineThousandDeadExecutors() { + workflowInstanceDao.recoverWorkflowInstancesFromDeadNodes(); + + int recoveredExecutors = jdbc.queryForObject( + "select count(*) from nflow_executor where executor_group = ? and recovered is not null", + Integer.class, executorDao.getExecutorGroup()); + assertThat(recoveredExecutors, is(9000)); + + Integer workflowExecutorId = jdbc.queryForObject( + "select executor_id from nflow_workflow where executor_group = ? and type = ?", + Integer.class, executorDao.getExecutorGroup(), "bulkTest"); + assertThat(workflowExecutorId, is((Integer) null)); + } + + @BeforeServerStop + public void cleanUp() { + String group = executorDao.getExecutorGroup(); + jdbc.update("delete from nflow_workflow_action where workflow_id in (select id from nflow_workflow where executor_group = ? and type = ?)", group, "bulkTest"); + jdbc.update("delete from nflow_workflow where executor_group = ? and type = ?", group, "bulkTest"); + jdbc.update("delete from nflow_executor where executor_group = ? and expires < current_timestamp", group); + } +} diff --git a/pom.xml b/pom.xml index 0bcc37740..b00d6ae57 100644 --- a/pom.xml +++ b/pom.xml @@ -109,7 +109,7 @@ 33.6.0-jre 1C - 9.10 + 9.10.1 4.2.1 3.6.1 0.10.0 @@ -124,10 +124,10 @@ 3.0 9.1.0.Final 7.0.2 - 2.21 - 3.1.3 - 3.1.3 - 3.1.3 + 2.22 + 3.1.4 + 3.1.4 + 3.1.4 3.31.0-GA 3.0.0 6.1.0 @@ -136,10 +136,10 @@ 4.1.0-M1 4.0.3 3.1.11 - 12.1.9 + 12.1.10 2.14.2 6.1.0 - 1.5.32 + 1.5.34 3.5.8 3.8.0 3.5.0 @@ -161,8 +161,8 @@ 3.6.2 4.0.0-M16 3.4.0 - 3.5.5 - 4.2.38 + 3.5.6 + 4.2.39 5.23.0 13.4.0.jre11 9.7.0