gc/snp-batch-validation
7
0
포크 0
코드 이슈 8 풀 리퀘스트 Actions 위키 활동

Merge pull request 'refactor(batch): 파티션 스텝 프로세스 공통 모듈화 및 LastExecutionUpdateTasklet 통합' (#77) from feature/ISSUE-73-partition-step-common-module into develop

소스 검색
This commit is contained in:
HYOJIN 2026-03-23 14:38:08 +09:00
커밋 c4ad3c2f42
19개의 변경된 파일과 261개의 추가작업 그리고 536개의 파일을 삭제
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
docs/RELEASE-NOTES.md
파일 보기

@ -58,6 +58,8 @@
- 실패 건 수동 재수집 시 414 Request-URI Too Long 오류 수정 (#71)
### 변경
- 파티션 스텝 프로세스 공통 모듈화 (StringListPartitioner, BasePartitionedJobConfig, LastExecutionUpdateTasklet) (#73)
- 13개 Job Config의 인라인 LastExecutionUpdateTasklet을 공통 모듈로 교체 (#74)
- 재수집 실패건 추적 기준 sourceStepExecutionId → sourceJobExecutionId로 변경 (#64)
- 실패 레코드 Upsert 패턴 적용 (동일 키 중복 방지)
- 재시도 상태 배지 표시 (대기/재시도 N/3/재시도 초과)

82
src/main/java/com/snp/batch/common/batch/config/BasePartitionedJobConfig.java Normal file
파일 보기

@ -0,0 +1,82 @@
package com.snp.batch.common.batch.config;
import lombok.extern.slf4j.Slf4j;
import org.springframework.batch.core.Step;
import org.springframework.batch.core.job.flow.FlowExecutionStatus;
import org.springframework.batch.core.job.flow.JobExecutionDecider;
import org.springframework.batch.core.partition.support.Partitioner;
import org.springframework.batch.core.repository.JobRepository;
import org.springframework.batch.core.step.builder.StepBuilder;
import org.springframework.batch.core.step.tasklet.Tasklet;
import org.springframework.core.task.TaskExecutor;
import org.springframework.transaction.PlatformTransactionManager;
/**
* 파티션 기반 병렬 처리 Job 구성을 위한 추상 클래스.
* 키 목록 조회 → 파티션 병렬 처리 → 후처리 패턴의 공통 인프라 제공.
*
* @param <I> 입력 타입 (Reader 출력, Processor 입력)
* @param <O> 출력 타입 (Processor 출력, Writer 입력)
*/
@Slf4j
public abstract class BasePartitionedJobConfig<I, O> extends BaseMultiStepJobConfig<I, O> {
public BasePartitionedJobConfig(JobRepository jobRepository, PlatformTransactionManager transactionManager) {
super(jobRepository, transactionManager);
}
/**
* 파티션 Step을 생성합니다.
*
* @param stepName 파티션 Step 이름
* @param workerStepName Worker Step 이름 (partitioner 등록에 사용)
* @param partitioner Partitioner 인스턴스
* @param workerStep Worker Step 인스턴스
* @param taskExecutor 병렬 실행용 TaskExecutor
* @param gridSize 파티션 수
* @return 구성된 파티션 Step
*/
protected Step createPartitionedStep(String stepName, String workerStepName,
Partitioner partitioner, Step workerStep,
TaskExecutor taskExecutor, int gridSize) {
return new StepBuilder(stepName, jobRepository)
.partitioner(workerStepName, partitioner)
.step(workerStep)
.taskExecutor(taskExecutor)
.gridSize(gridSize)
.build();
}
/**
* 키 건수 기반 Decider를 생성합니다.
* JobExecutionContext의 지정된 키 값이 0이면 EMPTY_RESPONSE, 아니면 NORMAL 반환.
*
* @param contextKey JobExecutionContext에서 조회할 int 키 이름
* @param jobName 로그에 표시할 Job 이름
* @return 키 건수 기반 JobExecutionDecider
*/
protected JobExecutionDecider createKeyCountDecider(String contextKey, String jobName) {
return (jobExecution, stepExecution) -> {
int totalCount = jobExecution.getExecutionContext().getInt(contextKey, 0);
if (totalCount == 0) {
log.info("[{}] Decider: EMPTY_RESPONSE - {} 0건으로 후속 스텝 스킵", jobName, contextKey);
return new FlowExecutionStatus("EMPTY_RESPONSE");
}
log.info("[{}] Decider: NORMAL - {} {} 건 처리 시작", jobName, contextKey, totalCount);
return new FlowExecutionStatus("NORMAL");
};
}
/**
* LastExecution 업데이트 Step을 생성합니다.
*
* @param stepName Step 이름
* @param tasklet LastExecutionUpdateTasklet 인스턴스
* @return 구성된 LastExecution 업데이트 Step
*/
protected Step createLastExecutionUpdateStep(String stepName, Tasklet tasklet) {
return new StepBuilder(stepName, jobRepository)
.tasklet(tasklet, transactionManager)
.build();
}
}

61
src/main/java/com/snp/batch/common/batch/partition/StringListPartitioner.java Normal file
파일 보기

@ -0,0 +1,61 @@
package com.snp.batch.common.batch.partition;
import lombok.extern.slf4j.Slf4j;
import org.springframework.batch.core.partition.support.Partitioner;
import org.springframework.batch.item.ExecutionContext;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
/**
* 문자열 키 목록을 N개 파티션으로 균등 분할하는 범용 Partitioner.
*
* <p>각 파티션의 ExecutionContext에 다음 값을 저장한다.</p>
* <ul>
* <li>{@code {contextKeyName}} — 해당 파티션에 할당된 키 목록 (CSV 형식)</li>
* <li>{@code partitionIndex} — 파티션 인덱스 (0-based)</li>
* <li>{@code partitionSize} — 해당 파티션의 키 수</li>
* </ul>
*/
@Slf4j
public class StringListPartitioner implements Partitioner {
private final List<String> allKeys;
private final int partitionCount;
private final String contextKeyName;
public StringListPartitioner(List<String> allKeys, int partitionCount, String contextKeyName) {
this.allKeys = allKeys;
this.partitionCount = partitionCount;
this.contextKeyName = contextKeyName;
}
@Override
public Map<String, ExecutionContext> partition(int gridSize) {
int totalSize = allKeys.size();
int actualPartitionCount = Math.min(partitionCount, Math.max(1, totalSize));
Map<String, ExecutionContext> partitions = new LinkedHashMap<>();
int partitionSize = (int) Math.ceil((double) totalSize / actualPartitionCount);
for (int i = 0; i < actualPartitionCount; i++) {
int fromIndex = i * partitionSize;
int toIndex = Math.min(fromIndex + partitionSize, totalSize);
if (fromIndex >= totalSize) break;
List<String> partitionKeys = allKeys.subList(fromIndex, toIndex);
ExecutionContext context = new ExecutionContext();
context.putString(contextKeyName, String.join(",", partitionKeys));
context.putInt("partitionIndex", i);
context.putInt("partitionSize", partitionKeys.size());
String partitionKey = "partition" + i;
partitions.put(partitionKey, context);
log.info("[StringListPartitioner] {} : 키 {} 건 (index {}-{})",
partitionKey, partitionKeys.size(), fromIndex, toIndex - 1);
}
log.info("[StringListPartitioner] 총 {} 개 파티션 생성 (전체 키: {} 건)",
partitions.size(), totalSize);
return partitions;
}
}

73
src/main/java/com/snp/batch/common/batch/tasklet/LastExecutionUpdateTasklet.java Normal file
파일 보기

@ -0,0 +1,73 @@
package com.snp.batch.common.batch.tasklet;
import lombok.extern.slf4j.Slf4j;
import org.springframework.batch.core.StepContribution;
import org.springframework.batch.core.scope.context.ChunkContext;
import org.springframework.batch.core.step.tasklet.Tasklet;
import org.springframework.batch.repeat.RepeatStatus;
import org.springframework.jdbc.core.JdbcTemplate;
import java.sql.Timestamp;
import java.time.LocalDateTime;
/**
* 배치 작업 완료 후 BATCH_LAST_EXECUTION 테이블의 LAST_SUCCESS_DATE를 업데이트하는 공통 Tasklet.
*
* <p>RECOLLECT 모드일 경우 업데이트를 스킵하며,
* Job ExecutionContext에 저장된 {@code batchToDate}를 기준으로 성공 날짜를 계산합니다.
* {@code batchToDate}가 없을 경우 현재 시간에서 {@code bufferHours}를 차감하여 사용합니다.</p>
*/
@Slf4j
public class LastExecutionUpdateTasklet implements Tasklet {
private static final String RECOLLECT_MODE = "RECOLLECT";
private final JdbcTemplate jdbcTemplate;
private final String targetSchema;
private final String apiKey;
private final int bufferHours;
public LastExecutionUpdateTasklet(JdbcTemplate jdbcTemplate, String targetSchema,
String apiKey, int bufferHours) {
this.jdbcTemplate = jdbcTemplate;
this.targetSchema = targetSchema;
this.apiKey = apiKey;
this.bufferHours = bufferHours;
}
@Override
public RepeatStatus execute(StepContribution contribution, ChunkContext chunkContext) {
String executionMode = chunkContext.getStepContext()
.getStepExecution().getJobExecution()
.getJobParameters().getString("executionMode", "NORMAL");
if (RECOLLECT_MODE.equals(executionMode)) {
log.info(">>>>> RECOLLECT 모드 - LAST_EXECUTION 업데이트 스킵");
return RepeatStatus.FINISHED;
}
String toDateStr = chunkContext.getStepContext()
.getStepExecution().getJobExecution()
.getExecutionContext().getString("batchToDate", null);
LocalDateTime successDate;
if (toDateStr != null) {
successDate = LocalDateTime.parse(toDateStr).minusHours(bufferHours);
log.info(">>>>> BATCH_LAST_EXECUTION 업데이트 시작 (캡처된 toDate - {}시간 버퍼: {})",
bufferHours, successDate);
} else {
successDate = LocalDateTime.now().minusHours(bufferHours);
log.warn(">>>>> batchToDate가 없어 현재 시간 - {}시간 버퍼 사용: {}", bufferHours, successDate);
}
jdbcTemplate.update(
String.format(
"UPDATE %s.BATCH_LAST_EXECUTION SET LAST_SUCCESS_DATE = ?, UPDATED_AT = NOW() WHERE API_KEY = ?",
targetSchema),
Timestamp.valueOf(successDate), apiKey
);
log.info(">>>>> BATCH_LAST_EXECUTION 업데이트 완료 (LAST_SUCCESS_DATE = {})", successDate);
return RepeatStatus.FINISHED;
}
}

34
src/main/java/com/snp/batch/jobs/compliance/batch/config/CompanyComplianceImportRangeJobConfig.java
파일 보기

@ -1,6 +1,7 @@
package com.snp.batch.jobs.compliance.batch.config;
import com.snp.batch.common.batch.config.BaseMultiStepJobConfig;
import com.snp.batch.common.batch.tasklet.LastExecutionUpdateTasklet;
import com.snp.batch.jobs.compliance.batch.dto.CompanyComplianceDto;
import com.snp.batch.jobs.compliance.batch.entity.CompanyComplianceEntity;
import com.snp.batch.jobs.compliance.batch.processor.CompanyComplianceDataProcessor;
@ -30,8 +31,6 @@ import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.transaction.PlatformTransactionManager;
import org.springframework.web.reactive.function.client.WebClient;
import java.sql.Timestamp;
import java.time.LocalDateTime;
import java.time.OffsetDateTime;
import java.time.ZoneId;
import java.time.format.DateTimeFormatter;
@ -211,36 +210,7 @@ public class CompanyComplianceImportRangeJobConfig extends BaseMultiStepJobConfi
*/
@Bean
public Tasklet companyComplianceLastExecutionUpdateTasklet() {
return (contribution, chunkContext) -> {
String executionMode = chunkContext.getStepContext()
.getStepExecution().getJobExecution()
.getJobParameters().getString("executionMode", "NORMAL");
if ("RECOLLECT".equals(executionMode)) {
log.info(">>>>> RECOLLECT 모드 - LAST_EXECUTION 업데이트 스킵");
return RepeatStatus.FINISHED;
}
String toDateStr = chunkContext.getStepContext()
.getStepExecution().getJobExecution()
.getExecutionContext().getString("batchToDate", null);
LocalDateTime successDate;
if (toDateStr != null) {
successDate = LocalDateTime.parse(toDateStr).minusHours(lastExecutionBufferHours);
log.info(">>>>> BATCH_LAST_EXECUTION 업데이트 시작 (캡처된 toDate - {}시간 버퍼: {})", lastExecutionBufferHours, successDate);
} else {
successDate = LocalDateTime.now().minusHours(lastExecutionBufferHours);
log.warn(">>>>> batchToDate가 없어 현재 시간 - {}시간 버퍼 사용: {}", lastExecutionBufferHours, successDate);
}
jdbcTemplate.update(
String.format("UPDATE %s.BATCH_LAST_EXECUTION SET LAST_SUCCESS_DATE = ?, UPDATED_AT = NOW() WHERE API_KEY = ?", targetSchema),
Timestamp.valueOf(successDate), getApiKey()
);
log.info(">>>>> BATCH_LAST_EXECUTION 업데이트 완료 (LAST_SUCCESS_DATE = {})", successDate);
return RepeatStatus.FINISHED;
};
return new LastExecutionUpdateTasklet(jdbcTemplate, targetSchema, getApiKey(), lastExecutionBufferHours);
}
@Bean(name = "CompanyComplianceLastExecutionUpdateStep")
public Step companyComplianceLastExecutionUpdateStep() {

34
src/main/java/com/snp/batch/jobs/compliance/batch/config/ComplianceImportRangeJobConfig.java
파일 보기

@ -1,6 +1,7 @@
package com.snp.batch.jobs.compliance.batch.config;
import com.snp.batch.common.batch.config.BaseMultiStepJobConfig;
import com.snp.batch.common.batch.tasklet.LastExecutionUpdateTasklet;
import com.snp.batch.jobs.compliance.batch.dto.ComplianceDto;
import com.snp.batch.jobs.compliance.batch.entity.ComplianceEntity;
import com.snp.batch.jobs.compliance.batch.processor.ComplianceDataProcessor;
@ -30,8 +31,6 @@ import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.transaction.PlatformTransactionManager;
import org.springframework.web.reactive.function.client.WebClient;
import java.sql.Timestamp;
import java.time.LocalDateTime;
import java.time.OffsetDateTime;
import java.time.ZoneId;
import java.time.format.DateTimeFormatter;
@ -213,36 +212,7 @@ public class ComplianceImportRangeJobConfig extends BaseMultiStepJobConfig<Compl
*/
@Bean
public Tasklet complianceLastExecutionUpdateTasklet() {
return (contribution, chunkContext) -> {
String executionMode = chunkContext.getStepContext()
.getStepExecution().getJobExecution()
.getJobParameters().getString("executionMode", "NORMAL");
if ("RECOLLECT".equals(executionMode)) {
log.info(">>>>> RECOLLECT 모드 - LAST_EXECUTION 업데이트 스킵");
return RepeatStatus.FINISHED;
}
String toDateStr = chunkContext.getStepContext()
.getStepExecution().getJobExecution()
.getExecutionContext().getString("batchToDate", null);
LocalDateTime successDate;
if (toDateStr != null) {
successDate = LocalDateTime.parse(toDateStr).minusHours(lastExecutionBufferHours);
log.info(">>>>> BATCH_LAST_EXECUTION 업데이트 시작 (캡처된 toDate - {}시간 버퍼: {})", lastExecutionBufferHours, successDate);
} else {
successDate = LocalDateTime.now().minusHours(lastExecutionBufferHours);
log.warn(">>>>> batchToDate가 없어 현재 시간 - {}시간 버퍼 사용: {}", lastExecutionBufferHours, successDate);
}
jdbcTemplate.update(
String.format("UPDATE %s.BATCH_LAST_EXECUTION SET LAST_SUCCESS_DATE = ?, UPDATED_AT = NOW() WHERE API_KEY = ?", targetSchema),
Timestamp.valueOf(successDate), getApiKey()
);
log.info(">>>>> BATCH_LAST_EXECUTION 업데이트 완료 (LAST_SUCCESS_DATE = {})", successDate);
return RepeatStatus.FINISHED;
};
return new LastExecutionUpdateTasklet(jdbcTemplate, targetSchema, getApiKey(), lastExecutionBufferHours);
}
@Bean(name = "ComplianceLastExecutionUpdateStep")
public Step complianceLastExecutionUpdateStep() {

36
src/main/java/com/snp/batch/jobs/event/batch/config/EventImportJobConfig.java
파일 보기

@ -1,6 +1,7 @@
package com.snp.batch.jobs.event.batch.config;
import com.snp.batch.common.batch.config.BaseMultiStepJobConfig;
import com.snp.batch.common.batch.tasklet.LastExecutionUpdateTasklet;
import com.snp.batch.jobs.event.batch.dto.EventDetailDto;
import com.snp.batch.jobs.event.batch.entity.EventDetailEntity;
import com.snp.batch.jobs.event.batch.processor.EventDataProcessor;
@ -21,7 +22,6 @@ import org.springframework.batch.core.step.tasklet.Tasklet;
import org.springframework.batch.item.ItemProcessor;
import org.springframework.batch.item.ItemReader;
import org.springframework.batch.item.ItemWriter;
import org.springframework.batch.repeat.RepeatStatus;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
@ -30,9 +30,6 @@ import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.transaction.PlatformTransactionManager;
import org.springframework.web.reactive.function.client.WebClient;
import java.sql.Timestamp;
import java.time.LocalDateTime;
@Slf4j
@Configuration
public class EventImportJobConfig extends BaseMultiStepJobConfig<EventDetailDto, EventDetailEntity> {
@ -159,36 +156,7 @@ public class EventImportJobConfig extends BaseMultiStepJobConfig<EventDetailDto,
*/
@Bean
public Tasklet eventLastExecutionUpdateTasklet() {
return (contribution, chunkContext) -> {
String executionMode = chunkContext.getStepContext()
.getStepExecution().getJobExecution()
.getJobParameters().getString("executionMode", "NORMAL");
if ("RECOLLECT".equals(executionMode)) {
log.info(">>>>> RECOLLECT 모드 - LAST_EXECUTION 업데이트 스킵");
return RepeatStatus.FINISHED;
}
String toDateStr = chunkContext.getStepContext()
.getStepExecution().getJobExecution()
.getExecutionContext().getString("batchToDate", null);
LocalDateTime successDate;
if (toDateStr != null) {
successDate = LocalDateTime.parse(toDateStr).minusHours(lastExecutionBufferHours);
log.info(">>>>> BATCH_LAST_EXECUTION 업데이트 시작 (캡처된 toDate - {}시간 버퍼: {})", lastExecutionBufferHours, successDate);
} else {
successDate = LocalDateTime.now().minusHours(lastExecutionBufferHours);
log.warn(">>>>> batchToDate가 없어 현재 시간 - {}시간 버퍼 사용: {}", lastExecutionBufferHours, successDate);
}
jdbcTemplate.update(
String.format("UPDATE %s.BATCH_LAST_EXECUTION SET LAST_SUCCESS_DATE = ?, UPDATED_AT = NOW() WHERE API_KEY = ?", targetSchema),
Timestamp.valueOf(successDate), getApiKey()
);
log.info(">>>>> BATCH_LAST_EXECUTION 업데이트 완료 (LAST_SUCCESS_DATE = {})", successDate);
return RepeatStatus.FINISHED;
};
return new LastExecutionUpdateTasklet(jdbcTemplate, targetSchema, getApiKey(), lastExecutionBufferHours);
}
@Bean(name = "EventLastExecutionUpdateStep")
public Step eventLastExecutionUpdateStep() {

35
src/main/java/com/snp/batch/jobs/movement/batch/config/AnchorageCallsRangeJobConfig.java
파일 보기

@ -2,6 +2,7 @@ package com.snp.batch.jobs.movement.batch.config;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.snp.batch.common.batch.config.BaseMultiStepJobConfig;
import com.snp.batch.common.batch.tasklet.LastExecutionUpdateTasklet;
import com.snp.batch.jobs.movement.batch.dto.AnchorageCallsDto;
import com.snp.batch.jobs.movement.batch.entity.AnchorageCallsEntity;
import com.snp.batch.jobs.movement.batch.processor.AnchorageCallsProcessor;
@ -20,7 +21,6 @@ import org.springframework.batch.core.step.tasklet.Tasklet;
import org.springframework.batch.item.ItemProcessor;
import org.springframework.batch.item.ItemReader;
import org.springframework.batch.item.ItemWriter;
import org.springframework.batch.repeat.RepeatStatus;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
@ -29,8 +29,6 @@ import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.transaction.PlatformTransactionManager;
import org.springframework.web.reactive.function.client.WebClient;
import java.sql.Timestamp;
import java.time.LocalDateTime;
@Slf4j
@Configuration
@ -149,36 +147,7 @@ public class AnchorageCallsRangeJobConfig extends BaseMultiStepJobConfig<Anchora
*/
@Bean
public Tasklet anchorageCallsLastExecutionUpdateTasklet() {
return (contribution, chunkContext) -> {
String executionMode = chunkContext.getStepContext()
.getStepExecution().getJobExecution()
.getJobParameters().getString("executionMode", "NORMAL");
if ("RECOLLECT".equals(executionMode)) {
log.info(">>>>> RECOLLECT 모드 - LAST_EXECUTION 업데이트 스킵");
return RepeatStatus.FINISHED;
}
String toDateStr = chunkContext.getStepContext()
.getStepExecution().getJobExecution()
.getExecutionContext().getString("batchToDate", null);
LocalDateTime successDate;
if (toDateStr != null) {
successDate = LocalDateTime.parse(toDateStr);
log.info(">>>>> BATCH_LAST_EXECUTION 업데이트 시작 (캡처된 toDate: {})", successDate);
} else {
successDate = LocalDateTime.now();
log.warn(">>>>> batchToDate가 없어 현재 시간 사용: {}", successDate);
}
jdbcTemplate.update(
String.format("UPDATE %s.BATCH_LAST_EXECUTION SET LAST_SUCCESS_DATE = ?, UPDATED_AT = NOW() WHERE API_KEY = ?", targetSchema),
Timestamp.valueOf(successDate), getApiKey()
);
log.info(">>>>> BATCH_LAST_EXECUTION 업데이트 완료 (LAST_SUCCESS_DATE = {})", successDate);
return RepeatStatus.FINISHED;
};
return new LastExecutionUpdateTasklet(jdbcTemplate, targetSchema, getApiKey(), 0);
}
@Bean(name = "AnchorageCallsLastExecutionUpdateStep")
public Step anchorageCallsLastExecutionUpdateStep() {

35
src/main/java/com/snp/batch/jobs/movement/batch/config/BerthCallsRangJobConfig.java
파일 보기

@ -2,6 +2,7 @@ package com.snp.batch.jobs.movement.batch.config;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.snp.batch.common.batch.config.BaseMultiStepJobConfig;
import com.snp.batch.common.batch.tasklet.LastExecutionUpdateTasklet;
import com.snp.batch.jobs.movement.batch.dto.BerthCallsDto;
import com.snp.batch.jobs.movement.batch.entity.BerthCallsEntity;
import com.snp.batch.jobs.movement.batch.processor.BerthCallsProcessor;
@ -20,7 +21,6 @@ import org.springframework.batch.core.step.tasklet.Tasklet;
import org.springframework.batch.item.ItemProcessor;
import org.springframework.batch.item.ItemReader;
import org.springframework.batch.item.ItemWriter;
import org.springframework.batch.repeat.RepeatStatus;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
@ -29,8 +29,6 @@ import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.transaction.PlatformTransactionManager;
import org.springframework.web.reactive.function.client.WebClient;
import java.sql.Timestamp;
import java.time.LocalDateTime;
@Slf4j
@Configuration
@ -143,36 +141,7 @@ public class BerthCallsRangJobConfig extends BaseMultiStepJobConfig<BerthCallsDt
*/
@Bean
public Tasklet berthCallsLastExecutionUpdateTasklet() {
return (contribution, chunkContext) -> {
String executionMode = chunkContext.getStepContext()
.getStepExecution().getJobExecution()
.getJobParameters().getString("executionMode", "NORMAL");
if ("RECOLLECT".equals(executionMode)) {
log.info(">>>>> RECOLLECT 모드 - LAST_EXECUTION 업데이트 스킵");
return RepeatStatus.FINISHED;
}
String toDateStr = chunkContext.getStepContext()
.getStepExecution().getJobExecution()
.getExecutionContext().getString("batchToDate", null);
LocalDateTime successDate;
if (toDateStr != null) {
successDate = LocalDateTime.parse(toDateStr);
log.info(">>>>> BATCH_LAST_EXECUTION 업데이트 시작 (캡처된 toDate: {})", successDate);
} else {
successDate = LocalDateTime.now();
log.warn(">>>>> batchToDate가 없어 현재 시간 사용: {}", successDate);
}
jdbcTemplate.update(
String.format("UPDATE %s.BATCH_LAST_EXECUTION SET LAST_SUCCESS_DATE = ?, UPDATED_AT = NOW() WHERE API_KEY = ?", targetSchema),
Timestamp.valueOf(successDate), getApiKey()
);
log.info(">>>>> BATCH_LAST_EXECUTION 업데이트 완료 (LAST_SUCCESS_DATE = {})", successDate);
return RepeatStatus.FINISHED;
};
return new LastExecutionUpdateTasklet(jdbcTemplate, targetSchema, getApiKey(), 0);
}
@Bean(name = "BerthCallsLastExecutionUpdateStep")
public Step berthCallsLastExecutionUpdateStep() {

35
src/main/java/com/snp/batch/jobs/movement/batch/config/CurrentlyAtRangeJobConfig.java
파일 보기

@ -2,6 +2,7 @@ package com.snp.batch.jobs.movement.batch.config;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.snp.batch.common.batch.config.BaseMultiStepJobConfig;
import com.snp.batch.common.batch.tasklet.LastExecutionUpdateTasklet;
import com.snp.batch.jobs.movement.batch.dto.CurrentlyAtDto;
import com.snp.batch.jobs.movement.batch.entity.CurrentlyAtEntity;
import com.snp.batch.jobs.movement.batch.processor.CurrentlyAtProcessor;
@ -20,7 +21,6 @@ import org.springframework.batch.core.step.tasklet.Tasklet;
import org.springframework.batch.item.ItemProcessor;
import org.springframework.batch.item.ItemReader;
import org.springframework.batch.item.ItemWriter;
import org.springframework.batch.repeat.RepeatStatus;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
@ -29,8 +29,6 @@ import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.transaction.PlatformTransactionManager;
import org.springframework.web.reactive.function.client.WebClient;
import java.sql.Timestamp;
import java.time.LocalDateTime;
@Slf4j
@Configuration
@ -141,36 +139,7 @@ public class CurrentlyAtRangeJobConfig extends BaseMultiStepJobConfig<CurrentlyA
*/
@Bean
public Tasklet currentlyAtLastExecutionUpdateTasklet() {
return (contribution, chunkContext) -> {
String executionMode = chunkContext.getStepContext()
.getStepExecution().getJobExecution()
.getJobParameters().getString("executionMode", "NORMAL");
if ("RECOLLECT".equals(executionMode)) {
log.info(">>>>> RECOLLECT 모드 - LAST_EXECUTION 업데이트 스킵");
return RepeatStatus.FINISHED;
}
String toDateStr = chunkContext.getStepContext()
.getStepExecution().getJobExecution()
.getExecutionContext().getString("batchToDate", null);
LocalDateTime successDate;
if (toDateStr != null) {
successDate = LocalDateTime.parse(toDateStr);
log.info(">>>>> BATCH_LAST_EXECUTION 업데이트 시작 (캡처된 toDate: {})", successDate);
} else {
successDate = LocalDateTime.now();
log.warn(">>>>> batchToDate가 없어 현재 시간 사용: {}", successDate);
}
jdbcTemplate.update(
String.format("UPDATE %s.BATCH_LAST_EXECUTION SET LAST_SUCCESS_DATE = ?, UPDATED_AT = NOW() WHERE API_KEY = ?", targetSchema),
Timestamp.valueOf(successDate), getApiKey()
);
log.info(">>>>> BATCH_LAST_EXECUTION 업데이트 완료 (LAST_SUCCESS_DATE = {})", successDate);
return RepeatStatus.FINISHED;
};
return new LastExecutionUpdateTasklet(jdbcTemplate, targetSchema, getApiKey(), 0);
}
@Bean(name = "CurrentlyAtLastExecutionUpdateStep")
public Step currentlyAtLastExecutionUpdateStep() {

35
src/main/java/com/snp/batch/jobs/movement/batch/config/DestinationsRangeJobConfig.java
파일 보기

@ -2,6 +2,7 @@ package com.snp.batch.jobs.movement.batch.config;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.snp.batch.common.batch.config.BaseMultiStepJobConfig;
import com.snp.batch.common.batch.tasklet.LastExecutionUpdateTasklet;
import com.snp.batch.jobs.movement.batch.dto.DestinationDto;
import com.snp.batch.jobs.movement.batch.entity.DestinationEntity;
import com.snp.batch.jobs.movement.batch.processor.DestinationProcessor;
@ -20,7 +21,6 @@ import org.springframework.batch.core.step.tasklet.Tasklet;
import org.springframework.batch.item.ItemProcessor;
import org.springframework.batch.item.ItemReader;
import org.springframework.batch.item.ItemWriter;
import org.springframework.batch.repeat.RepeatStatus;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
@ -29,8 +29,6 @@ import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.transaction.PlatformTransactionManager;
import org.springframework.web.reactive.function.client.WebClient;
import java.sql.Timestamp;
import java.time.LocalDateTime;
@Slf4j
@Configuration
@ -144,36 +142,7 @@ public class DestinationsRangeJobConfig extends BaseMultiStepJobConfig<Destinati
*/
@Bean
public Tasklet destinationsLastExecutionUpdateTasklet() {
return (contribution, chunkContext) -> {
String executionMode = chunkContext.getStepContext()
.getStepExecution().getJobExecution()
.getJobParameters().getString("executionMode", "NORMAL");
if ("RECOLLECT".equals(executionMode)) {
log.info(">>>>> RECOLLECT 모드 - LAST_EXECUTION 업데이트 스킵");
return RepeatStatus.FINISHED;
}
String toDateStr = chunkContext.getStepContext()
.getStepExecution().getJobExecution()
.getExecutionContext().getString("batchToDate", null);
LocalDateTime successDate;
if (toDateStr != null) {
successDate = LocalDateTime.parse(toDateStr);
log.info(">>>>> BATCH_LAST_EXECUTION 업데이트 시작 (캡처된 toDate: {})", successDate);
} else {
successDate = LocalDateTime.now();
log.warn(">>>>> batchToDate가 없어 현재 시간 사용: {}", successDate);
}
jdbcTemplate.update(
String.format("UPDATE %s.BATCH_LAST_EXECUTION SET LAST_SUCCESS_DATE = ?, UPDATED_AT = NOW() WHERE API_KEY = ?", targetSchema),
Timestamp.valueOf(successDate), getApiKey()
);
log.info(">>>>> BATCH_LAST_EXECUTION 업데이트 완료 (LAST_SUCCESS_DATE = {})", successDate);
return RepeatStatus.FINISHED;
};
return new LastExecutionUpdateTasklet(jdbcTemplate, targetSchema, getApiKey(), 0);
}
@Bean(name = "DestinationsLastExecutionUpdateStep")
public Step destinationsLastExecutionUpdateStep() {

35
src/main/java/com/snp/batch/jobs/movement/batch/config/ShipPortCallsRangeJobConfig.java
파일 보기

@ -2,6 +2,7 @@ package com.snp.batch.jobs.movement.batch.config;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.snp.batch.common.batch.config.BaseMultiStepJobConfig;
import com.snp.batch.common.batch.tasklet.LastExecutionUpdateTasklet;
import com.snp.batch.jobs.movement.batch.dto.PortCallsDto;
import com.snp.batch.jobs.movement.batch.entity.PortCallsEntity;
import com.snp.batch.jobs.movement.batch.processor.PortCallsProcessor;
@ -20,7 +21,6 @@ import org.springframework.batch.core.step.tasklet.Tasklet;
import org.springframework.batch.item.ItemProcessor;
import org.springframework.batch.item.ItemReader;
import org.springframework.batch.item.ItemWriter;
import org.springframework.batch.repeat.RepeatStatus;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
@ -29,8 +29,6 @@ import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.transaction.PlatformTransactionManager;
import org.springframework.web.reactive.function.client.WebClient;
import java.sql.Timestamp;
import java.time.LocalDateTime;
@Slf4j
@Configuration
@ -145,36 +143,7 @@ public class ShipPortCallsRangeJobConfig extends BaseMultiStepJobConfig<PortCall
*/
@Bean
public Tasklet portCallsLastExecutionUpdateTasklet() {
return (contribution, chunkContext) -> {
String executionMode = chunkContext.getStepContext()
.getStepExecution().getJobExecution()
.getJobParameters().getString("executionMode", "NORMAL");
if ("RECOLLECT".equals(executionMode)) {
log.info(">>>>> RECOLLECT 모드 - LAST_EXECUTION 업데이트 스킵");
return RepeatStatus.FINISHED;
}
String toDateStr = chunkContext.getStepContext()
.getStepExecution().getJobExecution()
.getExecutionContext().getString("batchToDate", null);
LocalDateTime successDate;
if (toDateStr != null) {
successDate = LocalDateTime.parse(toDateStr);
log.info(">>>>> BATCH_LAST_EXECUTION 업데이트 시작 (캡처된 toDate: {})", successDate);
} else {
successDate = LocalDateTime.now();
log.warn(">>>>> batchToDate가 없어 현재 시간 사용: {}", successDate);
}
jdbcTemplate.update(
String.format("UPDATE %s.BATCH_LAST_EXECUTION SET LAST_SUCCESS_DATE = ?, UPDATED_AT = NOW() WHERE API_KEY = ?", targetSchema),
Timestamp.valueOf(successDate), getApiKey()
);
log.info(">>>>> BATCH_LAST_EXECUTION 업데이트 완료 (LAST_SUCCESS_DATE = {})", successDate);
return RepeatStatus.FINISHED;
};
return new LastExecutionUpdateTasklet(jdbcTemplate, targetSchema, getApiKey(), 0);
}
@Bean(name = "PortCallsLastExecutionUpdateStep")
public Step portCallsLastExecutionUpdateStep() {

35
src/main/java/com/snp/batch/jobs/movement/batch/config/StsOperationRangeJobConfig.java
파일 보기

@ -2,6 +2,7 @@ package com.snp.batch.jobs.movement.batch.config;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.snp.batch.common.batch.config.BaseMultiStepJobConfig;
import com.snp.batch.common.batch.tasklet.LastExecutionUpdateTasklet;
import com.snp.batch.jobs.movement.batch.dto.StsOperationDto;
import com.snp.batch.jobs.movement.batch.entity.StsOperationEntity;
import com.snp.batch.jobs.movement.batch.processor.StsOperationProcessor;
@ -20,7 +21,6 @@ import org.springframework.batch.core.step.tasklet.Tasklet;
import org.springframework.batch.item.ItemProcessor;
import org.springframework.batch.item.ItemReader;
import org.springframework.batch.item.ItemWriter;
import org.springframework.batch.repeat.RepeatStatus;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
@ -29,8 +29,6 @@ import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.transaction.PlatformTransactionManager;
import org.springframework.web.reactive.function.client.WebClient;
import java.sql.Timestamp;
import java.time.LocalDateTime;
@Slf4j
@Configuration
@ -143,36 +141,7 @@ public class StsOperationRangeJobConfig extends BaseMultiStepJobConfig<StsOperat
*/
@Bean
public Tasklet stsOperationLastExecutionUpdateTasklet() {
return (contribution, chunkContext) -> {
String executionMode = chunkContext.getStepContext()
.getStepExecution().getJobExecution()
.getJobParameters().getString("executionMode", "NORMAL");
if ("RECOLLECT".equals(executionMode)) {
log.info(">>>>> RECOLLECT 모드 - LAST_EXECUTION 업데이트 스킵");
return RepeatStatus.FINISHED;
}
String toDateStr = chunkContext.getStepContext()
.getStepExecution().getJobExecution()
.getExecutionContext().getString("batchToDate", null);
LocalDateTime successDate;
if (toDateStr != null) {
successDate = LocalDateTime.parse(toDateStr);
log.info(">>>>> BATCH_LAST_EXECUTION 업데이트 시작 (캡처된 toDate: {})", successDate);
} else {
successDate = LocalDateTime.now();
log.warn(">>>>> batchToDate가 없어 현재 시간 사용: {}", successDate);
}
jdbcTemplate.update(
String.format("UPDATE %s.BATCH_LAST_EXECUTION SET LAST_SUCCESS_DATE = ?, UPDATED_AT = NOW() WHERE API_KEY = ?", targetSchema),
Timestamp.valueOf(successDate), getApiKey()
);
log.info(">>>>> BATCH_LAST_EXECUTION 업데이트 완료 (LAST_SUCCESS_DATE = {})", successDate);
return RepeatStatus.FINISHED;
};
return new LastExecutionUpdateTasklet(jdbcTemplate, targetSchema, getApiKey(), 0);
}
@Bean(name = "StsOperationLastExecutionUpdateStep")
public Step stsOperationLastExecutionUpdateStep() {

35
src/main/java/com/snp/batch/jobs/movement/batch/config/TerminalCallsRangeJobConfig.java
파일 보기

@ -2,6 +2,7 @@ package com.snp.batch.jobs.movement.batch.config;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.snp.batch.common.batch.config.BaseMultiStepJobConfig;
import com.snp.batch.common.batch.tasklet.LastExecutionUpdateTasklet;
import com.snp.batch.jobs.movement.batch.dto.TerminalCallsDto;
import com.snp.batch.jobs.movement.batch.entity.TerminalCallsEntity;
import com.snp.batch.jobs.movement.batch.processor.TerminalCallsProcessor;
@ -20,7 +21,6 @@ import org.springframework.batch.core.step.tasklet.Tasklet;
import org.springframework.batch.item.ItemProcessor;
import org.springframework.batch.item.ItemReader;
import org.springframework.batch.item.ItemWriter;
import org.springframework.batch.repeat.RepeatStatus;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
@ -29,8 +29,6 @@ import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.transaction.PlatformTransactionManager;
import org.springframework.web.reactive.function.client.WebClient;
import java.sql.Timestamp;
import java.time.LocalDateTime;
@Slf4j
@Configuration
@ -143,36 +141,7 @@ public class TerminalCallsRangeJobConfig extends BaseMultiStepJobConfig<Terminal
*/
@Bean
public Tasklet terminalCallsLastExecutionUpdateTasklet() {
return (contribution, chunkContext) -> {
String executionMode = chunkContext.getStepContext()
.getStepExecution().getJobExecution()
.getJobParameters().getString("executionMode", "NORMAL");
if ("RECOLLECT".equals(executionMode)) {
log.info(">>>>> RECOLLECT 모드 - LAST_EXECUTION 업데이트 스킵");
return RepeatStatus.FINISHED;
}
String toDateStr = chunkContext.getStepContext()
.getStepExecution().getJobExecution()
.getExecutionContext().getString("batchToDate", null);
LocalDateTime successDate;
if (toDateStr != null) {
successDate = LocalDateTime.parse(toDateStr);
log.info(">>>>> BATCH_LAST_EXECUTION 업데이트 시작 (캡처된 toDate: {})", successDate);
} else {
successDate = LocalDateTime.now();
log.warn(">>>>> batchToDate가 없어 현재 시간 사용: {}", successDate);
}
jdbcTemplate.update(
String.format("UPDATE %s.BATCH_LAST_EXECUTION SET LAST_SUCCESS_DATE = ?, UPDATED_AT = NOW() WHERE API_KEY = ?", targetSchema),
Timestamp.valueOf(successDate), getApiKey()
);
log.info(">>>>> BATCH_LAST_EXECUTION 업데이트 완료 (LAST_SUCCESS_DATE = {})", successDate);
return RepeatStatus.FINISHED;
};
return new LastExecutionUpdateTasklet(jdbcTemplate, targetSchema, getApiKey(), 0);
}
@Bean(name = "TerminalCallsLastExecutionUpdateStep")
public Step terminalCallsLastExecutionUpdateStep() {

35
src/main/java/com/snp/batch/jobs/movement/batch/config/TransitsRangeJobConfig.java
파일 보기

@ -1,6 +1,7 @@
package com.snp.batch.jobs.movement.batch.config;
import com.snp.batch.common.batch.config.BaseMultiStepJobConfig;
import com.snp.batch.common.batch.tasklet.LastExecutionUpdateTasklet;
import com.snp.batch.jobs.movement.batch.dto.TransitsDto;
import com.snp.batch.jobs.movement.batch.entity.TransitsEntity;
import com.snp.batch.jobs.movement.batch.processor.TransitsProcessor;
@ -19,7 +20,6 @@ import org.springframework.batch.core.step.tasklet.Tasklet;
import org.springframework.batch.item.ItemProcessor;
import org.springframework.batch.item.ItemReader;
import org.springframework.batch.item.ItemWriter;
import org.springframework.batch.repeat.RepeatStatus;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
@ -28,8 +28,6 @@ import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.transaction.PlatformTransactionManager;
import org.springframework.web.reactive.function.client.WebClient;
import java.sql.Timestamp;
import java.time.LocalDateTime;
@Slf4j
@Configuration
@ -141,36 +139,7 @@ public class TransitsRangeJobConfig extends BaseMultiStepJobConfig<TransitsDto,
*/
@Bean
public Tasklet transitsLastExecutionUpdateTasklet() {
return (contribution, chunkContext) -> {
String executionMode = chunkContext.getStepContext()
.getStepExecution().getJobExecution()
.getJobParameters().getString("executionMode", "NORMAL");
if ("RECOLLECT".equals(executionMode)) {
log.info(">>>>> RECOLLECT 모드 - LAST_EXECUTION 업데이트 스킵");
return RepeatStatus.FINISHED;
}
String toDateStr = chunkContext.getStepContext()
.getStepExecution().getJobExecution()
.getExecutionContext().getString("batchToDate", null);
LocalDateTime successDate;
if (toDateStr != null) {
successDate = LocalDateTime.parse(toDateStr);
log.info(">>>>> BATCH_LAST_EXECUTION 업데이트 시작 (캡처된 toDate: {})", successDate);
} else {
successDate = LocalDateTime.now();
log.warn(">>>>> batchToDate가 없어 현재 시간 사용: {}", successDate);
}
jdbcTemplate.update(
String.format("UPDATE %s.BATCH_LAST_EXECUTION SET LAST_SUCCESS_DATE = ?, UPDATED_AT = NOW() WHERE API_KEY = ?", targetSchema),
Timestamp.valueOf(successDate), getApiKey()
);
log.info(">>>>> BATCH_LAST_EXECUTION 업데이트 완료 (LAST_SUCCESS_DATE = {})", successDate);
return RepeatStatus.FINISHED;
};
return new LastExecutionUpdateTasklet(jdbcTemplate, targetSchema, getApiKey(), 0);
}
@Bean(name = "TransitsLastExecutionUpdateStep")
public Step transitsLastExecutionUpdateStep() {

36
src/main/java/com/snp/batch/jobs/pscInspection/batch/config/PscInspectionJobConfig.java
파일 보기

@ -2,6 +2,7 @@ package com.snp.batch.jobs.pscInspection.batch.config;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.snp.batch.common.batch.config.BaseMultiStepJobConfig;
import com.snp.batch.common.batch.tasklet.LastExecutionUpdateTasklet;
import com.snp.batch.jobs.facility.batch.processor.PortDataProcessor;
import com.snp.batch.jobs.pscInspection.batch.dto.PscInspectionDto;
import com.snp.batch.jobs.pscInspection.batch.entity.PscInspectionEntity;
@ -23,7 +24,6 @@ import org.springframework.batch.core.step.tasklet.Tasklet;
import org.springframework.batch.item.ItemProcessor;
import org.springframework.batch.item.ItemReader;
import org.springframework.batch.item.ItemWriter;
import org.springframework.batch.repeat.RepeatStatus;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
@ -32,9 +32,6 @@ import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.transaction.PlatformTransactionManager;
import org.springframework.web.reactive.function.client.WebClient;
import java.sql.Timestamp;
import java.time.LocalDateTime;
@Slf4j
@Configuration
public class PscInspectionJobConfig extends BaseMultiStepJobConfig<PscInspectionDto, PscInspectionEntity> {
@ -165,36 +162,7 @@ public class PscInspectionJobConfig extends BaseMultiStepJobConfig<PscInspection
*/
@Bean
public Tasklet pscLastExecutionUpdateTasklet() {
return (contribution, chunkContext) -> {
String executionMode = chunkContext.getStepContext()
.getStepExecution().getJobExecution()
.getJobParameters().getString("executionMode", "NORMAL");
if ("RECOLLECT".equals(executionMode)) {
log.info(">>>>> RECOLLECT 모드 - LAST_EXECUTION 업데이트 스킵");
return RepeatStatus.FINISHED;
}
String toDateStr = chunkContext.getStepContext()
.getStepExecution().getJobExecution()
.getExecutionContext().getString("batchToDate", null);
LocalDateTime successDate;
if (toDateStr != null) {
successDate = LocalDateTime.parse(toDateStr).minusHours(lastExecutionBufferHours);
log.info(">>>>> BATCH_LAST_EXECUTION 업데이트 시작 (캡처된 toDate - {}시간 버퍼: {})", lastExecutionBufferHours, successDate);
} else {
successDate = LocalDateTime.now().minusHours(lastExecutionBufferHours);
log.warn(">>>>> batchToDate가 없어 현재 시간 - {}시간 버퍼 사용: {}", lastExecutionBufferHours, successDate);
}
jdbcTemplate.update(
String.format("UPDATE %s.BATCH_LAST_EXECUTION SET LAST_SUCCESS_DATE = ?, UPDATED_AT = NOW() WHERE API_KEY = ?", targetSchema),
Timestamp.valueOf(successDate), getApiKey()
);
log.info(">>>>> BATCH_LAST_EXECUTION 업데이트 완료 (LAST_SUCCESS_DATE = {})", successDate);
return RepeatStatus.FINISHED;
};
return new LastExecutionUpdateTasklet(jdbcTemplate, targetSchema, getApiKey(), lastExecutionBufferHours);
}
@Bean(name = "PSCLastExecutionUpdateStep")
public Step pscLastExecutionUpdateStep() {

36
src/main/java/com/snp/batch/jobs/risk/batch/config/RiskImportRangeJobConfig.java
파일 보기

@ -1,6 +1,7 @@
package com.snp.batch.jobs.risk.batch.config;
import com.snp.batch.common.batch.config.BaseMultiStepJobConfig;
import com.snp.batch.common.batch.tasklet.LastExecutionUpdateTasklet;
import com.snp.batch.jobs.risk.batch.dto.RiskDto;
import com.snp.batch.jobs.risk.batch.entity.RiskEntity;
import com.snp.batch.jobs.risk.batch.processor.RiskDataProcessor;
@ -21,7 +22,6 @@ import org.springframework.batch.core.step.tasklet.Tasklet;
import org.springframework.batch.item.ItemProcessor;
import org.springframework.batch.item.ItemReader;
import org.springframework.batch.item.ItemWriter;
import org.springframework.batch.repeat.RepeatStatus;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
@ -30,9 +30,6 @@ import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.transaction.PlatformTransactionManager;
import org.springframework.web.reactive.function.client.WebClient;
import java.sql.Timestamp;
import java.time.LocalDateTime;
@Slf4j
@Configuration
public class RiskImportRangeJobConfig extends BaseMultiStepJobConfig<RiskDto, RiskEntity> {
@ -157,36 +154,7 @@ public class RiskImportRangeJobConfig extends BaseMultiStepJobConfig<RiskDto, Ri
*/
@Bean
public Tasklet riskLastExecutionUpdateTasklet() {
return (contribution, chunkContext) -> {
String executionMode = chunkContext.getStepContext()
.getStepExecution().getJobExecution()
.getJobParameters().getString("executionMode", "NORMAL");
if ("RECOLLECT".equals(executionMode)) {
log.info(">>>>> RECOLLECT 모드 - LAST_EXECUTION 업데이트 스킵");
return RepeatStatus.FINISHED;
}
String toDateStr = chunkContext.getStepContext()
.getStepExecution().getJobExecution()
.getExecutionContext().getString("batchToDate", null);
LocalDateTime successDate;
if (toDateStr != null) {
successDate = LocalDateTime.parse(toDateStr).minusHours(lastExecutionBufferHours);
log.info(">>>>> BATCH_LAST_EXECUTION 업데이트 시작 (캡처된 toDate - {}시간 버퍼: {})", lastExecutionBufferHours, successDate);
} else {
successDate = LocalDateTime.now().minusHours(lastExecutionBufferHours);
log.warn(">>>>> batchToDate가 없어 현재 시간 - {}시간 버퍼 사용: {}", lastExecutionBufferHours, successDate);
}
jdbcTemplate.update(
String.format("UPDATE %s.BATCH_LAST_EXECUTION SET LAST_SUCCESS_DATE = ?, UPDATED_AT = NOW() WHERE API_KEY = ?", targetSchema),
Timestamp.valueOf(successDate), getApiKey()
);
log.info(">>>>> BATCH_LAST_EXECUTION 업데이트 완료 (LAST_SUCCESS_DATE = {})", successDate);
return RepeatStatus.FINISHED;
};
return new LastExecutionUpdateTasklet(jdbcTemplate, targetSchema, getApiKey(), lastExecutionBufferHours);
}
@Bean(name = "RiskLastExecutionUpdateStep")
public Step riskLastExecutionUpdateStep() {

53
src/main/java/com/snp/batch/jobs/shipdetail/batch/config/ShipDetailPartitioner.java
파일 보기

@ -1,61 +1,18 @@
package com.snp.batch.jobs.shipdetail.batch.config;
import lombok.extern.slf4j.Slf4j;
import org.springframework.batch.core.partition.support.Partitioner;
import org.springframework.batch.item.ExecutionContext;
import com.snp.batch.common.batch.partition.StringListPartitioner;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
/**
* 선박제원정보 IMO 목록을 N개 파티션으로 분할하는 Partitioner.
* JobExecutionContext의 allImoNumbers를 읽어 파티션별 ExecutionContext에 할당.
* 공통 {@link StringListPartitioner}에 위임하며, ExecutionContext 키로 "partitionImoNumbers"를 사용.
*/
@Slf4j
public class ShipDetailPartitioner implements Partitioner {
public class ShipDetailPartitioner extends StringListPartitioner {
private final List<String> allImoNumbers;
private final int partitionCount;
public static final String CONTEXT_KEY = "partitionImoNumbers";
public ShipDetailPartitioner(List<String> allImoNumbers, int partitionCount) {
this.allImoNumbers = allImoNumbers;
this.partitionCount = partitionCount;
}
@Override
public Map<String, ExecutionContext> partition(int gridSize) {
int totalSize = allImoNumbers.size();
// 실제 파티션 수: IMO 수가 적으면 파티션 수를 줄임
int actualPartitionCount = Math.min(partitionCount, Math.max(1, totalSize));
Map<String, ExecutionContext> partitions = new LinkedHashMap<>();
int partitionSize = (int) Math.ceil((double) totalSize / actualPartitionCount);
for (int i = 0; i < actualPartitionCount; i++) {
int fromIndex = i * partitionSize;
int toIndex = Math.min(fromIndex + partitionSize, totalSize);
if (fromIndex >= totalSize) {
break;
}
List<String> partitionImos = allImoNumbers.subList(fromIndex, toIndex);
ExecutionContext context = new ExecutionContext();
context.putString("partitionImoNumbers", String.join(",", partitionImos));
context.putInt("partitionIndex", i);
context.putInt("partitionSize", partitionImos.size());
String partitionKey = "partition" + i;
partitions.put(partitionKey, context);
log.info("[ShipDetailPartitioner] {} : IMO {} 건 (index {}-{})",
partitionKey, partitionImos.size(), fromIndex, toIndex - 1);
}
log.info("[ShipDetailPartitioner] 총 {} 개 파티션 생성 (전체 IMO: {} 건)",
partitions.size(), totalSize);
return partitions;
super(allImoNumbers, partitionCount, CONTEXT_KEY);
}
}

70
src/main/java/com/snp/batch/jobs/shipdetail/batch/config/ShipDetailUpdateJobConfig.java
파일 보기

@ -1,7 +1,8 @@
package com.snp.batch.jobs.shipdetail.batch.config;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.snp.batch.common.batch.config.BaseMultiStepJobConfig;
import com.snp.batch.common.batch.config.BasePartitionedJobConfig;
import com.snp.batch.common.batch.tasklet.LastExecutionUpdateTasklet;
import com.snp.batch.global.repository.BatchFailedRecordRepository;
import com.snp.batch.jobs.shipdetail.batch.dto.ShipDetailDto;
import com.snp.batch.jobs.shipdetail.batch.entity.ShipDetailEntity;
@ -13,17 +14,14 @@ import com.snp.batch.service.BatchDateService;
import com.snp.batch.service.BatchFailedRecordService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.batch.core.Job;
import org.springframework.batch.core.JobExecution;
import org.springframework.batch.core.JobExecutionListener;
import org.springframework.batch.core.Step;
import org.springframework.batch.core.configuration.annotation.StepScope;
import org.springframework.batch.core.job.builder.JobBuilder;
import org.springframework.batch.core.job.flow.FlowExecutionStatus;
import org.springframework.batch.core.job.flow.JobExecutionDecider;
import org.springframework.batch.core.repository.JobRepository;
import org.springframework.batch.core.step.builder.StepBuilder;
import org.springframework.batch.core.step.tasklet.Tasklet;
import org.springframework.batch.repeat.RepeatStatus;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
@ -33,15 +31,13 @@ import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.transaction.PlatformTransactionManager;
import org.springframework.web.reactive.function.client.WebClient;
import java.sql.Timestamp;
import java.time.LocalDateTime;
import java.util.Arrays;
import java.util.Collections;
import java.util.List;
@Slf4j
@Configuration
public class ShipDetailUpdateJobConfig extends BaseMultiStepJobConfig<ShipDetailDto, ShipDetailEntity> {
public class ShipDetailUpdateJobConfig extends BasePartitionedJobConfig<ShipDetailDto, ShipDetailEntity> {
private final ShipDetailDataProcessor shipDetailDataProcessor;
private final ShipDetailDataWriter shipDetailDataWriter;
@ -167,17 +163,7 @@ public class ShipDetailUpdateJobConfig extends BaseMultiStepJobConfig<ShipDetail
@Bean
public JobExecutionDecider imoCountDecider() {
return (jobExecution, stepExecution) -> {
int totalImoCount = jobExecution.getExecutionContext().getInt("totalImoCount", 0);
if (totalImoCount == 0) {
log.info("[ShipDetailUpdateJob] Decider: EMPTY_RESPONSE - IMO 0건으로 LAST_EXECUTION 업데이트 스킵");
return new FlowExecutionStatus("EMPTY_RESPONSE");
}
log.info("[ShipDetailUpdateJob] Decider: NORMAL - IMO {} 건 처리 시작", totalImoCount);
return new FlowExecutionStatus("NORMAL");
};
return createKeyCountDecider("totalImoCount", getJobName());
}
// ========================================
@ -197,12 +183,10 @@ public class ShipDetailUpdateJobConfig extends BaseMultiStepJobConfig<ShipDetail
@Bean(name = "ShipDetailUpdatePartitionedStep")
public Step shipDetailUpdatePartitionedStep() {
return new StepBuilder("ShipDetailUpdatePartitionedStep", jobRepository)
.partitioner("ShipDetailUpdateStep", shipDetailPartitioner(null))
.step(shipDetailUpdateWorkerStep())
.taskExecutor(batchPartitionExecutor)
.gridSize(partitionCount)
.build();
return createPartitionedStep(
"ShipDetailUpdatePartitionedStep", "ShipDetailUpdateStep",
shipDetailPartitioner(null), shipDetailUpdateWorkerStep(),
batchPartitionExecutor, partitionCount);
}
@Bean
@ -260,7 +244,7 @@ public class ShipDetailUpdateJobConfig extends BaseMultiStepJobConfig<ShipDetail
// ========================================
@Bean(name = "ShipDetailUpdateJob")
public Job ShipDetailUpdateJob() {
public Job shipDetailUpdateJob() {
return job();
}
@ -270,42 +254,12 @@ public class ShipDetailUpdateJobConfig extends BaseMultiStepJobConfig<ShipDetail
@Bean
public Tasklet shipDetailLastExecutionUpdateTasklet() {
return (contribution, chunkContext) -> {
String executionMode = chunkContext.getStepContext()
.getStepExecution().getJobExecution()
.getJobParameters().getString("executionMode", "NORMAL");
if ("RECOLLECT".equals(executionMode)) {
log.info(">>>>> RECOLLECT 모드 - LAST_EXECUTION 업데이트 스킵");
return RepeatStatus.FINISHED;
}
String toDateStr = chunkContext.getStepContext()
.getStepExecution().getJobExecution()
.getExecutionContext().getString("batchToDate", null);
LocalDateTime successDate;
if (toDateStr != null) {
successDate = LocalDateTime.parse(toDateStr).minusHours(lastExecutionBufferHours);
log.info(">>>>> BATCH_LAST_EXECUTION 업데이트 시작 (캡처된 toDate - {}시간 버퍼: {})", lastExecutionBufferHours, successDate);
} else {
successDate = LocalDateTime.now().minusHours(lastExecutionBufferHours);
log.warn(">>>>> batchToDate가 없어 현재 시간 - {}시간 버퍼 사용: {}", lastExecutionBufferHours, successDate);
}
jdbcTemplate.update(
String.format("UPDATE %s.BATCH_LAST_EXECUTION SET LAST_SUCCESS_DATE = ?, UPDATED_AT = NOW() WHERE API_KEY = ?", targetSchema),
Timestamp.valueOf(successDate), getApiKey()
);
log.info(">>>>> BATCH_LAST_EXECUTION 업데이트 완료 (LAST_SUCCESS_DATE = {})", successDate);
return RepeatStatus.FINISHED;
};
return new LastExecutionUpdateTasklet(jdbcTemplate, targetSchema, getApiKey(), lastExecutionBufferHours);
}
@Bean(name = "ShipDetailLastExecutionUpdateStep")
public Step shipDetailLastExecutionUpdateStep() {
return new StepBuilder("ShipDetailLastExecutionUpdateStep", jobRepository)
.tasklet(shipDetailLastExecutionUpdateTasklet(), transactionManager)
.build();
return createLastExecutionUpdateStep("ShipDetailLastExecutionUpdateStep",
shipDetailLastExecutionUpdateTasklet());
}
}