Ikasan 2366 add housekeeping upon harvesting feature

ikasaneip/error-reporting/src/main/java/org/ikasan/error/reporting/ErrorReportingAutoConfiguration.java

@@ -12,6 +12,7 @@
 import org.ikasan.spec.housekeeping.HousekeepService;
 import org.ikasan.spec.serialiser.SerialiserFactory;
 import org.springframework.beans.factory.annotation.Qualifier;
+import org.springframework.beans.factory.annotation.Value;
 import org.springframework.beans.factory.config.ConfigurableBeanFactory;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
@@ -28,6 +29,9 @@
 @ImportResource("/error-reporting-transaction.xml")
 public class ErrorReportingAutoConfiguration
 {
+    @Value("${errorReportingHousekeepingJob-deleteOnceHarvested:false}")
+    private boolean deleteOnceHarvested;
+
     @Bean
     @Scope(value = ConfigurableBeanFactory.SCOPE_PROTOTYPE)
     public ErrorReportingService errorReportingService(ErrorReportingServiceFactory errorReportingServiceFactory) {
@@ -42,7 +46,7 @@ public ErrorReportingServiceFactory errorReportingServiceFactory(SerialiserFacto
 
     @Bean(name = "errorReportingManagementService")
     public ErrorReportingManagementService errorReportingManagementService(ErrorReportingServiceDao errorReportingServiceDao
-        , ErrorManagementDao errorManagementDao) {
+        , @Qualifier("errorManagementDao") ErrorManagementDao errorManagementDao) {
         return new ErrorReportingManagementServiceImpl(errorManagementDao, errorReportingServiceDao);
     }
 
@@ -51,9 +55,9 @@ public ErrorReportingServiceDao errorReportingServiceDao() {
         return new HibernateErrorReportingServiceDao();
     }
 
-    @Bean
+    @Bean(name = "errorManagementDao")
     public ErrorManagementDao errorManagementDao() {
-        return new HibernateErrorManagementDao();
+        return new HibernateErrorManagementDao(this.deleteOnceHarvested);
     }
 
     @Bean

ikasaneip/error-reporting/src/main/java/org/ikasan/error/reporting/dao/HibernateErrorManagementDao.java

@@ -73,6 +73,9 @@ public class HibernateErrorManagementDao implements ErrorManagementDao
     public static final String ERROR_OCCURRENCES_TO_DELETE_QUERY = "select uri from ErrorOccurrenceImpl eo " +
             " where eo.expiry < :" + NOW;
 
+    public static final String HARVESTED_ERROR_OCCURRENCES_TO_DELETE_QUERY = "select uri from ErrorOccurrenceImpl eo " +
+        " where eo.harvested = true order by eo.timestamp";
+
     public static final String ERROR_OCCURRENCE_DELETE_QUERY = "delete ErrorOccurrenceImpl eo " +
             " where eo.uri in(:" + EVENT_IDS + ")";
 
@@ -84,7 +87,21 @@ public class HibernateErrorManagementDao implements ErrorManagementDao
     @PersistenceContext(unitName = "error-reporting")
     private EntityManager entityManager;
 
-	@Override
+    private boolean deleteOnceHarvested;
+
+    /**
+     * HibernateErrorManagementDao is a class that represents a data access object for managing errors in Hibernate.
+     * It provides methods for saving, deleting, and finding error occurrences, as well as housekeeping
+     * and updating harvested records.
+     *
+     * @param deleteOnceHarvested a boolean flag indicating whether harvested error occurrences should be deleted
+     *                            after being processed
+     */
+    public HibernateErrorManagementDao(boolean deleteOnceHarvested) {
+        this.deleteOnceHarvested = deleteOnceHarvested;
+    }
+
+    @Override
 	public void saveErrorOccurrence(ErrorOccurrence errorOccurrence)
 	{
 		this.entityManager.persist(this.entityManager.contains(errorOccurrence) ? errorOccurrence : entityManager.merge(errorOccurrence));
@@ -243,8 +260,9 @@ public Long getNumberOfModuleErrors(String moduleName, boolean excluded, boolean
 
 	@Override
 	public void housekeep(final Integer numToHousekeep) {
-		Query query = this.entityManager.createQuery(ERROR_OCCURRENCES_TO_DELETE_QUERY);
-        query.setParameter(NOW, System.currentTimeMillis());
+		Query query = this.entityManager.createQuery(this.deleteOnceHarvested
+            ? HARVESTED_ERROR_OCCURRENCES_TO_DELETE_QUERY : ERROR_OCCURRENCES_TO_DELETE_QUERY);
+        if(!this.deleteOnceHarvested)query.setParameter(NOW, System.currentTimeMillis());
         query.setMaxResults(numToHousekeep);
 
         List<Long> errorUris = (List<Long>)query.getResultList();

ikasaneip/error-reporting/src/test/java/org/ikasan/error/reporting/ErrorReportingTestAutoConfiguration.java

@@ -3,6 +3,8 @@
 import com.arjuna.ats.internal.jta.transaction.arjunacore.TransactionManagerImple;
 import com.arjuna.ats.jta.UserTransaction;
 import jakarta.persistence.EntityManagerFactory;
+import org.ikasan.error.reporting.dao.ErrorManagementDao;
+import org.ikasan.error.reporting.dao.HibernateErrorManagementDao;
 import org.ikasan.serialiser.converter.JmsMapMessageConverter;
 import org.ikasan.serialiser.converter.JmsTextMessageConverter;
 import org.ikasan.serialiser.service.SerialiserFactoryKryoImpl;
@@ -30,6 +32,11 @@
 @ImportResource("/test-transaction.xml")
 public class ErrorReportingTestAutoConfiguration
 {
+    @Bean(name = "deleteOnceHarvestedErrorManagementDao")
+    public ErrorManagementDao deleteOnceHarvestedErrorManagementDao() {
+        return new HibernateErrorManagementDao(true);
+    }
+
     @Bean(name = {"ikasan.xads", "ikasan.ds"})
     public DataSource ikasanDataSource() {
         DriverManagerDataSource dataSource = new DriverManagerDataSource();

ikasaneip/error-reporting/src/test/java/org/ikasan/error/reporting/dao/HibernateErrorManagementDaoTest.java

@@ -73,6 +73,9 @@ public class HibernateErrorManagementDaoTest
 {    
     @Resource
     ErrorManagementDao errorManagementDao;
+
+    @Resource
+    ErrorManagementDao deleteOnceHarvestedErrorManagementDao;
 
     @Resource
 	ErrorReportingServiceDao errorReportingServiceDao;
@@ -125,9 +128,7 @@ public void test_count_error_occurrence_for_module()
     	this.errorReportingServiceDao.save(eo);
     	this.errorReportingServiceDao.save(eo1);
     	this.errorReportingServiceDao.save(eo2);
-
-    	System.out.println(this.errorManagementDao.getNumberOfModuleErrors("moduleName", false, false, new Date(System.currentTimeMillis() - 100000000), new Date(System.currentTimeMillis() + 100000000)));
-
+
     	Assert.assertTrue(this.errorManagementDao.getNumberOfModuleErrors("moduleName", false, false, new Date(System.currentTimeMillis() - 100000000), new Date(System.currentTimeMillis() + 100000000)) == 3);
     }
 
@@ -147,11 +148,50 @@ public void test_harvest_success()
             errorOccurrences.add(eo);
         }
 
-        Assert.assertEquals("Harvestable records == 1000", this.errorManagementDao.getHarvestableRecords(5000).size(), 1000);
+        Assert.assertEquals("Harvestable records == 1000", 1000, this.errorManagementDao.getHarvestableRecords(5000).size());
 
         this.errorManagementDao.updateAsHarvested(errorOccurrences);
 
-        Assert.assertEquals("Harvestable records == 0", this.errorManagementDao.getHarvestableRecords(5000).size(), 0);
+        Assert.assertEquals("Harvestable records == 0", 0, this.errorManagementDao.getHarvestableRecords(5000).size());
+    }
+
+    @Test
+    @DirtiesContext
+    public void test_housekeep_success()
+    {
+        this.errorManagementDao.setHarvestQueryOrdered(true);
+
+        for(int i=0; i<1000; i++)
+        {
+            ErrorOccurrenceImpl eo = new ErrorOccurrenceImpl("moduleName", "flowName", "flowElementName", "errorDetail", "errorMessage", "exceptionClass", 100, new byte[100], "errorString");
+            this.errorReportingServiceDao.save(eo);
+        }
+
+        Assert.assertEquals("Harvestable records == 1000", this.errorManagementDao.getHarvestableRecords(5000).size(), 1000);
+
+        this.errorManagementDao.housekeep(100);
+
+        Assert.assertEquals("Harvestable records == 900", this.errorManagementDao.getHarvestableRecords(5000).size(), 900);
+    }
+
+    @Test
+    @DirtiesContext
+    public void test_housekeep_once_harvested_success()
+    {
+        this.deleteOnceHarvestedErrorManagementDao.setHarvestQueryOrdered(true);
+
+        for(int i=0; i<1000; i++)
+        {
+            ErrorOccurrenceImpl eo = new ErrorOccurrenceImpl("moduleName", "flowName", "flowElementName", "errorDetail", "errorMessage", "exceptionClass", 100000000L, new byte[100], "errorString");
+            eo.setHarvested(true);
+            this.errorReportingServiceDao.save(eo);
+        }
+
+        Assert.assertEquals("Harvestable records == 1000", this.errorManagementDao.getHarvestableRecords(5000).size(), 1000);
+
+        this.deleteOnceHarvestedErrorManagementDao.housekeep(100);
+
+        Assert.assertEquals("Harvestable records == 900", this.errorManagementDao.getHarvestableRecords(5000).size(), 900);
     }
 
     @Test

ikasaneip/housekeeping/Readme.md

@@ -15,6 +15,7 @@ Each and every one of the jobs can be tuned and configured by following set of p
 - {jobName}-houseKeepingBatchSize defaults to 2500
 - {jobName}-transactionBatchSize defaults to 200
 - {jobName}-cronExpression defaults to '0 0/1 * * * ?'
+- {jobName}-deleteOnceHarvested defaults to false (NB applicable for replayHousekeepingJob, wiretapHousekeepingJob, errorReportingHousekeepingJob, systemEventServiceHousekeepingJob)
 - {jobName}-enabled defaults to true
 
 
@@ -27,24 +28,28 @@ Each and every one of the jobs can be tuned and configured by following set of p
 replayHousekeepingJob-houseKeepingBatchSize=2500
 replayHousekeepingJob-transactionBatchSize defaults=200
 replayHousekeepingJob-cronExpression=0 0/1 * * * ?
+replayHousekeepingJob-deleteOnceHarvested=false
 replayHousekeepingJob-enabled=true
 
 # Wiretap housekeeping settings
 wiretapHousekeepingJob-houseKeepingBatchSize=2500
 wiretapHousekeepingJob-transactionBatchSize defaults=200
 wiretapHousekeepingJob-cronExpression=0 0/1 * * * ?
+wiretapHousekeepingJob-deleteOnceHarvested=false
 wiretapHousekeepingJob-enabled=true
 
 # Error housekeeping settings
 errorReportingHousekeepingJob-houseKeepingBatchSize=2500
 errorReportingHousekeepingJob-transactionBatchSize defaults=200
 errorReportingHousekeepingJob-cronExpression=0 0/1 * * * ?
+errorReportingHousekeepingJob-deleteOnceHarvested=false
 errorReportingHousekeepingJob-enabled=true
 
 # SystemEvents housekeeping settings
 systemEventServiceHousekeepingJob-houseKeepingBatchSize=2500
 systemEventServiceHousekeepingJob-transactionBatchSize defaults=200
 systemEventServiceHousekeepingJob-cronExpression=0 0/1 * * * ?
+systemEventServiceHousekeepingJob-deleteOnceHarvested=false
 systemEventServiceHousekeepingJob-enabled=true
 
 # Duplicate Filter housekeeping settings

ikasaneip/replay/src/main/java/org/ikasan/replay/ReplayAutoConfiguration.java

@@ -7,6 +7,7 @@
 import org.ikasan.spec.replay.*;
 import org.ikasan.spec.serialiser.SerialiserFactory;
 import org.springframework.beans.factory.annotation.Qualifier;
+import org.springframework.beans.factory.annotation.Value;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
 import org.springframework.orm.jpa.JpaVendorAdapter;
@@ -19,8 +20,11 @@
 @Configuration
 public class ReplayAutoConfiguration {
 
+    @Value("${replayHousekeepingJob-deleteOnceHarvested:false}")
+    private boolean deleteOnceHarvested;
+
     @Bean(name = "replayManagementService")
-    public ReplayManagementService replayManagementService(ReplayDao replayDao, ReplayAuditDao replayAuditDao) {
+    public ReplayManagementService replayManagementService(@Qualifier("replayDao")ReplayDao replayDao, ReplayAuditDao replayAuditDao) {
         return new ReplayManagementServiceImpl(replayDao, replayAuditDao);
     }
 
@@ -30,13 +34,14 @@ public ReplayService replayService(ReplayAuditDao replayAuditDao) {
     }
 
     @Bean
-    public ReplayRecordService replayRecordService(ReplayDao replayDao, SerialiserFactory serialiserFactory) {
+    public ReplayRecordService replayRecordService(@Qualifier("replayDao")ReplayDao replayDao
+        , SerialiserFactory serialiserFactory) {
         return new ReplayRecordServiceImpl(serialiserFactory, replayDao);
     }
 
-    @Bean
+    @Bean(name = "replayDao")
     public ReplayDao replayDao() {
-        return new HibernateReplayDao();
+        return new HibernateReplayDao(this.deleteOnceHarvested);
     }
 
     @Bean

ikasaneip/replay/src/main/java/org/ikasan/replay/dao/HibernateReplayDao.java

@@ -93,6 +93,8 @@ public class HibernateReplayDao implements ReplayDao<Long>,ReplayAuditDao<Replay
 	public static final String REPLAY_EVENTS_TO_DELETE_QUERY = "select id from ReplayEventImpl re " +
 			" where re.expiry < :" + NOW;
 
+    public static final String HARVESTED_REPLAY_EVENTS_TO_DELETE_QUERY = "select id from ReplayEventImpl re " +
+        " where re.harvested=true order by re.timestamp ASC";
 	public static final String REPLAY_EVENTS_DELETE_QUERY = "delete ReplayEventImpl re " +
 			" where re.id in(:" + EVENT_IDS + ")";
 
@@ -111,14 +113,19 @@ where re.id not in(select distinct rae.id.replayAuditId from ReplayAuditEventImp
         " where w.id in(:" + EVENT_IDS + ")";
 
     private boolean isHarvestQueryOrdered = false;
+    private boolean deleteOnceHarvested;
 
     @PersistenceContext(unitName = "xa-replay")
     private EntityManager entityManager;
 
 
-    public HibernateReplayDao()
-    {
-
+    /**
+     * Constructs a new instance of the {@link HibernateReplayDao} class with the specified deleteOnceHarvested flag.
+     *
+     * @param deleteOnceHarvested a boolean value indicating whether replay events should be deleted once harvested
+     */
+    public HibernateReplayDao(boolean deleteOnceHarvested) {
+        this.deleteOnceHarvested = deleteOnceHarvested;
     }
 
     /* (non-Javadoc)
@@ -305,7 +312,6 @@ public List<ReplayEvent> getReplayEvents(List<String> moduleNames,
 
         if (payloadContent != null && payloadContent.length() > 0)
         {
-            //criteria.add(Restrictions.like("eventAsString", payloadContent, MatchMode.ANYWHERE));
             predicates.add( builder.like(root.get("eventAsString"),payloadContent));
         }
 
@@ -407,8 +413,9 @@ public Long getNumberReplayAuditEventsByAuditId(final Long id)
 	@Override
 	public void housekeep(final Integer numToHousekeep)
 	{
-        Query query = this.entityManager.createQuery(REPLAY_EVENTS_TO_DELETE_QUERY);
-        query.setParameter(NOW, System.currentTimeMillis());
+        Query query = this.entityManager.createQuery(this.deleteOnceHarvested ?
+            HARVESTED_REPLAY_EVENTS_TO_DELETE_QUERY : REPLAY_EVENTS_TO_DELETE_QUERY);
+        if(!this.deleteOnceHarvested)query.setParameter(NOW, System.currentTimeMillis());
         query.setMaxResults(numToHousekeep);
 
         List<Long> replayEventIds = query.getResultList();

ikasaneip/replay/src/test/java/org/ikasan/replay/ReplayTestAutoConfiguration.java

@@ -4,14 +4,20 @@
 import com.arjuna.ats.jta.UserTransaction;
 import jakarta.jms.MapMessage;
 import jakarta.jms.TextMessage;
+import org.ikasan.replay.dao.HibernateReplayDao;
+import org.ikasan.replay.service.ReplayManagementServiceImpl;
 import org.ikasan.serialiser.converter.JmsMapMessageConverter;
 import org.ikasan.serialiser.converter.JmsTextMessageConverter;
 import org.ikasan.serialiser.service.SerialiserFactoryKryoImpl;
 import org.ikasan.spec.module.ModuleContainer;
+import org.ikasan.spec.replay.ReplayAuditDao;
+import org.ikasan.spec.replay.ReplayDao;
+import org.ikasan.spec.replay.ReplayManagementService;
 import org.ikasan.spec.serialiser.Serialiser;
 import org.ikasan.spec.serialiser.SerialiserFactory;
 import org.jmock.Mockery;
 import org.jmock.imposters.ByteBuddyClassImposteriser;
+import org.springframework.beans.factory.annotation.Qualifier;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
 import org.springframework.context.annotation.ImportResource;
@@ -56,6 +62,17 @@ public ModuleContainer moduleContainer() {
         return mockery.mock(ModuleContainer.class);
     }
 
+    @Bean(name = "deleteOnceHarvestedReplayManagementService")
+    public ReplayManagementService deleteOnceHarvestedReplayManagementService(@Qualifier("deleteOnceHarvestedReplayDao")ReplayDao replayDao
+        , ReplayAuditDao replayAuditDao) {
+        return new ReplayManagementServiceImpl(replayDao, replayAuditDao);
+    }
+
+    @Bean(name = "deleteOnceHarvestedReplayDao")
+    public ReplayDao deleteOnceHarvestedReplayDao() {
+        return new HibernateReplayDao(true);
+    }
+
     @Bean(name = {"ikasan.xads", "ikasan.ds"})
     public DataSource ikasanDataSource() {
         DriverManagerDataSource dataSource = new DriverManagerDataSource();