Spring Batch 구현 - 2

 

♧ 전체 코드 : https://github.com/woodisco/pass-batch

GitHub - woodisco/pass-batch

 

JPA 연동

① repository 작성

• repository/booking
• repository/notification
• repository/packaze
• repository/pass
• repository/statistics
• repository/user

 
② BaseEntity 작성

@MappedSuperclass // 이 클래스를 상속받은 entity에서 아래 필드들을 컬럼으로 사용할 수 있다.
@EntityListeners(AuditingEntityListener.class) // Auditing 정보를 캡처하는 Listener
public abstract class BaseEntity {
    @CreatedDate
    @Column(updatable = false, nullable = false) // 업데이트를 하지 않도록, null이 되지 않도록 명시
    private LocalDateTime createdAt;
    @LastModifiedDate // 마지막 수정 일시를 생성
    private LocalDateTime modifiedAt;

 
③ JpaConfig 작성

@EnableJpaAuditing // JPA auditing을 활성화, BaseEntity를 사용할 수 있도록 하며 entity의 생성일시와 수정일시를 자동화하는 용도로 사용
@Configuration
public class JpaConfig {
}

 
④ Entity 작성

• PackageEntity.java

@Getter
@Setter
@ToString
@Entity
@Table(name = "package")
public class PackageEntity extends BaseEntity {
    @Id
    @GeneratedValue(strategy = GenerationType.IDENTITY) // 기본 키 생성을 DB에 위임(AUTO_INCREMENT)
    private Integer packageSeq;

    private String packageName;
    private Integer count;
    private Integer period;

}

 
⑤ application.yml 파일 수정

spring:
  datasource:
    url: jdbc:mysql://127.0.0.1:3306/batch_local?zeroDateTimeBehavior=convertToNull&characterEncoding=UTF-8&serverTimezone=Asia/Tokyo
    username: test
    password: test
    driver-class-name: com.mysql.cj.jdbc.Driver
    hikari: # Spring Boot2부터 default DBCP는 hikariCP
      maximum-pool-size: 20 # pool에 보관 가능한 최대 connection 수
  
  batch:
    jdbc:
      # Batch에 사용되는 SQL 초기화 스크립트를 실행, 임베디드인 경우에는 기본적으로 실행하는데, always 설정을 하면 임베디드 타입처럼 항상 활성화
      # 초기 수행한 이후에는 꺼두어도 됩니다.
      initialize-schema: always

 
⑥ DB 올리기

make db-up

 
⑦ 전체 실행
배치 프로그램 전체를 실행했더니, 버전문제로 인하여 배치의 메타데이터가 생성되지 않는 에러가 발생했다. 그래서 schema-mysql.sql 을 검색해서 해당 sql을 복사&실행하여 메타테이블을 생성했다.

위와 더불어  BatchConfig에도 @EnableBatchProcessing 추가해주고 다시 실행했더니 해결되었다.

---

repository 구현

① build.gradle 에 추가

// mapstruct
implementation 'org.mapstruct:mapstruct:1.5.2.Final'
annotationProcessor 'org.mapstruct:mapstruct-processor:1.5.2.Final'

// test
testImplementation "org.testcontainers:testcontainers:1.17.6"
implementation platform('org.testcontainers:testcontainers-bom:1.17.6')
testImplementation('org.testcontainers:mysql')

testImplementation 'org.springframework.boot:spring-boot-starter-test'
testImplementation 'org.springframework.batch:spring-batch-test'
testCompileOnly 'org.projectlombok:lombok'
testAnnotationProcessor 'org.projectlombok:lombok'

 
② 각 repository 생성

---

이용권 만료 배치

ended_at의 기간이 지나면 이용권이 만료된것으로 간주하고 스테이터스를 만료상태로 변경하도록 기능을 작성한다.

Chunk 기반으로 동작하는 Step은 ItemReader, ItemProcessor, ItemWriter로 구성된다.

- ItemReader : 파일에서 읽거나 DB에서 읽은 인풋을 처리할 수 있는 방법을 제공한다.

Step은 ItemReader를 사용해서 각 아이템을 개별적으로 읽은 다음, ItemProcessor에게 전달해 필요한 처리를 수행한다.
이 작업을 Chunk 사이즈가 될 때까지 반복하고 하나의 Chunk가 완성이 되면 목록을 ItemWriter로 전달한다.

ItemReader에서 데이터를 가져오는 방법으로는 2가지가 존재한다.

 
① ExpirePassesJobConfig 작성

버전 문제로 인해 JobBuilderFactory와 StepBuilderFactory를 사용할 수 없게 되었기 때문에 아래와 같이 수정했다.

@EnableBatchProcessing
@Configuration
public class ExpirePassesJobConfig {
    private final int CHUNK_SIZE = 5;

    private final EntityManagerFactory entityManagerFactory;

    public ExpirePassesJobConfig(EntityManagerFactory entityManagerFactory) {
        this.entityManagerFactory = entityManagerFactory;
    }

    @Bean
    public Job expirePassesJob(JobRepository jobRepository, Step expirePassesStep) {
        return new JobBuilder("expirePassesJob", jobRepository)
                .start(expirePassesStep)
                .build();
    }

    @Bean
    public Step expirePassesStep(JobRepository jobRepository, PlatformTransactionManager transactionManager) {
        return new StepBuilder("expirePassesStep", jobRepository)
                .<PassEntity, PassEntity>chunk(CHUNK_SIZE, transactionManager)
                .reader(expirePassesItemReader())
                .processor(expirePassesItemProcessor())
                .writer(expirePassesItemWriter())
                .build();
    }

    @Bean
    @StepScope
    public JpaCursorItemReader<PassEntity> expirePassesItemReader() {
        return new JpaCursorItemReaderBuilder<PassEntity>()
                .name("expirePassesItemReader")
                .entityManagerFactory(entityManagerFactory)
                .queryString("select p from PassEntity p where p.status = :status and p.endedAt <= :endedAt")
                .parameterValues(Map.of("status", PassStatus.PROGRESSED, "endedAt", LocalDateTime.now()))
                .build();
    }

    @Bean
    public ItemProcessor<PassEntity, PassEntity> expirePassesItemProcessor() {
        return passEntity -> {
            passEntity.setStatus(PassStatus.EXPIRED);
            passEntity.setExpiredAt(LocalDateTime.now());
            return passEntity;
        };
    }

    @Bean
    public JpaItemWriter<PassEntity> expirePassesItemWriter() {
        return new JpaItemWriterBuilder<PassEntity>()
                .entityManagerFactory(entityManagerFactory)
                .build();
    }
}

@StepScope
Spring Batch와 함께 사용되는 기능 중 하나입니다. 이것은 스텝 실행 동안에만 빈(Bean)이 생성되고 제공되는 것을 의미합니다. 이것은 Spring Batch의 처리 중에 매우 유용합니다. Spring Batch는 대용량 데이터를 처리하고, 대용량의 데이터를 가공하고, 이러한 프로세스를 관리하기 위해 사용됩니다.
일반적으로 Spring Batch에서는 Job Scope로 처리되는 것을 볼 수 있습니다. 즉, 스프링 배치의 모든 스텝은 하나의 작업 스코프(Job Scope) 내에서 실행됩니다. 따라서 일반적으로 스프링 배치의 빈(Bean)은 기본적으로 Job Scope에 따라 작동합니다.
그러나 때로는 스텝을 처리할 때마다 스텝 스코프(Step Scope)에 따라 빈(Bean)을 생성하고 사용해야 할 수 있습니다. 예를 들어, 스텝 간에 상태를 유지하기 위해 동일한 빈(Bean)을 재사용하는 대신, 각 스텝에 대해 새로운 인스턴스를 만들어야 할 수 있습니다. 이런 경우에 StepScope가 사용됩니다.

즉 @StepScope가 붙어 있는 메서드에서는 바로 그 메서드의 실행지점에 해당 @Bean을 Spring Bean으로 생성합니다.

참조 : https://ahndy84.tistory.com/25

② ExpirePassesJobConfigTest 작성

@Slf4j // 롬복(Lombok)의 애노테이션으로, Logger를 생성
@SpringBatchTest
@SpringBootTest
@ActiveProfiles("test") // "test" 프로파일이 활성화되도록 하는 애노테이션 & 테스트에 필요한 프로파일을 설정
@ContextConfiguration(classes = {ExpirePassesJobConfig.class, TestBatchConfig.class})
// 테스트 컨텍스트가 로드될 때 사용할 자바 구성 클래스를 지정합니다. ExpirePassesJobConfig와 TestBatchConfig 클래스가 테스트 컨텍스트에 로드됩니다.
public class ExpirePassesJobConfigTest {
    @Autowired
    private JobLauncherTestUtils jobLauncherTestUtils;

    @Autowired
    private PassRepository passRepository;

    @Test
    public void test_expirePassesStep() throws Exception {
        // given
        addPassEntities(10);

        // when
        JobExecution jobExecution = jobLauncherTestUtils.launchJob();
        JobInstance jobInstance = jobExecution.getJobInstance();

        // then
        assertEquals(ExitStatus.COMPLETED, jobExecution.getExitStatus());
        assertEquals("expirePassesJob", jobInstance.getJobName());
    }

    private void addPassEntities(int size) {
        final LocalDateTime now = LocalDateTime.now();
        final Random random = new Random();

        List<PassEntity> passEntities = new ArrayList<>();
        for(int i=0; i<size; i++) {
            PassEntity passEntity = new PassEntity();
            passEntity.setPackageSeq(1);
            passEntity.setUserId("A" + 1000000 + i);
            passEntity.setStatus(PassStatus.PROGRESSED);
            passEntity.setRemainingCount(random.nextInt(11));
            passEntity.setStartedAt(now);
            passEntity.setEndedAt(now.minusDays(60));
            passEntities.add(passEntity);
        }

        passRepository.saveAll(passEntities);
    }
}

 
③ TestBatchConfig 작성

@Configuration
@EnableJpaAuditing // JPA Auditing을 활성화하는 애노테이션으로, 엔티티 변경 시간 추적을 가능하게 합니다.
@EnableAutoConfiguration // 스프링 부트의 자동 구성을 활성화합니다.
@EnableBatchProcessing
@EntityScan("com.fastcampus.pass.repository") // 엔티티를 검색하기 위한 패키지를 설정합니다.
@EnableJpaRepositories("com.fastcampus.pass.repository") // JPA 리포지토리를 활성화하고, 리포지토리를 검색하기 위한 패키지를 설정합니다.
@EnableTransactionManagement // 트랜잭션 관리를 활성화하는 애노테이션입니다.@Transactional 애노테이션이 동작하도록 해줍니다.
public class TestBatchConfig {
}

 
④ ExpirePassesJobConfigTest 실행
테스트 실행 시 아래와 같은 에러 발생

PreparedStatementCallback; SQL [INSERT INTO BATCH_JOB_INSTANCE(JOB_INSTANCE_ID, JOB_NAME, JOB_KEY, VERSION)
	VALUES (?, ?, ?, ?)
]; Duplicate entry '0' for key 'BATCH_JOB_INSTANCE.PRIMARY'
org.springframework.dao.DuplicateKeyException: PreparedStatementCallback; SQL [INSERT INTO BATCH_JOB_INSTANCE(JOB_INSTANCE_ID, JOB_NAME, JOB_KEY, VERSION)
	VALUES (?, ?, ?, ?)
]; Duplicate entry '0' for key 'BATCH_JOB_INSTANCE.PRIMARY'

조사한 결과 :
Spring Batch가 ID를 자동으로 생성하는 것을 비활성화할 필요 없이 현재 DB에 배치 관련 테이블에 데이터를 전부 삭제하고 다시 실행하니 테스트에 성공했다.

이 문제는 Spring Batch가 배치 잡을 생성할 때 일어나는 중복 키 관련 문제입니다. 기본적으로 Spring Batch는 배치 작업 인스턴스의 ID를 자동으로 생성하려고 시도합니다. 그러나 이미 데이터베이스에 0번 ID를 가진 작업 인스턴스가 있기 때문에 중복 키 예외가 발생합니다. 이 문제를 해결하기 위해서는 Spring Batch가 ID를 자동으로 생성하는 것을 비활성화해야 합니다.

 
출처 : 패스트캠퍼스 10개 프로젝트로 완성하는 백엔드 웹개발(Java/Spring) 초격차 패키지 Online