Merge pull request #5 from lerna-stack/fix-hangup-caused-by-sequential-execution-id-extractor

SequentialExecutionIdExtractor に関連するバッチ処理が終了しない問題を解決する

Author: negokaz

CHANGELOG.md

@@ -0,0 +1,16 @@
+# 変更履歴
+
+nablarch-fw-batch-parallelizable に関する注目すべき変更はこのファイルで文書化されます。
+
+このファイルの書き方に関する推奨事項については、[Keep a Changelog](https://keepachangelog.com/ja/1.0.0/) を確認してください。
+[Semantic Versioning](https://semver.org/spec/v2.0.0.html) を採用しています。
+
+## Unreleased
+
+### FIXED
+- `ControllableParallelExecutor.sequentialExecutionId` で `null` を返すとバッチ処理が終了しない [#4](https://github.com/lerna-stack/nablarch-fw-batch-parallelizable/issues/4)
+
+## [v1.2.0] - 2021-3-25
+[v1.2.0]: https://github.com/lerna-stack/nablarch-fw-batch-parallelizable/releases/tag/v1.2.0
+
+- Initial release

README.md

@@ -238,6 +238,9 @@ I/Oバウンドな処理の場合はこの値を⼤きくするにともないCP
 - [アーキテクチャ](docs/architecture.md)
 - [開発ガイド](docs/developers-guide.md)
 
+## Changelog
+注目すべき変更は [CHANGELOG](CHANGELOG.md) から確認できます。
+
 ## License
 
 `nablarch-fw-batch-parallelizable` is released under the terms of the [Apache License Version 2.0](LICENSE.txt).

src/main/java/lerna/nablarch/batch/parallelizable/handler/ControllableParallelExecutionHandler.java

@@ -404,14 +404,40 @@ public MultiStatus handle(Object data, ExecutionContext context) {
 
         final Source<Object, NotUsed> partitionedSource =
                 DataReaderSource.create(context).toMat(
-                    PartitionHub.ofStateful(Object.class, () -> (info, d) -> {
-                        final int executionId = Math.abs(executor.sequentialExecutionId(d).getAsInt() % parallelism);
-                        // 下流でエラーが発生して killSwitch が発動した場合は size が減少して executionId と consumer の ID がずれる。
-                        // そのような状況になると同じ executionId の要素が同時実行されてしまうので、size が変ったときは全ての要素を無視する。
-                        // consumerId として負の値を返すと無視される
-                        return info.size() == parallelism ? info.consumerIdByIdx(executionId) : -1;
-                    }, parallelism, handleDataBufferSize),
-                    Keep.right()
+                        PartitionHub.ofStateful(Object.class, () -> (info, d) -> {
+                            // Consumer identifier として 負の値を返すと要素はドロップされる
+                            // See also https://doc.akka.io/api/akka/2.6.9/akka/stream/javadsl/PartitionHub$.html
+                            final int CONSUMER_IDENTIFIER_TO_DROP_THE_ELEMENT = -1;
+                            try {
+                                final ControllableParallelExecutor.SequentialExecutionIdExtractor executionIdExtractor =
+                                        executor.sequentialExecutionId(d);
+                                if (executionIdExtractor == null) {
+                                    // SequentialExecutionIdExtractor が null の場合には、データを振り分けることができない。
+                                    // 自動的に復旧することは難しいため、Stream 全体を中止し、処理を継続できないことをユーザに通知する。
+                                    final String executorInfo =
+                                            String.format("%s@%s", executor.getClass().getName(), Integer.toHexString(executor.hashCode()));
+                                    final String message =
+                                            String.format(
+                                                    "%s: ControllableParallelExecutor.sequentialExecutionId(element) should not return null.",
+                                                    executorInfo
+                                            );
+                                    killSwitch.abort(new IllegalStateException(message));
+                                    return CONSUMER_IDENTIFIER_TO_DROP_THE_ELEMENT;
+                                }
+                                final int executionId = Math.abs(executionIdExtractor.getAsInt() % parallelism);
+                                // 下流でエラーが発生して killSwitch が発動した場合は size が減少して executionId と consumer の ID がずれる。
+                                // そのような状況になると同じ executionId の要素が同時実行されてしまうので、size が変ったときは全ての要素を無視する。
+                                return info.size() == parallelism ? info.consumerIdByIdx(executionId) : CONSUMER_IDENTIFIER_TO_DROP_THE_ELEMENT;
+                            } catch (Throwable cause) {
+                                // 次のような理由でデータを振り分ける処理がうまくいかない場合がある。
+                                //  * SequentialExecutionIdExtractor を取得する際に例外が発生した
+                                //  * SequentialExecutionIdExtractor.getAsInt で例外が発生した
+                                // 自動的に復旧することは難しいため、Stream 全体を中止し、処理を継続できないことをユーザに通知する。
+                                killSwitch.abort(cause);
+                                return CONSUMER_IDENTIFIER_TO_DROP_THE_ELEMENT;
+                            }
+                        }, parallelism, handleDataBufferSize),
+                        Keep.right()
                 ).run(actorSystem);
 
         final Supplier<Flow<TransactionalSession, Result, CompletionStage<MultiStatus>>> handleFlow =

src/main/java/lerna/nablarch/batch/parallelizable/handler/ControllableParallelExecutor.java

@@ -25,18 +25,27 @@ public interface ControllableParallelExecutor<IN> extends Handler<IN, Result>, D
     /**
      * 入力データをどの単位で逐次処理するかを定義するためのメソッド。
      *
+     * <p>
+     * このメソッドの実装で例外を throw してはならない。
+     * このメソッドで例外が throw された場合、バッチ処理は中止される。
+     * このメソッドで受け取った element 以外の要素も処理されない可能性がある。
+     *
      * @param element 処理対象の入力データ
-     * @return 逐次処理する{@link SequentialExecutionIdExtractor}
+     * @return 逐次処理する {@link SequentialExecutionIdExtractor}。
+     *         null を返してはならない。このメソッドが null を返す場合、バッチ処理は中止される。
+     *         このメソッドで受け取った element 以外の要素も処理されない可能性がある。
      */
     SequentialExecutionIdExtractor sequentialExecutionId(IN element);
 
     /**
-     * <p>
      * 逐次処理する単位を定義する関数。
-     * </p>
+     *
      * <p>
      * 入力データの種類を識別する ID は int で表される。同じ ID を持つ入力データはそれぞれが並列に処理されず、逐次的に処理されることが保証される。
-     * </p>
+     *
+     * <p>
+     * {@link #getAsInt} で例外を throw してはならない。
+     * 例外が throw された場合、バッチ処理は中止される。
      *
      * @since 2018/04/09
      */

src/test/java/lerna/nablarch/batch/parallelizable/handler/ControllableParallelExecutionHandlerTest.java

@@ -210,6 +210,92 @@ public Result handle(String s, ExecutionContext executionContext) {
         assertThat(result.isSuccess(), is(false));
     }
 
+    @Test
+    public void ハンドラのSequentialExecutionIdExtractorがnullであれば例外IllegalStateExceptionが親ハンドラにスローされる() {
+        int parallelism = 5;
+
+        ControllableParallelExecutionHandler handler = new ControllableParallelExecutionHandler();
+        handler.setParallelism(parallelism);
+        handler.setConnectionFactory((s) -> new TestTransactionManagerConnectionBase());
+        handler.setTransactionFactory(s -> new TestTransactionBase());
+
+        ControllableParallelExecutor<String> executor = new TestControllableParallelExecutorBase() {
+            @Override
+            public SequentialExecutionIdExtractor sequentialExecutionId(String element) {
+                return null;
+            }
+        };
+
+        ExecutionContext context = new ExecutionContext();
+        context.addHandler(handler);
+        context.addHandler(executor);
+
+        // 親ハンドラは handleNext から例外を catch できる
+        final IllegalStateException exception =
+                assertThrows(IllegalStateException.class, () -> context.handleNext("input"));
+        final String executorInfo =
+                String.format("%s@%s", executor.getClass().getName(), Integer.toHexString(executor.hashCode()));
+        final String expectMessage =
+                String.format("%s: ControllableParallelExecutor.sequentialExecutionId(element) should not return null.", executorInfo);
+        assertThat(exception.getMessage(), is(expectMessage));
+
+    }
+
+    @Test
+    public void ハンドラのSequentialExecutionIdExtractor取得で例外が発生したら親ハンドラにその例外がリスローされる() {
+        int parallelism = 5;
+
+        ControllableParallelExecutionHandler handler = new ControllableParallelExecutionHandler();
+        handler.setParallelism(parallelism);
+        handler.setConnectionFactory((s) -> new TestTransactionManagerConnectionBase());
+        handler.setTransactionFactory(s -> new TestTransactionBase());
+
+        ExecutionContext context = new ExecutionContext();
+        context.addHandler(handler);
+        context.addHandler(new TestControllableParallelExecutorBase() {
+            @Override
+            public SequentialExecutionIdExtractor sequentialExecutionId(String element) {
+                throw new ArithmeticException(String.format("Could not create SequentialExecutionIdExtractor from the element(%s).", element));
+            }
+        });
+
+        // 親ハンドラは handleNext から例外を catch できる
+        final ArithmeticException exception =
+                assertThrows(ArithmeticException.class, () -> context.handleNext("input"));
+        assertThat(exception.getMessage(), is("Could not create SequentialExecutionIdExtractor from the element(1)."));
+
+    }
+
+    @Test
+    public void ハンドラのSequentialExecutionIdExtractorで例外が発生したら親ハンドラにその例外がリスローされる() {
+        int parallelism = 5;
+
+        ControllableParallelExecutionHandler handler = new ControllableParallelExecutionHandler();
+        handler.setParallelism(parallelism);
+        handler.setConnectionFactory((s) -> new TestTransactionManagerConnectionBase());
+        handler.setTransactionFactory(s -> new TestTransactionBase());
+
+        ExecutionContext context = new ExecutionContext();
+        context.addHandler(handler);
+        context.addHandler(new TestControllableParallelExecutorBase() {
+            @Override
+            public SequentialExecutionIdExtractor sequentialExecutionId(String element) {
+                return new SequentialExecutionIdExtractor() {
+                    @Override
+                    public int getAsInt() {
+                        throw new ArithmeticException(String.format("Could not calculate ID from the element(%s).", element));
+                    }
+                };
+            }
+        });
+
+        // 親ハンドラは handleNext から例外を catch できる
+        final ArithmeticException exception =
+                assertThrows(ArithmeticException.class, () -> context.handleNext("input"));
+        assertThat(exception.getMessage(), is("Could not calculate ID from the element(1)."));
+
+    }
+
     @Test
     public void executionId毎にcommitIntervalを迎えたときと終了時に1度コミットされる() {
         int numberOfData = 100;