JpaPagingItemReader 사용 시, ItemReader 타입으로 bean을 등록하지 말자.

서론.

가끔식 에러하나에 많은 시간을 투자하게되는 힘든일이 발생합니다.
오늘이 바로 그날이었습니다..

<img src="https://static.podo-dev.com/blogs/images/2020/02/27/origin/8961c68f-bc38-4020-bad4-417e9f1799c1.png" alt="base64.png" style="width:400px;">

<br>

본론.

spring-batch 테스트 코드를 짜는데, 문제가 발생합니다..

<br>

job을 실행하니, 다음 에러를 확인했습니다.

java.lang.NullPointerException: null
	at org.springframework.batch.item.database.JpaPagingItemReader.doReadPage(JpaPagingItemReader.java:192) ~[spring-batch-infrastructure-4.2.1.RELEASE.jar:4.2.1.RELEASE]
	at org.springframework.batch.item.database.AbstractPagingItemReader.doRead(AbstractPagingItemReader.java:110) ~[spring-batch-infrastructure-4.2.1.RELEASE.jar:4.2.1.RELEASE]
	at org.springframework.batch.item.support.AbstractItemCountingItemStreamItemReader.read(AbstractItemCountingItemStreamItemReader.java:93) ~[spring-batch-infrastructure-4.2.1.RELEASE.jar:4.2.1.RELEASE]
	at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method) ~[na:1.8.0_211]
	at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62) ~[na:1.8.0_211]
	at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43) ~[na:1.8.0_211]
	at java.lang.reflect.Method.invoke(Method.java:498) ~[na:1.8.0_211]

JpaPagingItemReader 내부에 entityManager가 null이어서 발생하는 에러입니다.

<br>

JpaPagingItemReader가 내부에서
최초에 doOpen()을 호출함으로써
entityManager를 초기화하는데, 문제는 doOpen()을 호출하지 않습니다.

왜 doOpen()이 호출이 안되는지,
여러여러 찾아보니, 해당 링크에서 답을 얻었습니다.

https://jira.spring.io/si/jira.issueviews:issue-html/BATCH-2256/BATCH-2256.html

<br>

문제는 @Bean을 등록 시에 반환 타입을 ItemReader로 받은 것이 문제입니다.

    @StepScope
    @Bean
    public ItemReader<? extends BlaBla> searchJobReader(EntityManagerFactory entityManagerFactory) {
        return new JpaPagingItemReaderBuilder<BlaBla>().blabla();
    }

<br>

ItemReader로 상위캐스팅되어 반환 받으니,
ItemStream의 구현체가 아니기 때문에 초기화가 안되는 것입니다.

원인은 프록시다!
관련된 딥한 내용은 하단에 정리하였습니다.

public interface ItemReader<T> {}

<br>

ItemStream에 open() 메소드가 정의되어있습니다.
open() 메소드 JpaPagingItemReader# doOpen() 메소드를 호출합니다.

public interface ItemStream {

	/**
	 * Open the stream for the provided {@link ExecutionContext}.
	 *
	 * @param executionContext current step's {@link org.springframework.batch.item.ExecutionContext}.  Will be the
	 *                            executionContext from the last run of the step on a restart.
	 * @throws IllegalArgumentException if context is null
	 */
	void open(ExecutionContext executionContext) throws ItemStreamException; /// `doOpen()`호출, `AbstractItemCountingItemStreamItemReader`

<br>

전체적으로 그려보면 다음 구조입니다.

ItemReader <|.. ItemStreamReader
ItemStream <|.. ItemStreamReader
ItemStreamReader <|-- AbstractItemStreamItemReader
AbstractItemStreamItemReader <|-- AbstractItemCountingItemStreamItemReader
AbstractItemCountingItemStreamItemReader <|-- AbstractPagingItemReader
AbstractPagingItemReader <-- JpaPagingItemReader

interface ItemStream{
    abstract open()
}

interface ItemStreamReader
interface ItemReader
abstract class AbstractItemStreamItemReader
abstract class AbstractItemCountingItemStreamItemReader{
    open();
    abstract doOpen();
}
abstract class AbstractPagingItemReader{
    doOpen();
}
abstract class JpaPagingItemReader{
    doOpen();
}

<br>

그래서 다음과 같이 반환타입을 바꾸면, 정상 작동하게 됩니다.

    @StepScope
    @Bean
    public ItemStreamReader<? extends BlaBla> searchJobReader(EntityManagerFactory entityManagerFactory) {
        return new JpaPagingItemReaderBuilder<BlaBla>().blabla();
    }

public interface ItemStreamReader<T> extends ItemStream, ItemReader<T> {}

<br>

더 딥하게 따라가야 합니다.

왜(?) 라는 의문에 따라가봤습니다,
아니 상위캐스팅 됬다고 모른다고?! 이해가안되네..

<br>

jobLanucher# run()을 호출하면,
SimpleStepBuilder# build()가 호출됩니다

<br>

- SimpleStepBuilder.java
build() 메소드에서
registerAsStreamsAndListeners() 메소드를 호출되는것을 알 수 있습니다.

/**
	 * Build a step with the reader, writer, processor as provided.
	 *
	 * @see org.springframework.batch.core.step.builder.AbstractTaskletStepBuilder#build()
	 */
	@Override
	public TaskletStep build() {

		registerStepListenerAsItemListener();
		registerAsStreamsAndListeners(reader, processor, writer); // !!
		return super.build();
	}

<br>

- SimpleStepBuilder.java
registerAsStreamsAndListeners() 메소드는
reader , processor, writer가 ItemStream 인지를 검증합니다.
(!!!) ItemReader로 반환하면, 여기서 걸리지가 않습니다??

	protected void registerAsStreamsAndListeners(ItemReader<? extends I> itemReader,
			ItemProcessor<? super I, ? extends O> itemProcessor, ItemWriter<? super O> itemWriter) {
		for (Object itemHandler : new Object[] { itemReader, itemWriter, itemProcessor }) {
			if (itemHandler instanceof ItemStream) { // ItemStream 나와라 ㅡㅡ..
				stream((ItemStream) itemHandler); // 넵!
			}
        //..
    //..

<br>

아니 근데 왜안걸릴까요?
다음 상황에서, 자손 클래스로 생성 했다면,
instanceOf는 true를 반환하는게 맞습니다.

Parent <|-- Child

void test(){
    Parent child = new Child;
    child instanceOf Parent // true (!!)
}

<br>

궁금해서 Bean 생성 메소드에서 찍어봅니다.
엥 왜 true..?`

 @StepScope
    @Bean(JOB_BEAN_NAME +"StepReader")
    public ItemReader<? extends Stock> searchJobReader(EntityManagerFactory entityManagerFactory) {
        final ItemReader<Blabla> itemReader = new JpaPagingItemReaderBuilder<BlaBla>().blabla();

        System.out.println(stockQuerydslPagingItemReader.getClass()); // JpaPagingItemReaderBuilder
        System.out.println(itemReader instanceof ItemStream); // true (!)
        System.out.println(itemReader instanceof ItemStreamReader); // true (!)
        System.out.println(itemReader instanceof ItemReader); // true (!)

        return stockQuerydslPagingItemReader;
    }

<br>

당황스럽지만 침착하고, 메소드를 호출하는 stepFactory에서 찍어봅니다
잉 왜 넌 false..?

  @Bean
    public Step removeThumbnailRowStep() {
        final ItemReader<? extends Blabla> reader = searchJobReader(entityManagerFactory);

        System.out.println(reader.getClass());  // class com.sun.proxy.$Proxy98 (1)
        System.out.println(reader instanceof ItemStream);   // false (!!)
        System.out.println(reader instanceof ItemStreamReader); // false (!!)
        System.out.println(reader instanceof ItemReader); // true (!!)
        System.out.println("#########");
    }

원인은 스프링이 Bean 등록과정에서, proxy로 감싸버린 겁니다.
ItemReader가 proxy 클래스임 확인할 수 있습니다.

<br>

spring-boot는 CGLIB 프록시를 사용합니다.
ItemReader를 상속받는 Proxy 하위클래스를 정의합니다.
Proxy의 인스턴스를 생성합니다.그리고 reader를 대신하여 사용합니다.

Proxy는 target이라는 멤버변수를 가지고 있습니다.
여기서 target은 reader 입니다.
그리고 target의 메소드를 모두 복사하여 가지고 있습니다.
Proxy# a() 메소드를 호출하면,
target# a() 메소드를 포워딩 함으로써, Proxy를 구현합니다.

ItemReader <|-- JpaItemPagingReader
ItemReader <|-- Proxy
class Proxy{
    ItemReader targetClass;
}

<br>

핵심은, ItemReader을 상속받는 Proxy의 인스턴스를 새로 생성하여 사용하는 것입니다.
더 이상 참조하는 대상이
JpaItemPagingReader 클래스의 인스턴스가 아니게 되는 것을 말합니다.
따라서, 기존 계층 관계는 무너지게됩니다.
reader instanceOf JpaItemPagingReader는 false를 반환하게 됩니다.

<br>

결과적으로는 reader instanseOf ItemStream은 false를 반환합니다.

그리고, stepBuilder는 생각합니다.

넌 ItemStream이 아니네?? 패스다 요놈아!

불쌍한 reader는 open() 초기화가 되지 않습니다..

<br>

@Bean 등록시, proxy를 안쓰는 방법도 있습니다.
다음과 같이 config해주면 됩니다.

@Configuration(proxyBeanMethods = false)

해당 설정 시, proxy를 사용하지 않습니다.
따라서, reader instanceOf JpaItemPagingReader는 true를 반환 합니다.
다만, 이 방법을 메소드를 호출 할때마다 @Bean을 생성하게됩니다.

자세한내용은 http://wonwoo.ml/index.php/post/2000 잘 정리되어있습니다!!

<br>

더 따라가보기(!)

부록.. 구질구질하게 끝까지 따라가보겠습니다..(구질구질)

<br>

- AbstractTaskletStepBuilder.java
registerAsStreamsAndListeners() 메소드가 호출하는
AbstractTaskletStepBuilder# stream() 메소드를 따라갑니다.
streams 멤버 변수에, 인자로 받은 itemStream을 추가하는게 보입니다.

	/**
	 * Register a stream for callbacks that manage restart data.
	 *
	 * @param stream the stream to register
	 * @return this for fluent chaining
	 */
	public AbstractTaskletStepBuilder<B> stream(ItemStream stream) {
		streams.add(stream); // !!
		return this;
	}

<br>

- AbstractTaskletStepBuilder.java
자 이제,
AbstractTaskletStepBuilder# build() 메소드는 step을 반합니다.
step에 streams 멤버변수를 주입합니다. //step#setStreams()

	public TaskletStep build() {
        //..
		step.setStreams(streams.toArray(new ItemStream[0])); // !!
        //..
		return step;
	}

<br>

- TaskletStep.java
step# setStreams()를 따라가보겠습니다. 여정이 거의 끝나갑니다.
this.stream에 for문을 돌며 streams을 등록합니다.
this.stream은 CompositeItemStream 의 인스턴스 입니다.

    private CompositeItemStream stream = new CompositeItemStream();

    public void setStreams(ItemStream[] streams) {
        for (int i = 0; i < streams.length; i++) {
            registerStream(streams[i]); // !!
        }
    }

    public void registerStream(ItemStream stream) {
        this.stream.register(stream); // !!
    }

<br>

- AbstractStep.java
step이 생성됬습니다.
이제 step을 실행하게되면 step#excecute()가 호출됩니다.
그리고 대망에 open() 메소드가 호출됩니다.

@Override
	public final void execute(StepExecution stepExecution) throws JobInterruptedException,
	UnexpectedJobExecutionException {
            //..
			open(stepExecution.getExecutionContext()); // !!
            //..
    }

<br>

- TaskletStep.java
open()메소드는 CompositeItemStream# open()를 호출합니다.

    Override
	protected void open(ExecutionContext ctx) throws Exception {
		stream.open(ctx); // CompositeItemStream#open();
	}

<br>

- CompositeItemStream.java
CompositeItemStream# open()이 호출되면,
멤버변수 streams의 for문을 돌며, ItemStream# open() 호출되게 됩니다.
해당 streams는 이전에 build()하면서, 주입된 ItemStream입니다.
따라서, ItemStreamReader# open()도 호출됩니다.

   @Override
	public void open(ExecutionContext executionContext) throws ItemStreamException {
		for (ItemStream itemStream : streams) {
			itemStream.open(executionContext); //!!
		}
	}

<br>

- JpaPagingItemReader.java
최종적으로 구현계층에 따라,
JpaPagingItemReader# doOpen()이 호출되면서, entityManager가 주입됩니다

	@Override
	protected void doOpen() throws Exception {
		super.doOpen();

		entityManager = entityManagerFactory.createEntityManager(jpaPropertyMap);
		if (entityManager == null) {
			throw new DataAccessResourceFailureException("Unable to obtain an EntityManager");
		}
		// set entityManager to queryProvider, so it participates
		// in JpaPagingItemReader's managed transaction
		if (queryProvider != null) {
			queryProvider.setEntityManager(entityManager);
		}

	}

<br>

끝.

확실히 따라가면서, 원리를 이해하니, 이해가 박힙니다..
읽어주셔서 감사합니다 :)