SSP-2183: Adjusting Listeners to Better Populate the Emailed Report in the Event of a Rollback

ssp-data-importer-impl/src/main/java/org/jasig/ssp/util/importer/job/listener/RawItemValidateProcessorListener.java

@@ -18,24 +18,36 @@
  */
 package org.jasig.ssp.util.importer.job.listener;
 
+import java.util.ArrayList;
+import java.util.List;
+
 import org.jasig.ssp.util.importer.job.domain.RawItem;
-import org.jasig.ssp.util.importer.job.tasklet.BatchFinalizer;
+import org.jasig.ssp.util.importer.job.report.ErrorEntry;
+import org.jasig.ssp.util.importer.job.report.StepType;
 import org.jasig.ssp.util.importer.job.validation.map.metadata.validation.violation.TableViolationException;
 import org.jasig.ssp.util.importer.job.validation.map.metadata.validation.violation.ViolationException;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.batch.core.ItemProcessListener;
-import org.springframework.batch.core.scope.context.StepContext;
-import org.springframework.batch.core.scope.context.StepSynchronizationManager;
+import org.springframework.batch.core.StepExecution;
+import org.springframework.batch.core.annotation.BeforeStep;
+import org.springframework.batch.item.ExecutionContext;
+
+/**
+ * ProcessListener is being leveraged to process process errors as they happen instead of
+ * waiting for a skip listener to process them.  The skip listener waits until right before
+ * a commit to happen to process(which will not happen if a rollback is called).
+ **/
 
 public class RawItemValidateProcessorListener implements ItemProcessListener<RawItem, RawItem> {
 
     private static final  Logger logger = LoggerFactory.getLogger(RawItemValidateProcessorListener.class);
     Boolean hasTableViolation = false;
+    private ExecutionContext executionContext;
+    private final static String EOL = System.getProperty("line.separator");
 
     @Override
     public void beforeProcess(RawItem item) {
-
     }
 
     @Override
@@ -44,21 +56,36 @@ public void afterProcess(RawItem item, RawItem result) {
 
     @Override
     public void onProcessError(RawItem item, Exception e) {
-        if(!e.getClass().equals(TableViolationException.class) && !e.getClass().equals(ViolationException.class) || hasTableViolation == false){
-            logger.error(e.getMessage());
+        if (!e.getClass().equals(TableViolationException.class) && !e.getClass().equals(ViolationException.class) || hasTableViolation == false) {
+            logger.error("ERROR on Upsert Process: " + EOL + e.getMessage());
+        } else if (e.getClass().equals(TableViolationException.class)){
+            hasTableViolation = true;
+            ViolationException violation = (TableViolationException) e;
+            logger.error("ERROR on Upsert Process, line:" + violation.getLineNumber() + EOL + e.getMessage());
         }
+        putErrorInJobContext(item, e);
+    }
 
-        if (e.getClass().equals(ViolationException.class)){
-            ViolationException violation = (ViolationException)e;
-            String EOL = System.getProperty("line.separator");
-            logger.error("error on line:" + violation.getLineNumber() + EOL + e.getMessage());
+    @SuppressWarnings("unchecked")    public void putErrorInJobContext(RawItem item, Throwable t) {
+        String lineNumber = null;
+        if(t instanceof ViolationException) {
+            lineNumber = ((ViolationException) t).getLineNumber() != null ? ((ViolationException) t).getLineNumber().toString() : null;
         }
-
-        if(e.getClass().equals(TableViolationException.class) && hasTableViolation == false){
-            hasTableViolation = true;
-            ViolationException violation = (TableViolationException)e;
-            String EOL = System.getProperty("line.separator");
-            logger.error("error on line:" + violation.getLineNumber() + EOL + e.getMessage());
+
+        String fileName = item.getResource().getFilename();
+        String[] tableName = fileName.split("\\.");
+        ErrorEntry error = new ErrorEntry(tableName[0], item.getRecord().toString(), t.getMessage(), StepType.VALIDATE);
+        error.setLineNumber(lineNumber);
+        List<ErrorEntry> errors =(List<ErrorEntry>) executionContext.get("errors");
+        if(errors == null) {
+            errors = new ArrayList<ErrorEntry>();
         }
+        errors.add(error);
+        executionContext.put("errors", errors);
+    }
+
+    @BeforeStep
+    public void saveStepExecution(StepExecution stepExecution) {
+        this.executionContext = stepExecution.getJobExecution().getExecutionContext();
     }
 }

ssp-data-importer-impl/src/main/java/org/jasig/ssp/util/importer/job/listener/ValidationSkipListener.java → ssp-data-importer-impl/src/main/java/org/jasig/ssp/util/importer/job/listener/ReadListener.java

@@ -27,63 +27,62 @@
 import org.jasig.ssp.util.importer.job.validation.map.metadata.validation.violation.ViolationException;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
-import org.springframework.batch.core.SkipListener;
+import org.springframework.batch.core.ItemReadListener;
 import org.springframework.batch.core.StepExecution;
 import org.springframework.batch.core.annotation.BeforeStep;
-import org.springframework.batch.core.scope.context.StepContext;
-import org.springframework.batch.core.scope.context.StepSynchronizationManager;
+import org.springframework.batch.item.ExecutionContext;
 
+/**
+ * ReadListener is being leveraged to process read errors as they happen instead of
+ * waiting for a skip listener to process them.  The skip listener waits until right before
+ * a commit to happen to process(which will not happen if a rollback is called).
+ **/
 
-public class ValidationSkipListener implements SkipListener<RawItem, RawItem> {
+public class ReadListener implements ItemReadListener<RawItem> {
 
-    private StepExecution stepExecution;
+    private static final  Logger logger = LoggerFactory.getLogger(ReadListener.class);
+    private ExecutionContext executionContext;
 
-    Logger logger = LoggerFactory.getLogger(ValidationSkipListener.class);
     @Override
-    public void onSkipInRead(Throwable t) {
-        logger.error("ERROR on Upsert Read", t);
+    public void beforeRead() {
     }
 
     @Override
-    public void onSkipInWrite(RawItem item, Throwable t) {
-        logger.error("ERROR on Upsert Read", t);
+    public void afterRead(RawItem item) {
     }
 
-    @SuppressWarnings("unchecked")
     @Override
-    public void onSkipInProcess(RawItem item, Throwable t) {
-        logger.error("ERROR on Upsert Process", t);
-
+    public void onReadError(Exception e) {
+        if (e.getClass().equals(ViolationException.class)) {
+            ViolationException violation = (ViolationException) e;
+            String EOL = System.getProperty("line.separator");
+            logger.error("ERROR on Read, line:" + violation.getLineNumber() + EOL + e.getMessage());
+        } else {
+            logger.error("ERROR on Read" + e.getMessage());
+        }
+        putErrorInJobContext(e);        
+    }
+
+    @SuppressWarnings("unchecked")
+    public void putErrorInJobContext(Throwable t) {
         String lineNumber = null;
         if(t instanceof ViolationException)
         {
-            lineNumber = ((ViolationException)t).getLineNumber() != null ? ((ViolationException)t).getLineNumber().toString() : null;
+            lineNumber = ((ViolationException) t).getLineNumber() != null ? ((ViolationException) t).getLineNumber().toString() : null;
         }
-
-        String fileName = item.getResource().getFilename();
-        String[] tableName = fileName.split("\\.");
-        ErrorEntry error = new ErrorEntry(tableName[0],item.getRecord().toString(),t.getMessage(),StepType.VALIDATE);
+        ErrorEntry error = new ErrorEntry(null, null, t.getMessage(), StepType.READ);
         error.setLineNumber(lineNumber);
-        List<ErrorEntry> errors =(List<ErrorEntry>) stepExecution.getJobExecution().getExecutionContext().get("errors");
-        if(errors == null)
-        {
+        List<ErrorEntry> errors = (List<ErrorEntry>) executionContext.get("errors");
+        if(errors == null) {
             errors = new ArrayList<ErrorEntry>();
-        } 
+        }
         errors.add(error);
-        stepExecution.getJobExecution().getExecutionContext().put("errors", errors);
-    }
-
-    public StepExecution getStepExecution() {
-        return stepExecution;
-    }
-
-    public void setStepExecution(StepExecution stepExecution) {
-        this.stepExecution = stepExecution;
+        executionContext.put("errors", errors);
     }
 
     @BeforeStep
     public void saveStepExecution(StepExecution stepExecution) {
-        this.stepExecution = stepExecution;
+        this.executionContext = stepExecution.getJobExecution().getExecutionContext();
     }
 
 }

ssp-data-importer-impl/src/main/java/org/jasig/ssp/util/importer/job/listener/WriteListener.java

@@ -0,0 +1,100 @@
+/**
+ * Licensed to Jasig under one or more contributor license
+ * agreements. See the NOTICE file distributed with this work
+ * for additional information regarding copyright ownership.
+ * Jasig licenses this file to you under the Apache License,
+ * Version 2.0 (the "License"); you may not use this file
+ * except in compliance with the License. You may obtain a
+ * copy of the License at:
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on
+ * an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.jasig.ssp.util.importer.job.listener;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import org.jasig.ssp.util.importer.job.domain.RawItem;
+import org.jasig.ssp.util.importer.job.report.ErrorEntry;
+import org.jasig.ssp.util.importer.job.report.StepType;
+import org.jasig.ssp.util.importer.job.validation.map.metadata.validation.violation.ViolationException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.batch.core.ItemWriteListener;
+import org.springframework.batch.core.StepExecution;
+import org.springframework.batch.core.annotation.BeforeStep;
+import org.springframework.batch.item.ExecutionContext;
+
+/**
+ * WriteListener is being leveraged to process write errors as they happen instead of
+ * waiting for a skip listener to process them.  The skip listener waits until right before
+ * a commit to happen to process(which will not happen if a rollback is called).
+ **/
+
+public class WriteListener implements ItemWriteListener<RawItem> {
+
+    private static final  Logger logger = LoggerFactory.getLogger(WriteListener.class);
+    private ExecutionContext executionContext;
+
+    @Override
+    public void beforeWrite(List<? extends RawItem> items) {
+    }
+
+    @Override
+    public void afterWrite(List<? extends RawItem> items) {
+    }
+
+    @Override
+    public void onWriteError(Exception e, List<? extends RawItem> items) {
+        if (e.getClass().equals(ViolationException.class)){
+            ViolationException violation = (ViolationException) e;
+            String EOL = System.getProperty("line.separator");
+            logger.error("ERROR on Write, line:" + violation.getLineNumber() + EOL + e.getMessage());
+        } else {
+            logger.error("ERROR on Write" + e.getMessage());
+        }
+        putErrorInJobContext(e, items);        
+    }
+
+    @SuppressWarnings("unchecked")
+    public void putErrorInJobContext(Throwable t, List<? extends RawItem> items) {
+        String lineNumber = null;
+        if(t instanceof ViolationException) {
+            lineNumber = ((ViolationException) t).getLineNumber() != null ? ((ViolationException) t).getLineNumber().toString() : null;
+        }
+        RawItem lastItem = null;
+        String fileName = null;
+        String tableName = null;
+        String record = null;
+        if (items != null && !items.isEmpty()) {
+            lastItem = items.get(items.size() - 1);
+            fileName = lastItem.getResource().getFilename();
+            tableName = fileName.split("\\.")[0];
+            record = lastItem.getRecord().toString();
+        }
+
+        ErrorEntry error = new ErrorEntry(tableName, record, t.getMessage(), StepType.WRITE);
+        error.setLineNumber(lineNumber);
+        List<ErrorEntry> errors =(List<ErrorEntry>) executionContext.get("errors");
+        if(errors == null) {
+            errors = new ArrayList<ErrorEntry>();
+        }
+        errors.add(error);
+        executionContext.put("errors", errors);
+    }
+
+    @BeforeStep
+    public void saveStepExecution(StepExecution stepExecution) {
+        this.executionContext = stepExecution.getJobExecution().getExecutionContext();
+    }
+
+}

ssp-data-importer-impl/src/main/java/org/jasig/ssp/util/importer/job/report/StepType.java

@@ -19,6 +19,8 @@
 package org.jasig.ssp.util.importer.job.report;
 
 public enum StepType {
-VALIDATE,
-STAGEUPSERT
+    READ,
+    VALIDATE,
+    STAGEUPSERT,
+    WRITE
 }

ssp-data-importer-impl/src/main/java/org/jasig/ssp/util/importer/job/tasklet/PartialUploadGuard.java

@@ -93,7 +93,7 @@ public void setJobOperator(JobOperator jobOperator){
     public void setDirectory(Resource directory) throws PartialUploadGuardException, IOException{
         this.directory = directory;
         if(!directory.getFile().exists())
-            throw new PartialUploadGuardException("Input directory does not exist. Locoation:" + directory.getFile().getPath());
+            throw new PartialUploadGuardException("Input directory does not exist. Location:" + directory.getFile().getPath());
     }
 
     private Long getJobExecutionId(){

ssp-data-importer-impl/src/main/resources/importdatajob-context.xml

@@ -92,7 +92,8 @@
     <bean id="rawItemValidateProcessorListener"
       class="org.jasig.ssp.util.importer.job.listener.RawItemValidateProcessorListener">
     </bean>
-
+    <bean id="readListener" class="org.jasig.ssp.util.importer.job.listener.ReadListener" />
+    <bean id="writeListener" class="org.jasig.ssp.util.importer.job.listener.WriteListener" />
     <!-- BATCH CLEANUP -->
 
     <bean id="batchFinaliser" class="org.jasig.ssp.util.importer.job.tasklet.BatchFinalizer"
@@ -146,7 +147,6 @@
             <property name="metadataRepository" ref="metadataRepository" />
     </bean>
      <bean id="stagingAndUpsertListener" class="org.jasig.ssp.util.importer.job.listener.StagingAndUpsertListener"/>
-     <bean id="validationSkipListener" class="org.jasig.ssp.util.importer.job.listener.ValidationSkipListener"/>
      <bean id="parsingListener" class="org.jasig.ssp.util.importer.job.listener.ParsingListener"/>
      <bean id="stagingTableTruncator" class="org.jasig.ssp.util.importer.job.listener.StagingTableTruncator"
              p:dataSource-ref="dataSource">
@@ -236,10 +236,11 @@
 
                     <!-- Step-, chunk-, item-read, item-process, item-write, skip-scoped listeners, if needed. -->
                     <batch:listeners>
-                        <batch:listener ref="validationSkipListener" />
                         <batch:listener ref="processedRawCsvItemWriter" />
                         <batch:listener ref="parsingListener" />
+                        <batch:listener ref="readListener" />
                         <batch:listener ref="rawItemValidateProcessorListener" />
+                        <batch:listener ref="writeListener" />
                     </batch:listeners>
                 </batch:chunk>
             </batch:tasklet>