分割コミットをする方法がわかりません

Question

前提・実現したいこと

SpringBatchとMyBatisを使ってCSVをデータベースに登録するバッチを作成しています。
バッチはチャンクモデルで作成していて、以下のような構成になっています。

1. ItemReader・・・アクティブテーブルを読む
2. ItemProcessor・・・アクティブデータに登録されているパスのCSVを読む
3. ItemWriter・・・CSVファイルの中身を登録する

1のアクティブテーブルは、他システムが登録しているもので、キー情報とCSVのファイルパスが登録されています。
ItemProcessorでそのファイルを読み込み、フォーマットを統一してListに詰めます。
ItemWriterでListを受取ってDBに登録します。

最初に調べた時に、StepBuilderFactoryにchunk(1000)など設定しておくと、1000件ずつコミットしてくれると理解しました。
が、実際に動かしてみて初めて、CSVファイルの中身の件数じゃなくて、Readerで取得されるデータの件数で制御されることに気づきました…。

どのようにしたら分割でコミットができるのでしょうか？

発生している問題・エラーメッセージ

org.springframework.dao.TransientDataAccessResourceException: 
batch.XXXMapper.insertBulk (batch index #1) failed. 
Cause: java.sql.tion: Packet for query is too large (5,636,232 > 4,194,304). You can change this value on the server by setting the 'max_allowed_packet' variable.

max_allowed_packetを変えたらいいよと言われてるのは分かるのですが、このファイルが最大という訳でもないので分割でコミットしたいと思ってます。

該当のソースコード

■Step

    @Bean
    public Step insertCsvStep(
    		ItemReader<Activity> reader,
			ItemProcessor<Activity, List<XXXData>> processor,
			ItemWriter<List<XXXData>> writer) {
		return stepBuilderFactory.get("insertCsvStep")
                .<Activity, List<XXXData>> chunk(1000)
				.reader(reader)
				.processor(processor)
				.writer(writer)
				.build();
    }

■ItemReader
※単純にデータを取ってくるだけなので省略（多くて10件ぐらい）

■ItemProcessor
※ファイルによって項目数が違うのでsplitで分割しながら読んでます

	@Override
	public List<XXXData> process(Activity activity) throws Exception {
		BufferedReader br = null;
        // ItemWriterに渡すリスト
		List<XXXData> result = new ArrayList<XXXData>();
		try {
			File f = new File(activity.getPath());
			br = new BufferedReader(new FileReader(f));

			String sLine;
			while ((sLine = br.readLine()) != null) {
				csv = sLine.split(",", 0);
				
				XXXData data = new XXXData();
                :
                (諸々の加工処理）
                :
				result.add(data);
			}
		} catch (Exception e) {
			// 省略
		} finally {
			if(br != null) br.close();
		}
		return result;
	}

■ItemWriter

	@Bean
	public MyBatisBatchItemWriter<List<XXXData>> insertXXXData() {
		return new MyBatisBatchItemWriterBuilder<List<XXXData>>()
					.sqlSessionFactory(sqlSessionFactory)
					.statementId("batch.infra.mapper.XXXDataMapper.insertBulk")
					.build();
	}

■Mapper(xml)

  <insert id="insertBulk" parameterType="java.util.List">
    INSERT INTO xxx_data (
      system_id,
      :
    ) 
    VALUES 
      <foreach item="record" collection="list" separator=",">
        (
          #{record.system_id,jdbcType=VARCHAR},
          :
        )
      </foreach>  
  </insert>

試したこと

application.ymlで下記のように設定を追加してみましたが、同じでした。

mybatis:
  executor-type: BATCH
  configuration:
    default-executor-type: BATCH

補足情報（FW/ツールのバージョンなど）

SpringBoot2.3.1(プロジェクト作成時にSpringBatchを選択）
Java 13

Answer 1

Taskletで書いてDefaultTransactionDefinitionを使ったら自由にコミットできました。
最初に参考したサイトに「ItemReaderもWriterもいらない処理はTaskletで書けます」と書いてあるのを見て、DBアクセスが無い処理＝Taskletのことだと曲解してしまっていました。

		int insertCounter = 0;

		BufferedReader br = null;
		File f = new File(path);
		String sLine = "";
		
	    DefaultTransactionDefinition txDefinition = new DefaultTransactionDefinition();
	    txDefinition.setPropagationBehavior(TransactionDefinition.PROPAGATION_REQUIRED);
	    TransactionStatus txStatus =  null;
		try {
			String[] csv = null;
			br = new BufferedReader(new FileReader(f));
			while ((sLine = br.readLine()) != null) {
				csv = sLine.split(",", 0);
				// トランザクションの開始
                if (insertCounter % 1000 == 0) {
                	txStatus = txManager.getTransaction(txDefinition);
                }
                XXXData data = new XXXData();
                :
                (諸々の加工処理）
                :
				XXXDataMapper.insert(data);
				insertCounter++;
				
				// コミットサイズになったらコミット
				if(insertCounter % 1000 == 0) {
					txManager.commit(txStatus);
				}
			}
		} catch(Exception e) {
			log.error("Exception:::" + sLine);
			if(txManager != null) {
				txManager.rollback(txStatus);
			}
			throw e;
		} finally {
			if(br != null) {
				br.close();
			}
            if (txStatus != null && !txStatus.isCompleted()) {
                // 最後のチャンクについて、トランザクションをコミットする
				txManager.commit(txStatus);
            }
		}