Java EE 7 検証環境構築(12) jBatch 簡易サンプル作成と Arquillian でユニットテスト

jBatchの概要を押さえて、jBatch用の環境(Java EE7対応、EARプロジェクトの生成)を行ったので、Java EE 7 チュートリアルから、バッチの簡単な例を試してみる。(日本語訳サイトはこちら)

1.簡単な使用例

ジョブ定義言語(JSL)をXMLで作成する。XMLファイルは、META-INF/batch-jobs 配下に格納する。

今回の流れでは、batch-ejb プロジェクトの、META-INF/batch-jobs 配下に、simplejob.xml という名前で作成した。

チャンクステップとタスクステップのジョブ定義。input_file、output_file パラメータを具体的なファイルに設定してみる。

<?xml version="1.0" encoding="UTF-8"?>
<!-- http://docs.oracle.com/javaee/7/tutorial/doc/batch-processing003.htm -->
<job id="simplejob" xmlns="http://xmlns.jcp.org/xml/ns/javaee" 
                    version="1.0">
  <properties>
    <property name="input_file" value="c:\work\test.txt"/>
    <property name="output_file" value="c:\work\test_out.txt"/>
  </properties>
  <step id="mychunk" next="mytask">
    <chunk>
      <reader ref="MyReader"></reader>
      <processor ref="MyProcessor"></processor>
      <writer ref="MyWriter"></writer>
    </chunk>
  </step>
  <step id="mytask">
    <batchlet ref="MyBatchlet"></batchlet>
    <end on="COMPLETED"/>
  </step>
</job>

2.作成するクラス

ジョブ定義ファイルで参照するクラスを作成する。

クラス	役割
MyCheckpoint	チェックポイント。処理済みの位置を保持する。
MyReader	アイテムを読み込む。ジョブが再開された場合は、チェックポイントから読み込み位置を取得する。
MyProcesser	読み込んだアイテムを処理。
MyWriter	処理済みアイテムを出力する

2.1 チャンクステップ

2.1.1 Checkpoint クラス

多くの場合、チャンク指向ステップのためにチェックポイントを実装すべき
以下のクラスは、処理されたテキストファイルの行を保持する

@SuppressWarnings("serial")
public class MyCheckpoint implements Serializable {
    
    private long lineNum = 0;
    
    public void increase() { 
        lineNum++; 
    }
    public long getLineNum() { 
        return lineNum; 
    }
}

2.1.2 ItemReader クラス

ItemReader実装クラスは、ジョブが再開された場合には与えられたチェックポイントから入力ファイルの読み込みを続行する
アイテムはテキストファイルの各行に対応
より複雑なシナリオでは、アイテムは独自のJavaクラスで入力元はデータベース等になる
@Named で、MyReader という名前をつけ、定義ファイルから名前で参照

@Dependent
@Named("MyReader")
public class MyReader implements javax.batch.api.chunk.ItemReader {
    private MyCheckpoint checkpoint;
    private BufferedReader breader;
    
    @Inject
    JobContext jobCtx;
    
    public MyReader() {}

    /*
     * Itemを読むためのReaderを準備する
     * 仮引数は、ジョブインスタンスの最終チェックポイント
     */
    @Override
    public void open(Serializable ckpt) throws Exception {
        if (ckpt == null) {
            checkpoint = new MyCheckpoint();
        } else {
            checkpoint = (MyCheckpoint) ckpt;
        }
        String fileName = jobCtx.getProperties()
                                .getProperty("input_file");
        breader = new BufferedReader(new FileReader(fileName));
        for (long i = 0; i < checkpoint.getLineNum(); i++) {
            breader.readLine();
        }
    }

    @Override
    public void close() throws Exception {
        breader.close();
    }

    @Override
    public Object readItem() throws Exception {
        String line = breader.readLine();
        return line;
    }

    /* 
     * Readerに対して、、現在のチェックポイントを返す
     */
    @Override
    public Serializable checkpointInfo() throws Exception {
        return checkpoint;
    }
}

2.1.3 ItemProcessor クラス

読み込んだアイテムを処理する
このケースでは、大文字に変換するだけ
@Named で、MyProcessor という名前をつけ、定義ファイルから名前で参照

@Dependent
@Named("MyProcessor")
public class MyProcessor implements javax.batch.api.chunk.ItemProcessor {
    public MyProcessor() {}

    @Override
    public Object processItem(Object obj) throws Exception {
        String line = (String) obj;
        return line.toUpperCase();
    }
}

2.1.4 ItemWriter クラス

ItemWriterの実装クラスはItemProcessorを通過したデータをファイルへ出力
もしチェックポイントが無ければ出力ファイルを上書きし、そうでなければ、ファイルのEOFへの書き込みを再開
アイテムはチャンクごとに書き込まれる
@Named で、MyWriter という名前をつけ、定義ファイルから名前で参照

@Dependent
@Named("MyWriter")
public class MyWriter implements javax.batch.api.chunk.ItemWriter {
    private BufferedWriter bwriter;
    
    @Inject
    private JobContext jobCtx;

    @Override
    public void open(Serializable ckpt) throws Exception {
        String fileName = jobCtx.getProperties()
                                .getProperty("output_file");
        bwriter = new BufferedWriter(new FileWriter(fileName, 
                                                    (ckpt != null)));
    }

    @Override
    public void writeItems(List<Object> items) throws Exception {
        for (int i = 0; i < items.size(); i++) {
            String line = (String) items.get(i);
            bwriter.write(line);
            bwriter.newLine();
        }
    }

    @Override
    public Serializable checkpointInfo() throws Exception {
        return new MyCheckpoint();
    }

    @Override
    public void close() throws Exception {
        bwriter.close();
    }
}

2.2 タスクステップ

2.2.1 Batchlet

チャンク指向でうまく扱えないような処理を扱うことが出来る
この例では、タスクステップは出力ファイルのサイズを表示

@Dependent
@Named("MyBatchlet")
public class MyBatchlet implements javax.batch.api.Batchlet {
    
    @Inject
    private JobContext jobCtx;
    
    @Override
    public String process() throws Exception {
        String fileName = jobCtx.getProperties()
                                .getProperty("output_file");
        System.out.println(""+(new File(fileName)).length());
        return "COMPLETED";
    }

    @Override
    public void stop() throws Exception {
        // TODO Auto-generated method stub
    }
}

2. Arquillian でユニットテストを行う

さて、チュートリアルに従い、実装したので、Arquillian によるユニットテストを行おう。。。と思う。

https://github.com/javaee-samples/javaee7-samples

Github に Java EE 7 のサンプルが上がっているので、それを参考にArquillian のテストケースを作成。

最終的には、上記のような構成になる。サンプルのテストケースを利用するには、BatchTestHelper クラスが必要なので、

https://github.com/javaee-samples/javaee7-samples/blob/master/util/src/main/java/org/javaee7/util/BatchTestHelper.java

をプロジェクトに組み込む。

2.1 BatchTestHelper

/**
 * @author Roberto Cortez
 * @see https://github.com/javaee-samples/javaee7-samples/blob/master/util/src/main/java/org/javaee7/util/BatchTestHelper.java
 */
public final class BatchTestHelper {
    private static final int MAX_TRIES = 10;
    private static final int THREAD_SLEEP = 100;

    private BatchTestHelper() {
        throw new UnsupportedOperationException();
    }

    /**
     * We need to keep the test running because JobOperator runs the batch job in an asynchronous way, so the
     * JobExecution can be properly updated with the running job status.
     *
     * @param jobExecution
     *         the JobExecution of the job that is being runned on JobOperator.
     *
     * @throws InterruptedException thrown by Thread.sleep.
     */
    public static void keepTestAlive(JobExecution jobExecution) throws InterruptedException {
        int maxTries = 0;
        while (!jobExecution.getBatchStatus().equals(BatchStatus.COMPLETED)) {
            if (maxTries < MAX_TRIES) {
                maxTries++;
                Thread.sleep(THREAD_SLEEP);
            } else {
                break;
            }
        }
    }

    /**
     * Convert the Metric array contained in StepExecution to a key-value map for easy access to Metric parameters.
     *
     * @param metrics
     *         a Metric array contained in StepExecution.
     *
     * @return a map view of the metrics array.
     */
    public static Map<metric.metrictype   , long> getMetricsMap(Metric[] metrics) {
        Map<metric.metrictype   , long> metricsMap = new HashMap<>();
        for (Metric metric : metrics) {
            metricsMap.put(metric.getType(), metric.getValue());
        }
        return metricsMap;
    }
}

2.2 テストケース

テストケースとは言うが、まずは、Arquillian 上でバッチが起動出来ればよい。

@Deployment アノテーションをつけたデプロイメソッドで、Webアーカイブを作成する。

addPackage に今回のバッチ関連クラスが入ったパッケージを指定して、アーカイブに含めている
addAsWebInfResource の行では、CDIを利用するために、空のbean.xml をアーカイブに含めている
addAsResourceの行では、ジョブ定義ファイルをアーカイブに含めている。
addClass の行では、CDIにロガーなどを登録するクラスを登録
- https://www.typea.info/blog/index.php/2014/05/08/java_ee_7_5_jboss_toolscdi
System.out.print で war.toString を呼ぶことで、アーカイブの構成がコンソールに出力されるので、思ったような構成になっているかを確認することが出来る。

@Test を付与した、テストメソッドでは、JobOperator から、ジョブ定義XMLで定義した、”simplejob”を呼び出しバッチを処理している。

@RunWith(Arquillian.class)
public class SimpleJobTest {

    @Deployment
    public static WebArchive createDeployment() {
        WebArchive war = ShrinkWrap.create(WebArchive.class, "test.war")
                .addClass(BatchTestHelper.class)
                .addPackage("info.typea.tallarico.job.simplejob")
                .addAsWebInfResource(EmptyAsset.INSTANCE, ArchivePaths.create("beans.xml"))
                .addAsResource("META-INF/batch-jobs/simpleJob.xml")
                .addClass(Resources.class)
                ;
        System.out.println(war.toString(true));
        return war;
    }
    
    @Inject
    Logger logger;
    
    @Test
    public void simpleJobTest() throws Exception {
        JobOperator jobOperator = BatchRuntime.getJobOperator();
        Long executionId = jobOperator.start("simplejob", new Properties());
        JobExecution jobExecution = jobOperator.getJobExecution(executionId);
        
        BatchTestHelper.keepTestAlive(jobExecution);
        
        List stepExecutions = jobOperator.getStepExecutions(executionId);
        for (StepExecution stepExecution : stepExecutions) {
            String stepName = stepExecution.getStepName();
            
            switch(stepName) {
            case "mychunk":
            case "mytask":
                Map<metric.metrictype   , Long> metricsMap 
                    = BatchTestHelper.getMetricsMap(stepExecution.getMetrics());
                
                logger.info("===== STEP NAME:" + stepName + "=====");
                logger.info("READ COUNT:" + metricsMap.get(Metric.MetricType.READ_COUNT).longValue());
                logger.info("WRITE COUNT:" + metricsMap.get(Metric.MetricType.WRITE_COUNT).longValue());
                logger.info("COMMIT COUNT:" + metricsMap.get(Metric.MetricType.COMMIT_COUNT).longValue());
                break;
            }
        }
    }
}

その他の設定は、

Java EE 7 検証環境構築(1) WildFly + JBoss Tools で EARプロジェクトを作成し Arquillian でユニットテストをグリーンにするところまで

を参考に。