[kie-roadmap#52] Improve LogCleanupCommand, do not delete RequestInfo and ErrorInfo records of active process instances

jbpm-audit/src/main/java/org/jbpm/process/audit/query/AbstractAuditDeleteBuilderImpl.java

@@ -50,14 +50,20 @@ protected static class Subquery {
         private String queryBase;
         private int queryParamId;
         private QueryWhere where;
+        private String join;
 
         private QueryAndParameterAppender queryAndParameterAppender;
 
         public Subquery(String field, String queryBase, int queryParamId) {
+            this(field, queryBase, queryParamId, "in");
+        }
+
+        public Subquery(String field, String queryBase, int queryParamId, String join) {
             this.field = field;
             this.queryBase = queryBase;
             this.queryParamId = queryParamId;
             this.where = new QueryWhere();
+            this.join = join;
         }
 
         public Subquery parameter(String listId, Object... values) {
@@ -79,7 +85,7 @@ public String build() {
             if (queryAndParameterAppender == null) {
                 queryAndParameterAppender = QueryHelper.createQuery(queryBase, where, new HashMap<>(), queryParamId);
             }
-            return field + " in (" + queryAndParameterAppender.toSQL() + ")";
+            return field + " " + join + " (" + queryAndParameterAppender.toSQL() + ")";
         }
 
     }

jbpm-services/jbpm-executor-cdi/src/test/filtered-resources/META-INF/persistence.xml

@@ -10,6 +10,7 @@
     <mapping-file>META-INF/Executor-orm.xml</mapping-file>
     <class>org.jbpm.executor.entities.ErrorInfo</class>
     <class>org.jbpm.executor.entities.RequestInfo</class>
+    <class>org.jbpm.process.audit.ProcessInstanceLog</class>
     <class>org.jbpm.runtime.manager.impl.jpa.ExecutionErrorInfo</class>
     <exclude-unlisted-classes>true</exclude-unlisted-classes>
 

jbpm-services/jbpm-executor/src/main/java/org/jbpm/executor/impl/jpa/ErrorInfoDeleteBuilderImpl.java

@@ -22,6 +22,7 @@
 import org.jbpm.process.audit.JPAAuditLogService;
 import org.jbpm.process.audit.query.AbstractAuditDeleteBuilderImpl;
 import org.kie.api.runtime.CommandExecutor;
+import org.kie.api.runtime.process.ProcessInstance;
 import org.kie.internal.runtime.manager.audit.query.ErrorInfoDeleteBuilder;
 
 import static org.kie.internal.query.QueryParameterIdentifiers.EXECUTOR_TIME_LIST;
@@ -78,4 +79,25 @@ protected String getQueryTable() {
         return ERROR_INFO_LOG_DELETE;
     }
 
+    @Override
+    protected boolean isSubquerySupported() {
+        return true;
+    }
+
+    @Override
+    protected Subquery applyParameters(Subquery subquery) {
+        return subquery;
+    }
+
+    @Override
+    protected Subquery getSubQuery() {
+        String queryBaseStr = "SELECT r.id from RequestInfo r where r.status in ('ERROR', 'CANCELLED', 'DONE') AND " +
+        "(r.processInstanceId is null OR r.processInstanceId not in " +
+            "(SELECT spl.processInstanceId FROM ProcessInstanceLog spl where spl.status in (" +
+            ProcessInstance.STATE_PENDING + "," + // 0
+            ProcessInstance.STATE_ACTIVE + "," + // 1
+            ProcessInstance.STATE_SUSPENDED + // 4
+            ")))";
+        return new Subquery("l.requestInfo", queryBaseStr, 1);
+    }
 }

jbpm-services/jbpm-executor/src/main/java/org/jbpm/executor/impl/jpa/ExecutionErrorInfoDeleteBuilderImpl.java

@@ -20,6 +20,7 @@
 import org.jbpm.process.audit.JPAAuditLogService;
 import org.jbpm.process.audit.query.AbstractAuditDeleteBuilderImpl;
 import org.jbpm.runtime.manager.impl.jpa.ExecutionErrorInfo;
+import org.kie.api.runtime.process.ProcessInstance;
 import org.kie.internal.runtime.manager.audit.query.ExecutionErrorInfoDeleteBuilder;
 
 import static org.kie.internal.query.QueryParameterIdentifiers.ERROR_DATE_LIST;
@@ -52,4 +53,24 @@ public ExecutionErrorInfoDeleteBuilder dateRangeEnd(Date rangeEnd) {
         addRangeParameter(ERROR_DATE_LIST, "date range end", rangeEnd, false);
         return this;
     }
+
+    @Override
+    protected boolean isSubquerySupported() {
+        return true;
+    }
+
+    @Override
+    protected Subquery applyParameters(Subquery subquery) {
+        return subquery;
+    }
+
+    @Override
+    protected Subquery getSubQuery() {
+        String queryBaseStr = "SELECT spl.processInstanceId FROM ProcessInstanceLog spl where spl.status in (" +
+            ProcessInstance.STATE_PENDING + "," + // 0
+            ProcessInstance.STATE_ACTIVE + "," + // 1
+            ProcessInstance.STATE_SUSPENDED + // 4
+            ")";
+        return new Subquery("l.processInstanceId", queryBaseStr, 1, "not in");
+    }
 }

jbpm-services/jbpm-executor/src/main/java/org/jbpm/executor/impl/jpa/RequestInfoDeleteBuilderImpl.java

@@ -23,6 +23,7 @@
 import org.jbpm.process.audit.query.AbstractAuditDeleteBuilderImpl;
 import org.kie.api.executor.STATUS;
 import org.kie.api.runtime.CommandExecutor;
+import org.kie.api.runtime.process.ProcessInstance;
 import org.kie.internal.runtime.manager.audit.query.RequestInfoLogDeleteBuilder;
 
 import static org.kie.internal.query.QueryParameterIdentifiers.DEPLOYMENT_ID_LIST;
@@ -84,7 +85,7 @@ public RequestInfoLogDeleteBuilder status(STATUS... status) {
 		if (checkIfNull(status)) {
 			return this;
 		}
-		
+
 		addObjectParameter(EXECUTOR_STATUS_LIST, "status", status);
         return this;
 	}
@@ -99,5 +100,24 @@ protected String getQueryTable() {
         return REQUES_INFO_LOG_DELETE;
     }
 
+    @Override
+    protected boolean isSubquerySupported() {
+        return true;
+    }
+
+    @Override
+    protected Subquery applyParameters(Subquery subquery) {
+        return subquery;
+    }
 
+    @Override
+    protected Subquery getSubQuery() {
+        String queryBaseStr = "select ri.id from RequestInfo ri where ri.processInstanceId is null " +
+            "or ri.processInstanceId not in (select pil.processInstanceId from ProcessInstanceLog pil where pil.status in (" +
+            ProcessInstance.STATE_PENDING + "," + // 0
+            ProcessInstance.STATE_ACTIVE + "," + // 1
+            ProcessInstance.STATE_SUSPENDED + // 4
+            "))";
+        return new Subquery("l.id", queryBaseStr, 1);
+    }
 }

jbpm-services/jbpm-executor/src/test/filtered-resources/META-INF/persistence.xml

@@ -10,6 +10,7 @@
     <mapping-file>META-INF/Executor-orm.xml</mapping-file>
     <class>org.jbpm.executor.entities.ErrorInfo</class>
     <class>org.jbpm.executor.entities.RequestInfo</class>
+    <class>org.jbpm.process.audit.ProcessInstanceLog</class>
     <class>org.jbpm.runtime.manager.impl.jpa.ExecutionErrorInfo</class>
     <exclude-unlisted-classes>true</exclude-unlisted-classes>
 

jbpm-services/jbpm-executor/src/test/java/org/jbpm/executor/impl/wih/CleanupLogCommandWithProcessTest.java

@@ -31,6 +31,7 @@
 
 import org.jbpm.executor.ExecutorServiceFactory;
 import org.jbpm.executor.impl.ExecutorServiceImpl;
+import org.jbpm.executor.impl.jpa.ExecutorJPAAuditService;
 import org.jbpm.executor.test.CountDownAsyncJobListener;
 import org.jbpm.process.audit.JPAAuditLogService;
 import org.jbpm.process.instance.impl.demo.DoNothingWorkItemHandler;
@@ -186,7 +187,7 @@ public void testRunProcessWithAsyncHandlerDontDeleteActive() throws Exception {
                     public Map<String, WorkItemHandler> getWorkItemHandlers(RuntimeEngine runtime) {
 
                         Map<String, WorkItemHandler> handlers = super.getWorkItemHandlers(runtime);
-                        handlers.put("async", new DoNothingWorkItemHandler());
+                        handlers.put("async", new AsyncWorkItemHandler(executorService, "org.jbpm.executor.ThrowExceptionCommand"));
                         return handlers;
                     }
 
@@ -204,17 +205,20 @@ public Map<String, WorkItemHandler> getWorkItemHandlers(RuntimeEngine runtime) {
         assertEquals(0, getNodeInstanceLogSize("ScriptTask"));
         assertEquals(0, getTaskLogSize("ScriptTask"));
         assertEquals(0, getVariableLogSize("ScriptTask"));
+        assertEquals(0, getRequestInfoSize());
+        assertEquals(0, getErrorInfoSize());
 
         Date startDate = new Date();
 
         ProcessInstance processInstance = ksession.startProcess("ScriptTask");
         assertEquals(ProcessInstance.STATE_ACTIVE, processInstance.getState());
-        
+
         assertEquals(1, getProcessLogSize("ScriptTask"));
         assertEquals(5, getNodeInstanceLogSize("ScriptTask"));
         assertEquals(0, getTaskLogSize("ScriptTask"));
         assertEquals(0, getVariableLogSize("ScriptTask"));
-
+        assertEquals(1, getRequestInfoSize());
+
         scheduleLogCleanup(false, true, false, startDate, "ScriptTask", "yyyy-MM-dd", manager.getIdentifier());
         countDownListener.waitTillCompleted();
         System.out.println("Aborting process instance " + processInstance.getId());
@@ -225,6 +229,8 @@ public Map<String, WorkItemHandler> getWorkItemHandlers(RuntimeEngine runtime) {
         assertEquals(5, getNodeInstanceLogSize("ScriptTask"));
         assertEquals(0, getTaskLogSize("ScriptTask"));
         assertEquals(0, getVariableLogSize("ScriptTask"));
+        assertEquals(2, getRequestInfoSize());
+        assertEquals(1, getErrorInfoSize());
 
         runtime.getKieSession().abortProcessInstance(processInstance.getId());
 
@@ -235,6 +241,8 @@ public Map<String, WorkItemHandler> getWorkItemHandlers(RuntimeEngine runtime) {
         assertEquals(6, getNodeInstanceLogSize("ScriptTask"));
         assertEquals(0, getTaskLogSize("ScriptTask"));
         assertEquals(0, getVariableLogSize("ScriptTask"));
+        assertEquals(2, getRequestInfoSize());
+        assertEquals(1, getErrorInfoSize());
 
         // and start another one to keep it active while cleanup happens
         processInstance = ksession.startProcess("ScriptTask");
@@ -244,6 +252,8 @@ public Map<String, WorkItemHandler> getWorkItemHandlers(RuntimeEngine runtime) {
         assertEquals(11, getNodeInstanceLogSize("ScriptTask"));
         assertEquals(0, getTaskLogSize("ScriptTask"));
         assertEquals(0, getVariableLogSize("ScriptTask"));
+        assertEquals(3, getRequestInfoSize());
+        //assertEquals(2, getErrorInfoSize());
 
         Thread.sleep(1000);
 
@@ -255,6 +265,8 @@ public Map<String, WorkItemHandler> getWorkItemHandlers(RuntimeEngine runtime) {
         assertEquals(5, getNodeInstanceLogSize("ScriptTask"));
         assertEquals(0, getTaskLogSize("ScriptTask"));
         assertEquals(0, getVariableLogSize("ScriptTask"));
+        assertEquals(2, getRequestInfoSize());
+        assertEquals(1, getErrorInfoSize());
     }
 
     @Test
@@ -327,9 +339,33 @@ private ExecutorService buildExecutorService() {
         executorService = ExecutorServiceFactory.newExecutorService(emf);
 
         executorService.init();
+        executorService.setRetries(0);
 
         return executorService;
     }
+
+    @Test
+    public void testCleanupLogOfScheduler() throws Exception {
+        CountDownAsyncJobListener countDownListener = configureListener(1);
+        RuntimeEnvironment environment = RuntimeEnvironmentBuilder.Factory.get().newDefaultBuilder()
+                .userGroupCallback(userGroupCallback)
+                .entityManagerFactory(emf)
+                .get();
+
+        manager = RuntimeManagerFactory.Factory.get().newSingletonRuntimeManager(environment);
+        int NO_OF_JOBS = 3;
+        for(int i=0;i<NO_OF_JOBS;i++) {
+            scheduleLogCleanup(true, true, true, new Date(), null, "yyyy-MM-dd HH:mm:ss", manager.getIdentifier());
+        }
+        assertEquals(NO_OF_JOBS, getRequestInfoSize());
+        Thread.sleep(1000);
+
+        scheduleLogCleanup(true, true, false, new Date(), null, "yyyy-MM-dd HH:mm:ss", manager.getIdentifier());
+        countDownListener.reset(1);
+        countDownListener.waitTillCompleted();
+
+        assertEquals(1, getRequestInfoSize());
+    }
 
 	private void scheduleLogCleanup(boolean skipProcessLog,
 			boolean skipTaskLog, boolean skipExecutorLog, Date olderThan,
@@ -387,4 +423,18 @@ private int getTaskVariableLogSize(String processId) {
                 .getResultList()
                 .size();
     }
+
+    private int getRequestInfoSize() {
+        return new ExecutorJPAAuditService(emf).requestInfoQueryBuilder()
+                .build()
+                .getResultList()
+                .size();
+    }
+
+    private int getErrorInfoSize() {
+        return new ExecutorJPAAuditService(emf).errorInfoQueryBuilder()
+                .build()
+                .getResultList()
+                .size();
+    }
 }

jbpm-test-coverage/src/test/java/org/jbpm/test/functional/log/ExecutorLogCleanTest.java

@@ -112,6 +112,9 @@ public void deleteErrorLogsByDate() throws Exception {
         List<ErrorInfo> errorList = getExecutorService().getAllErrors(new QueryContext());
         Assertions.assertThat(errorList).hasSize(2);
 
+        // Abort running process instance
+        ksession.abortProcessInstance(pi.getId());
+
         // Delete a record
         int resultCount = auditService.errorInfoLogDeleteBuilder()
                 .date(errorList.get(0).getTime())
@@ -132,9 +135,6 @@ public void deleteErrorLogsByDate() throws Exception {
 
         // Assert remaining records
         Assertions.assertThat(getExecutorService().getAllErrors(new QueryContext())).hasSize(remaining);
-
-        // Abort running process instance
-        ksession.abortProcessInstance(pi.getId());
     }
 
 }