Apache Camel で `transacted()` と `split()` を組み合わせた際に StackOverflowError

（2021/07/21追記）**本問題は Camel 3.8 で修正された。**

* [Upgrading Camel 3.7 to 3.8 # Transactions and Multicast, Splitter, or Recipient List EIPs](https://camel.apache.org/manual/latest/camel-3x-upgrade-guide-3_8.html#_transactions_and_multicast_splitter_or_recipient_list_eips)
* [\[CAMEL-16103\] Stack size increases in split with transacted](https://issues.apache.org/jira/browse/CAMEL-16103)

---

## 概要
Apache Camel でDBから読み取ったレコードをループ処理したところ、**1000件近いデータ**は処理できず **StackOverflowError** が発生した。

調べていくと、 `split()` でループさせるだけなら問題ないが、それが `transacted()` より後である場合に発生することがわかった。

うまい解決ができていないが、再現と回避のために試したことをまとめておく。

### 回避策の案

* JVMのスタックサイズを増やす
	* お手軽、コード変更は不要
	* データが増えれば結局 StackOverflowError が発生する
* ループ内を別スレッドで処理する
	* 変更は1〜2行で済む
	* **トランザクションから外れる**ので、DBを読み書きしているような処理では採用できない
* `split()` を多重にして、一段あたりのループ回数を抑える
	* ループ内の処理も合わせて修正する必要がある
	* データが増えれば結局 StackOverflowError が発生する<br>（とはいえ1重で800回だったのなら、3重で1500万回のループには耐えられる見込み[^calc-3]）

[^calc-3]: 各段250回とすれば、全体のループは 250 \*\* 3 = 15625000回、スタック消費は 250 \* 3 = 750 （ ≦ 800 ）。

### テスト環境
以前に勉強用に作ったサンプルを流用したため、環境も同じ。
https://qiita.com/HMMNRST/items/da89ce94c203633a8fb3

* Java: 1.8.0_191
* Maven: 3.6.3
* Camel: 3.2.0
	* Spring: 5.2.5.RELEASE
	* Spring Boot: 2.2.6.RELEASE
* MySQL: 5.7.30

## 単純な再現例
長さ1万の配列をループ処理するだけのrouteを作って実験してみる。

以下のコードでは、進捗を見れるようにループの中で時々インデックスを出力している[^breakpoint]。

```java:src/main/java/org/example/mycamelapp/routes/SampleRoute.java
@Component
public class SampleRoute extends RouteBuilder {

	@Override
	public void configure() throws Exception {
		from("timer:sample?repeatCount=1")
				.transacted()

				.split(constant(new int[10000]))
					.process(exchange -> {
						int index = exchange.getProperty(Exchange.SPLIT_INDEX, Integer.class);
						if ((index & (index - 1)) == 0) System.out.println(index);
					})
				.end()

				.log("transaction finished.")
		;
	}
}
```

JVMのスタックサイズを指定して実行する。（ `-Xss` オプション）

```bash:terminal
cd path/to/app

# create JAR file
mvn clean package spring-boot:repackage

# run (send SIGINT after 20 seconds)
timeout -sINT 20s \
	java -Xss256k -jar target/mycamelapp-1.0-SNAPSHOT.jar
```

手元の環境では、スタックサイズ256KBだとループ140回程度、1MB（デフォルト）だと800回程度、4MBだと3000回程度で `StackOverflowError` が発生した。16MBなら1万回は大丈夫だった。

一方で、 `.transacted()` を消した場合は256KBでも問題なく処理できた。

（2021/07/21追記） Camel 3.8 を使った場合も問題なく処理できた。

[^breakpoint]: IDEで `.process()` のラムダ式内にブレークポイントを設置すれば、メッセージ処理を一時停止してスタックトレースの確認もできる。

### スタックトレースの確認
camel の内部のソースコードを理解するのは難しいが、どういう動作でエラーが発生したか確認しておく。ループ中の `StackOverflowError` なので、恐らく**再帰的な呼び出し**になっていると予想はつく。

実際、以下の部分が再帰になっていた。

| Stack Trace | Source Code (3.2.0) |
|:--|:--|
| ... | ... |
| CamelInternalProcessor.process() | CamelInternalProcessor.java:286 |
| MulticastProcessor\$MulticastState.lambda\$run\$1() | MulticastProcessor.java:[364](https://github.com/apache/camel/blob/camel-3.2.0/core/camel-base/src/main/java/org/apache/camel/processor/MulticastProcessor.java#L364) |
| AsyncCompletionService\$Task.run() | AsyncCompletionService.java:[150](https://github.com/apache/camel/blob/camel-3.2.0/core/camel-util/src/main/java/org/apache/camel/util/concurrent/AsyncCompletionService.java#L150) |
| DefaultReactiveExecutor\$Worker.schedule() | DefaultReactiveExecutor.java:[148](https://github.com/apache/camel/blob/camel-3.2.0/core/camel-base/src/main/java/org/apache/camel/impl/engine/DefaultReactiveExecutor.java#L148) |
| DefaultReactiveExecutor.schedule() | DefaultReactiveExecutor.java:[55](https://github.com/apache/camel/blob/camel-3.2.0/core/camel-base/src/main/java/org/apache/camel/impl/engine/DefaultReactiveExecutor.java#L55) |
| CamelInternalProcessor\$AsyncAfterTask.done() | CamelInternalProcessor.java:[186](https://github.com/apache/camel/blob/camel-3.2.0/core/camel-base/src/main/java/org/apache/camel/processor/CamelInternalProcessor.java#L186) |
| TransactionErrorHandler.process() | TransactionErrorHandler.java:[126](https://github.com/apache/camel/blob/camel-3.2.0/components/camel-spring/src/main/java/org/apache/camel/spring/spi/TransactionErrorHandler.java#L126) |
| CamelInternalProcessor.process() | CamelInternalProcessor.java:[286](https://github.com/apache/camel/blob/camel-3.2.0/core/camel-base/src/main/java/org/apache/camel/processor/CamelInternalProcessor.java#L286) |

`.transacted()` を消した場合のスタックトレースも見てみたが、もっと根元に近い部分からメソッド呼び出しが変わっていて、うまく比較できなかった。

### エラー回避策（欠点あり）
`.split()` に使える設定を眺めていたところ、並列処理にして**ループの度にスレッドを分ければいい**のではないかと思いついた。スレッドプールの指定もできるので[^exec-svc]、処理順序を保つ必要があるなら1スレッドだけ割り当てればいい。

```diff_java:diff
 				.split(constant(new int[10000]))
+					.parallelProcessing()
+					.executorService(Executors.newSingleThreadExecutor())
+
 					.process(exchange -> {
```

スレッドを分けるオーバーヘッドのためか、処理は遅くなる。ループ内部の処理時間が十分長ければ気にしなくていいと思う。（未確認）

もっと重大な問題について次節で扱う。

[^exec-svc]: スレッドプールを指定した場合は自動的に `.parallelProcessing()` もONになるので、両方指定する必要は無い。

## DBの絡む例
「スレッドを別にした場合、DBのトランザクションは共有されるのか？」という不安が浮かんだので、もう少し複雑にしたrouteで実験する。

ループの外でDBレコードを作成し、中で取得する。同じトランザクションでないとこの新規レコードは見えない。

```java:src/main/java/org/example/mycamelapp/routes/SampleRoute.java
@Component
public class SampleRoute extends RouteBuilder {

	@Override
	public void configure() throws Exception {
		from("timer:sample?repeatCount=1")
				.transacted()

				// create a new task and get its ID
				.to("sql:INSERT INTO task () VALUES ()")
				.to("sql:SELECT LAST_INSERT_ID()?outputType=SelectOne")
				.setHeader("task_id", body())

				.setBody(constant(new int[10000]))
				.split(body())
					.parallelProcessing() // BAD!!

					// get the task
					.to("sql:SELECT * FROM task WHERE task_id = :#task_id?outputType=SelectOne")

					// (debug)
					.process(exchange -> {
						int index = exchange.getProperty(Exchange.SPLIT_INDEX, Integer.class);
						if ((index & (index - 1)) == 0)
							System.out.printf("index: %d, body: %s%n", index, exchange.getIn().getBody());
					})
				.end()

				.log("transaction finished.")
		;
	}
}
```

これを実行すると、SELECT文でレコードを取得できていない（nullになっている）ことがわかる。 `.parallelProcessing()` を消した場合はレコードを取得できる（が StackOverflowError になる）。

### エラー回避策（第2案）
他にも色々と試していたところ、再帰処理は `split()` が終わる（対応する `end()` を抜ける）と完了することがわかった。これは**ループのネストは問題になりにくい**ことを示している[^nest]。ならば**現在1重のループを敢えて多重に変形する**と解決するかもしれない。

このためには、 `split()` に渡すデータを分割しなければいけない。 ruby でいう [`#each_slice`](https://docs.ruby-lang.org/ja/latest/method/Enumerable/i/each_slice.html) みたいなことをしたいのだが、 camel に適切なものがあるか分からなかったので、類似のメソッド `tokenize()` を参考に自作した。

```java:src/main/java/org/example/mycamelapp/routes/SampleRoute.java
@Component
public class SampleRoute extends RouteBuilder {

	@Override
	public void configure() throws Exception {
		from("timer:sample?repeatCount=1")
				.transacted()

				// create a new task and get its ID
				.to("sql:INSERT INTO task () VALUES ()")
				.to("sql:SELECT LAST_INSERT_ID()?outputType=SelectOne")
				.setHeader("task_id", body())

				.setBody(constant(new int[10000]))
				.split(group(body(), 1000))
				.split(group(body(), 100))
				.split(body())

					// get the task
					.to("sql:SELECT * FROM task WHERE task_id = :#task_id?outputType=SelectOne")

					// (debug)
					.process(exchange -> {
						int index = exchange.getProperty(Exchange.SPLIT_INDEX, Integer.class);
						if ((index & (index - 1)) == 0)
							System.out.printf("index: %d, body: %s%n", index, exchange.getIn().getBody());
					})
				.end()
				.end()
				.end()

				.log("transaction finished.")
		;
	}

	// org.apache.camel.support.builder.*
	public static ValueBuilder group(ValueBuilder builder, int group) {
		return new ValueBuilder(ExpressionBuilder.groupIteratorExpression(
				builder.getExpression(), null, "" + group, false
		));
	}
}
```

<!--

```java
	public static ValueBuilder chunk(ValueBuilder builder, Expression size) {
		Expression expression = builder.getExpression();

		Expression newExp = new ExpressionAdapter() {

			@Override
			public void init(CamelContext context) {
				expression.init(context);
				size.init(context);
			}

			@Override
			public Object evaluate(Exchange exchange) {
				List<Object> list = expression.evaluate(exchange, List.class);
				Integer sliceSize = size.evaluate(exchange, Integer.class);

				List<List<Object>> nestedList = new ArrayList<>();
				List<Object> slicedlist = new ArrayList<>();
				int i = 0;
				for (Object obj : list) {
					slicedlist.add(obj);
					if (++i % sliceSize == 0) {
						nestedList.add(slicedlist);
						slicedlist = new ArrayList<>();
					}
				}
				if (!slicedlist.isEmpty()) {
					nestedList.add(slicedlist);
				}
				return nestedList;
			}

			@Override
			public String toString() {
				return "chunk(" + expression + ", " + size + ")";
			}
		};

		return new ValueBuilder(newExp);
	}
```

-->

ループを 100 \* 10 \* 10 ≧ 10000 に分けたので、スタックの消費は 100 + 10 + 10 = 120 程度で済む。実際このコードはスタックサイズ256KB（元々は140回が限界）でもエラーにならなかった。
※ 3重なら3乗根が最もスタックを節約できる

この例ではループの中は修正していないが、 `Exchange.SPLIT_INDEX` の値は一番内側のループのカウントなので、元と異なるログ出力になっている。他にもループを多重化したことへの対応が必要になるパターンは多いと思われる。

[^nest]: 各ループ回数の掛け算でなく足し算で StackOverflowError になるかどうか決まるということ。

## 参考

* Camel のドキュメント
	* [Transactional Client](https://camel.apache.org/components/latest/eips/transactional-client.html)
	* [Split](https://camel.apache.org/components/latest/eips/split-eip.html)
* Camel のソースコード
	* [org.apache.camel.support.builder.ValueBuilder.tokenize()](https://github.com/apache/camel/blob/camel-3.2.0/core/camel-support/src/main/java/org/apache/camel/support/builder/ValueBuilder.java#L184-L192)
	* 「[スタックトレースの確認](#スタックトレースの確認)」節も参照
* `transacted()` と `split()` の組み合わせに関する質問
	* [Splitter and Transacted Camel Route - Stack Overflow](https://stackoverflow.com/questions/60984781)
	* [Transactions in camel routes with Splitter parallel processing - Stack Overflow](https://stackoverflow.com/questions/29738533)