Merge pull request #212 from radcortez/master

Fixed Chunk Exception Test.

Author: arun-gupta

batch/chunk-exception/pom.xml

@@ -10,8 +10,8 @@
 
     <artifactId>chunk-exception</artifactId>
     <packaging>war</packaging>
-
-    <name>${project.artifactId}</name>
+    <name>Batch Chunk Exception</name>
+    <description>Chunk Exception Handling - Retrying and Skipping</description>
 
     <dependencies>
         <dependency>

batch/chunk-exception/src/main/java/org/javaee7/batch/chunk/exception/ChunkExceptionRecorder.java

@@ -7,4 +7,5 @@
  */
 public class ChunkExceptionRecorder {
     public static CountDownLatch chunkExceptionsCountDownLatch = new CountDownLatch(3);
+    public static int retryReadExecutions = 0;
 }

batch/chunk-exception/src/main/java/org/javaee7/batch/chunk/exception/MyInputRecord.java

@@ -39,14 +39,14 @@
  */
 package org.javaee7.batch.chunk.exception;
 
+import java.io.Serializable;
+
 /**
  * @author Arun Gupta
  */
-public class MyInputRecord {
+public class MyInputRecord implements Serializable {
     private int id;
-            
-    public MyInputRecord() { }
-    
+
     public MyInputRecord(int id) {
         this.id = id;
     }
@@ -58,7 +58,22 @@ public int getId() {
     public void setId(int id) {
         this.id = id;
     }
-    
+
+    @Override
+    public boolean equals(Object o) {
+        if (this == o) return true;
+        if (o == null || getClass() != o.getClass()) return false;
+
+        MyInputRecord that = (MyInputRecord) o;
+
+        return id == that.id;
+    }
+
+    @Override
+    public int hashCode() {
+        return id;
+    }
+
     @Override
     public String toString() {
         return "MyInputRecord: " + id;

batch/chunk-exception/src/main/java/org/javaee7/batch/chunk/exception/MyItemProcessor.java

@@ -51,11 +51,10 @@ public class MyItemProcessor implements ItemProcessor {
     @Override
     public Object processItem(Object t) {
         System.out.println("MyItemProcessor.processItem: " + t);
-        
-        if (((MyInputRecord)t).getId() == 6)
+
+        if (((MyInputRecord) t).getId() == 6)
             throw new NullPointerException();
-        
-//        return (t.getId() % 2 == 0) ? null : new MyOutputRecord(t.getId() * 2);
-        return new MyOutputRecord(((MyInputRecord)t).getId() * 2);
+
+        return new MyOutputRecord(((MyInputRecord) t).getId());
     }
 }

batch/chunk-exception/src/main/java/org/javaee7/batch/chunk/exception/MyItemReader.java

@@ -39,33 +39,55 @@
  */
 package org.javaee7.batch.chunk.exception;
 
-import java.io.Serializable;
-import java.util.StringTokenizer;
 import javax.batch.api.chunk.AbstractItemReader;
 import javax.inject.Named;
+import java.io.Serializable;
+import java.util.StringTokenizer;
 
 /**
  * @author Arun Gupta
  */
 @Named
 public class MyItemReader extends AbstractItemReader {
-    
+
     private StringTokenizer tokens;
-    
+
+    private MyInputRecord lastElement;
+    private boolean alreadyFailed;
+
     @Override
-    public void open(Serializable c) {
+    public void open(Serializable checkpoint) {
         tokens = new StringTokenizer("1,2,3,4,5,6,7,8,9,10", ",");
+
+        // This will place the nextToken into the last batch checkpoint. Called on exception retry.
+        if (checkpoint != null) {
+            while (!Integer.valueOf(tokens.nextToken()).equals(((MyInputRecord) checkpoint).getId())) {
+                System.out.println("Skipping already read elements");
+            }
+        }
     }
-    
+
     @Override
     public Object readItem() {
         if (tokens.hasMoreTokens()) {
             int token = Integer.valueOf(tokens.nextToken());
-            if (token == 3)
-                throw new IllegalArgumentException();
-                
-            return new MyInputRecord(token);
+
+            // Simulate a read exception when the token is equal to 5. Do it once only.
+            if (token == 5 && !alreadyFailed) {
+                alreadyFailed = true;
+                throw new IllegalArgumentException("Could not read record");
+            }
+
+            lastElement = new MyInputRecord(token);
+            System.out.println("MyItemReader.readItem " + lastElement);
+            return lastElement;
         }
         return null;
     }
+
+    @Override
+    public Serializable checkpointInfo() throws Exception {
+        // This is used internally by batch to stop the retry. Remember to implement equals on the read elements.
+        return lastElement;
+    }
 }

batch/chunk-exception/src/main/java/org/javaee7/batch/chunk/exception/MyItemWriter.java

@@ -39,9 +39,9 @@
  */
 package org.javaee7.batch.chunk.exception;
 
-import java.util.List;
 import javax.batch.api.chunk.AbstractItemWriter;
 import javax.inject.Named;
+import java.util.List;
 
 /**
  * @author Arun Gupta
@@ -51,7 +51,7 @@ public class MyItemWriter extends AbstractItemWriter {
 
     @Override
     public void writeItems(List list) {
-        if (list.contains(new MyOutputRecord(2))) {
+        if (list.contains(new MyOutputRecord(8))) {
             throw new IllegalArgumentException();
         }
 

batch/chunk-exception/src/main/java/org/javaee7/batch/chunk/exception/MyOutputRecord.java

@@ -39,14 +39,14 @@
  */
 package org.javaee7.batch.chunk.exception;
 
+import java.io.Serializable;
+
 /**
  * @author Arun Gupta
  */
-public class MyOutputRecord {
+public class MyOutputRecord implements Serializable {
     private int id;
-            
-    public MyOutputRecord() { }
-    
+
     public MyOutputRecord(int id) {
         this.id = id;
     }
@@ -58,11 +58,6 @@ public int getId() {
     public void setId(int id) {
         this.id = id;
     }
-    
-    @Override
-    public String toString() {
-        return "MyOutputRecord: " + id;
-    }
 
     @Override
     public boolean equals(Object o) {
@@ -78,4 +73,9 @@ public boolean equals(Object o) {
     public int hashCode() {
         return id;
     }
+
+    @Override
+    public String toString() {
+        return "MyOutputRecord: " + id;
+    }
 }

batch/chunk-exception/src/main/java/org/javaee7/batch/chunk/exception/MyRetryProcessorListener.java

@@ -0,0 +1,15 @@
+package org.javaee7.batch.chunk.exception;
+
+import javax.batch.api.chunk.listener.RetryProcessListener;
+import javax.inject.Named;
+
+/**
+ * @author Roberto Cortez
+ */
+@Named
+public class MyRetryProcessorListener implements RetryProcessListener {
+    @Override
+    public void onRetryProcessException(Object item, Exception ex) throws Exception {
+        System.out.println("MyRetryProcessorListener.onRetryProcessException");
+    }
+}

batch/chunk-exception/src/main/java/org/javaee7/batch/chunk/exception/MyRetryReadListener.java

@@ -0,0 +1,17 @@
+package org.javaee7.batch.chunk.exception;
+
+import javax.batch.api.chunk.listener.RetryReadListener;
+import javax.inject.Named;
+
+/**
+ * @author Roberto Cortez
+ */
+@Named
+public class MyRetryReadListener implements RetryReadListener {
+    @Override
+    public void onRetryReadException(Exception ex) throws Exception {
+        ChunkExceptionRecorder.retryReadExecutions++;
+        ChunkExceptionRecorder.chunkExceptionsCountDownLatch.countDown();
+        System.out.println("MyRetryReadListener.onRetryReadException " + ex.getMessage());
+    }
+}

batch/chunk-exception/src/main/java/org/javaee7/batch/chunk/exception/MyRetryWriteListener.java

@@ -0,0 +1,16 @@
+package org.javaee7.batch.chunk.exception;
+
+import javax.batch.api.chunk.listener.RetryWriteListener;
+import javax.inject.Named;
+import java.util.List;
+
+/**
+ * @author Roberto Cortez
+ */
+@Named
+public class MyRetryWriteListener implements RetryWriteListener {
+    @Override
+    public void onRetryWriteException(List<Object> items, Exception ex) throws Exception {
+        System.out.println("MyRetryWriteListener.onRetryWriteException");
+    }
+}