BAEL-2168 Java EE 7 Batch Processing (#5645)

* jberet batch

* Batch Understanding

* partition

* exception

* some more changes
This commit is contained in:
Amitabh Mandal
2018-11-25 11:01:22 +01:00
committed by Grzegorz Piwowarek
parent 2f9fee0391
commit f784c3bb79
39 changed files with 1473 additions and 414 deletions
+521 -414
View File
File diff suppressed because it is too large Load Diff
@@ -0,0 +1,11 @@
package com.baeldung.batch.understanding;
import javax.batch.api.chunk.listener.SkipReadListener;
import javax.inject.Named;
@Named
public class ChunkExceptionSkipReadListener implements SkipReadListener {
@Override
public void onSkipReadItem(Exception e) throws Exception {
}
}
@@ -0,0 +1,12 @@
package com.baeldung.batch.understanding;
import javax.batch.api.chunk.AbstractCheckpointAlgorithm;
import javax.inject.Named;
@Named
public class CustomCheckPoint extends AbstractCheckpointAlgorithm {
@Override
public boolean isReadyToCheckpoint() throws Exception {
return SimpleChunkItemReader.COUNT % 5 == 0;
}
}
@@ -0,0 +1,14 @@
package com.baeldung.batch.understanding;
import javax.batch.api.Decider;
import javax.batch.runtime.StepExecution;
import javax.inject.Named;
@Named
public class DeciderJobSequence implements Decider {
@Override
public String decide(StepExecution[] ses) throws Exception {
return "nothing";
}
}
@@ -0,0 +1,20 @@
package com.baeldung.batch.understanding;
import javax.batch.api.AbstractBatchlet;
import javax.batch.api.BatchProperty;
import javax.batch.runtime.BatchStatus;
import javax.inject.Inject;
import javax.inject.Named;
@Named
public class InjectSimpleBatchLet extends AbstractBatchlet {
@Inject
@BatchProperty(name = "name")
private String nameString;
@Override
public String process() throws Exception {
System.out.println("Value passed in = " + nameString);
return BatchStatus.COMPLETED.toString();
}
}
@@ -0,0 +1,13 @@
package com.baeldung.batch.understanding;
import javax.batch.api.AbstractBatchlet;
import javax.batch.runtime.BatchStatus;
import javax.inject.Named;
@Named
public class SimpleBatchLet extends AbstractBatchlet {
@Override
public String process() throws Exception {
return BatchStatus.FAILED.toString();
}
}
@@ -0,0 +1,12 @@
package com.baeldung.batch.understanding;
import javax.batch.api.chunk.ItemProcessor;
import javax.inject.Named;
@Named
public class SimpleChunkItemProcessor implements ItemProcessor {
@Override
public Integer processItem(Object t) {
return ((Integer) t).intValue() % 2 == 0 ? null : ((Integer) t).intValue();
}
}
@@ -0,0 +1,27 @@
package com.baeldung.batch.understanding;
import java.io.Serializable;
import java.util.StringTokenizer;
import javax.batch.api.chunk.AbstractItemReader;
import javax.inject.Named;
@Named
public class SimpleChunkItemReader extends AbstractItemReader {
private StringTokenizer tokens;
public static int COUNT = 0;
@Override
public Integer readItem() throws Exception {
if (tokens.hasMoreTokens()) {
COUNT++;
String tempTokenize = tokens.nextToken();
return Integer.valueOf(tempTokenize);
}
return null;
}
@Override
public void open(Serializable checkpoint) throws Exception {
tokens = new StringTokenizer("1,2,3,4,5,6,7,8,9,10", ",");
}
}
@@ -0,0 +1,31 @@
package com.baeldung.batch.understanding;
import java.io.Serializable;
import java.util.StringTokenizer;
import javax.batch.api.chunk.AbstractItemReader;
import javax.inject.Named;
@Named
public class SimpleChunkItemReaderError extends AbstractItemReader {
private StringTokenizer tokens;
public static int COUNT = 0;
@Override
public Integer readItem() throws Exception {
if (tokens.hasMoreTokens()) {
COUNT++;
int token = Integer.valueOf(tokens.nextToken());
if (token == 3) {
throw new RuntimeException("Something happened");
}
return Integer.valueOf(token);
}
return null;
}
@Override
public void open(Serializable checkpoint) throws Exception {
tokens = new StringTokenizer("1,2,3,4,5,6,7,8,9,10", ",");
}
}
@@ -0,0 +1,13 @@
package com.baeldung.batch.understanding;
import java.util.List;
import javax.batch.api.chunk.AbstractItemWriter;
import javax.inject.Named;
@Named
public class SimpleChunkWriter extends AbstractItemWriter {
@Override
public void writeItems(List<Object> items) throws Exception {
}
}
@@ -0,0 +1,41 @@
package com.baeldung.batch.understanding.exception;
import java.io.Serializable;
public class MyInputRecord implements Serializable {
private int id;
public MyInputRecord(int id) {
this.id = id;
}
public int getId() {
return id;
}
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;
}
}
@@ -0,0 +1,15 @@
package com.baeldung.batch.understanding.exception;
import javax.batch.api.chunk.ItemProcessor;
import javax.inject.Named;
@Named
public class MyItemProcessor implements ItemProcessor {
@Override
public Object processItem(Object t) {
if (((MyInputRecord) t).getId() == 6) {
throw new NullPointerException();
}
return new MyOutputRecord(((MyInputRecord) t).getId());
}
}
@@ -0,0 +1,42 @@
package com.baeldung.batch.understanding.exception;
import javax.batch.api.chunk.AbstractItemReader;
import javax.inject.Named;
import java.io.Serializable;
import java.util.StringTokenizer;
@Named
public class MyItemReader extends AbstractItemReader {
private StringTokenizer tokens;
private MyInputRecord lastElement;
private boolean alreadyFailed;
@Override
public void open(Serializable checkpoint) {
tokens = new StringTokenizer("1,2,3,4,5,6,7,8,9,10", ",");
if (checkpoint != null) {
while (!Integer.valueOf(tokens.nextToken())
.equals(((MyInputRecord) checkpoint).getId())) {
}
}
}
@Override
public Object readItem() {
if (tokens.hasMoreTokens()) {
int token = Integer.valueOf(tokens.nextToken());
if (token == 5 && !alreadyFailed) {
alreadyFailed = true;
throw new IllegalArgumentException("Could not read record");
}
lastElement = new MyInputRecord(token);
return lastElement;
}
return null;
}
@Override
public Serializable checkpointInfo() throws Exception {
return lastElement;
}
}
@@ -0,0 +1,19 @@
package com.baeldung.batch.understanding.exception;
import javax.batch.api.chunk.AbstractItemWriter;
import javax.inject.Named;
import java.util.List;
@Named
public class MyItemWriter extends AbstractItemWriter {
private static int retries = 0;
@Override
public void writeItems(List list) {
if (retries <= 3 && list.contains(new MyOutputRecord(8))) {
retries++;
throw new UnsupportedOperationException();
}
}
}
@@ -0,0 +1,39 @@
package com.baeldung.batch.understanding.exception;
import java.io.Serializable;
public class MyOutputRecord implements Serializable {
private int id;
public MyOutputRecord(int id) {
this.id = id;
}
public int getId() {
return id;
}
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;
MyOutputRecord that = (MyOutputRecord) o;
return id == that.id;
}
@Override
public int hashCode() {
return id;
}
@Override
public String toString() {
return "MyOutputRecord: " + id;
}
}
@@ -0,0 +1,11 @@
package com.baeldung.batch.understanding.exception;
import javax.batch.api.chunk.listener.RetryProcessListener;
import javax.inject.Named;
@Named
public class MyRetryProcessorListener implements RetryProcessListener {
@Override
public void onRetryProcessException(Object item, Exception ex) throws Exception {
}
}
@@ -0,0 +1,11 @@
package com.baeldung.batch.understanding.exception;
import javax.batch.api.chunk.listener.RetryReadListener;
import javax.inject.Named;
@Named
public class MyRetryReadListener implements RetryReadListener {
@Override
public void onRetryReadException(Exception ex) throws Exception {
}
}
@@ -0,0 +1,12 @@
package com.baeldung.batch.understanding.exception;
import javax.batch.api.chunk.listener.RetryWriteListener;
import javax.inject.Named;
import java.util.List;
@Named
public class MyRetryWriteListener implements RetryWriteListener {
@Override
public void onRetryWriteException(List<Object> items, Exception ex) throws Exception {
}
}
@@ -0,0 +1,11 @@
package com.baeldung.batch.understanding.exception;
import javax.batch.api.chunk.listener.SkipProcessListener;
import javax.inject.Named;
@Named
public class MySkipProcessorListener implements SkipProcessListener {
@Override
public void onSkipProcessItem(Object t, Exception e) throws Exception {
}
}
@@ -0,0 +1,11 @@
package com.baeldung.batch.understanding.exception;
import javax.batch.api.chunk.listener.SkipReadListener;
import javax.inject.Named;
@Named
public class MySkipReadListener implements SkipReadListener {
@Override
public void onSkipReadItem(Exception e) throws Exception {
}
}
@@ -0,0 +1,13 @@
package com.baeldung.batch.understanding.exception;
import javax.batch.api.chunk.listener.SkipWriteListener;
import javax.inject.Named;
import java.util.List;
@Named
public class MySkipWriteListener implements SkipWriteListener {
@Override
public void onSkipWriteItem(List list, Exception e) throws Exception {
list.remove(new MyOutputRecord(2));
}
}
@@ -0,0 +1,12 @@
<?xml version="1.0" encoding="UTF-8"?>
<job id="customCheckPoint" xmlns="http://xmlns.jcp.org/xml/ns/javaee" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://xmlns.jcp.org/xml/ns/javaee http://xmlns.jcp.org/xml/ns/javaee/jobXML_1_0.xsd" version="1.0">
<step id="firstChunkStep" >
<chunk item-count="3" checkpoint-policy="custom">
<reader ref="simpleChunkItemReader"/>
<processor ref="simpleChunkItemProcessor"/>
<writer ref="simpleChunkWriter"/>
<checkpoint-algorithm ref="customCheckPoint"/>
</chunk>
</step>
</job>
@@ -0,0 +1,18 @@
<?xml version="1.0" encoding="UTF-8"?>
<job id="decideJobSequence" xmlns="http://xmlns.jcp.org/xml/ns/javaee"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://xmlns.jcp.org/xml/ns/javaee http://xmlns.jcp.org/xml/ns/javaee/jobXML_1_0.xsd"
version="1.0">
<step id="firstBatchStepStep1" next="firstDecider">
<batchlet ref="simpleBatchLet" />
</step>
<decision id="firstDecider" ref="deciderJobSequence">
<next on="nothing" to="firstBatchStepStep3"/>
</decision>
<step id="firstBatchStepStep2">
<batchlet ref="simpleBatchLet" />
</step>
<step id="firstBatchStepStep3">
<batchlet ref="simpleBatchLet" />
</step>
</job>
@@ -0,0 +1,21 @@
<?xml version="1.0" encoding="UTF-8"?>
<job id="flowJobSequence" xmlns="http://xmlns.jcp.org/xml/ns/javaee"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://xmlns.jcp.org/xml/ns/javaee http://xmlns.jcp.org/xml/ns/javaee/jobXML_1_0.xsd"
version="1.0">
<flow id="flow1" next="firstBatchStepStep3">
<step id="firstChunkStepStep1" next="firstBatchStepStep2">
<chunk item-count="3">
<reader ref="simpleChunkItemReader" />
<processor ref="simpleChunkItemProcessor" />
<writer ref="simpleChunkWriter" />
</chunk>
</step>
<step id="firstBatchStepStep2">
<batchlet ref="simpleBatchLet" />
</step>
</flow>
<step id="firstBatchStepStep3">
<batchlet ref="simpleBatchLet" />
</step>
</job>
@@ -0,0 +1,13 @@
<?xml version="1.0" encoding="UTF-8"?>
<job id="injectSimpleBatchLet" xmlns="http://xmlns.jcp.org/xml/ns/javaee"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://xmlns.jcp.org/xml/ns/javaee http://xmlns.jcp.org/xml/ns/javaee/jobXML_1_0.xsd"
version="1.0">
<step id="firstStep" >
<batchlet ref="injectSimpleBatchLet">
<properties>
<property name="name" value="helloThere" />
</properties>
</batchlet>
</step>
</job>
@@ -0,0 +1,24 @@
<?xml version="1.0" encoding="UTF-8"?>
<job id="injectSimpleBatchLet"
xmlns="http://xmlns.jcp.org/xml/ns/javaee"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://xmlns.jcp.org/xml/ns/javaee http://xmlns.jcp.org/xml/ns/javaee/jobXML_1_0.xsd"
version="1.0">
<step id="firstStep">
<batchlet ref="injectSimpleBatchLet">
<properties>
<property name="name" value="#{partitionPlan['name']}" />
</properties>
</batchlet>
<partition>
<plan partitions="2">
<properties partition="0">
<property name="name" value="firstPartition" />
</properties>
<properties partition="1">
<property name="name" value="secondpPartition" />
</properties>
</plan>
</partition>
</step>
</job>
@@ -0,0 +1,9 @@
<?xml version="1.0" encoding="UTF-8"?>
<job id="simpleBatchLet" xmlns="http://xmlns.jcp.org/xml/ns/javaee"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://xmlns.jcp.org/xml/ns/javaee http://xmlns.jcp.org/xml/ns/javaee/jobXML_1_0.xsd"
version="1.0">
<step id="firstStep" >
<batchlet ref="simpleBatchLet"/>
</step>
</job>
@@ -0,0 +1,11 @@
<?xml version="1.0" encoding="UTF-8"?>
<job id="simpleChunk" xmlns="http://xmlns.jcp.org/xml/ns/javaee" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://xmlns.jcp.org/xml/ns/javaee http://xmlns.jcp.org/xml/ns/javaee/jobXML_1_0.xsd" version="1.0">
<step id="firstChunkStep" >
<chunk item-count="3">
<reader ref="simpleChunkItemReader"/>
<processor ref="simpleChunkItemProcessor"/>
<writer ref="simpleChunkWriter"/>
</chunk>
</step>
</job>
@@ -0,0 +1,24 @@
<?xml version="1.0" encoding="UTF-8"?>
<job id="simpleErrorChunk"
xmlns="http://xmlns.jcp.org/xml/ns/javaee"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://xmlns.jcp.org/xml/ns/javaee http://xmlns.jcp.org/xml/ns/javaee/jobXML_1_0.xsd"
version="1.0">
<step id="firstErrorChunkStep" >
<chunk item-count="3">
<reader ref="simpleChunkItemReaderError"/>
<processor ref="simpleChunkItemProcessor"/>
<writer ref="simpleChunkWriter"/>
</chunk>
</step>
<step id="firstErrorSkipChunkStep" >
<chunk item-count="3" skip-limit="3">
<reader ref="simpleChunkItemReaderError"/>
<processor ref="simpleChunkItemProcessor"/>
<writer ref="simpleChunkWriter"/>
<skippable-exception-classes>
<include class="java.lang.RuntimeException"/>
</skippable-exception-classes>
</chunk>
</step>
</job>
@@ -0,0 +1,26 @@
<job id="simpleErrorSkipChunk" xmlns="http://xmlns.jcp.org/xml/ns/javaee" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://xmlns.jcp.org/xml/ns/javaee http://xmlns.jcp.org/xml/ns/javaee/jobXML_1_0.xsd" version="1.0">
<step id="errorStep" >
<listeners>
<listener ref="mySkipReadListener"/>
<listener ref="mySkipProcessorListener"/>
<listener ref="mySkipWriteListener"/>
<listener ref="myRetryReadListener"/>
<listener ref="myRetryProcessorListener"/>
<listener ref="myRetryWriteListener"/>
</listeners>
<chunk checkpoint-policy="item" item-count="3" skip-limit="3" retry-limit="3">
<reader ref="myItemReader"/>
<processor ref="myItemProcessor"/>
<writer ref="myItemWriter"/>
<skippable-exception-classes>
<include class="java.lang.RuntimeException"/>
<include class="java.lang.UnsupportedOperationException"/>
</skippable-exception-classes>
<retryable-exception-classes>
<include class="java.lang.IllegalArgumentException"/>
<include class="java.lang.UnsupportedOperationException"/>
</retryable-exception-classes>
</chunk>
</step>
</job>

Some files were not shown because too many files have changed in this diff Show More