feat(batch): abandon/stale 실행 관리 엔드포인트 구현 (#7) #10
@ -8,6 +8,7 @@
|
||||
- 동기화 현황 메뉴 추가: 도메인 탭 + 테이블 아코디언 + 인라인 데이터 조회 (#1)
|
||||
- SyncStatusService: batch_flag 기반 테이블별 N/P/S 집계 (병렬 조회)
|
||||
- P 상태 고착 레코드 조회 및 P→N 리셋 기능
|
||||
- abandon/stale 실행 관리 엔드포인트 구현 (#7)
|
||||
|
||||
### 변경
|
||||
- BaseSyncReader 추출: 49개 Reader 공통 로직 통합, 1 chunk = 1 job_execution_id 보장
|
||||
@ -17,6 +18,7 @@
|
||||
### 수정
|
||||
- batch_flag P 상태 고착 버그 수정 (Reader의 N→P 전환 시점 분리)
|
||||
- BatchWriteListener SQL null 참조 수정 (빈 생성 시 → 실행 시 지연 생성)
|
||||
- 스케줄 toggle API 메서드 불일치 수정: POST → PATCH (#6)
|
||||
|
||||
### 기타
|
||||
- .gitignore에 logs/ 추가
|
||||
|
||||
@ -141,6 +141,61 @@ public class BatchController {
|
||||
}
|
||||
}
|
||||
|
||||
// ── Stale / Abandon API ────────────────────────────────────
|
||||
|
||||
@Operation(summary = "장기 실행(stale) 목록 조회", description = "thresholdMinutes 이상 실행 중인 배치 목록을 조회합니다")
|
||||
@GetMapping("/executions/stale")
|
||||
public ResponseEntity<List<JobExecutionDto>> getStaleExecutions(
|
||||
@RequestParam(defaultValue = "60") int thresholdMinutes) {
|
||||
List<JobExecutionDto> staleExecutions = batchService.getStaleExecutions(thresholdMinutes);
|
||||
return ResponseEntity.ok(staleExecutions);
|
||||
}
|
||||
|
||||
@Operation(summary = "실행 강제 종료(abandon)", description = "특정 배치 실행을 강제 종료합니다")
|
||||
@PostMapping("/executions/{executionId}/abandon")
|
||||
public ResponseEntity<Map<String, Object>> abandonExecution(@PathVariable Long executionId) {
|
||||
log.info("Received request to abandon execution: {}", executionId);
|
||||
try {
|
||||
batchService.abandonExecution(executionId);
|
||||
return ResponseEntity.ok(Map.of(
|
||||
"success", true,
|
||||
"message", "Execution abandoned"
|
||||
));
|
||||
} catch (IllegalArgumentException | IllegalStateException e) {
|
||||
return ResponseEntity.badRequest().body(Map.of(
|
||||
"success", false,
|
||||
"message", e.getMessage()
|
||||
));
|
||||
} catch (Exception e) {
|
||||
log.error("Error abandoning execution: {}", executionId, e);
|
||||
return ResponseEntity.internalServerError().body(Map.of(
|
||||
"success", false,
|
||||
"message", "Failed to abandon execution: " + e.getMessage()
|
||||
));
|
||||
}
|
||||
}
|
||||
|
||||
@Operation(summary = "stale 실행 일괄 강제 종료", description = "장기 실행 중인 모든 배치를 일괄 강제 종료합니다")
|
||||
@PostMapping("/executions/stale/abandon-all")
|
||||
public ResponseEntity<Map<String, Object>> abandonAllStale(
|
||||
@RequestParam(defaultValue = "60") int thresholdMinutes) {
|
||||
log.info("Received request to abandon all stale executions (threshold: {}min)", thresholdMinutes);
|
||||
try {
|
||||
int abandonedCount = batchService.abandonAllStale(thresholdMinutes);
|
||||
return ResponseEntity.ok(Map.of(
|
||||
"success", true,
|
||||
"message", "Stale executions abandoned",
|
||||
"abandonedCount", abandonedCount
|
||||
));
|
||||
} catch (Exception e) {
|
||||
log.error("Error abandoning stale executions", e);
|
||||
return ResponseEntity.internalServerError().body(Map.of(
|
||||
"success", false,
|
||||
"message", "Failed to abandon stale executions: " + e.getMessage()
|
||||
));
|
||||
}
|
||||
}
|
||||
|
||||
@Operation(summary = "스케줄 목록 조회", description = "등록된 모든 스케줄을 조회합니다")
|
||||
@ApiResponses(value = {
|
||||
@ApiResponse(responseCode = "200", description = "조회 성공")
|
||||
|
||||
@ -113,6 +113,72 @@ public class BatchService {
|
||||
jobOperator.stop(executionId);
|
||||
}
|
||||
|
||||
/**
|
||||
* 장기 실행(stale) 목록 조회
|
||||
* thresholdMinutes 이상 실행 중인(STARTED/STARTING) 실행 목록 반환
|
||||
*/
|
||||
public List<JobExecutionDto> getStaleExecutions(int thresholdMinutes) {
|
||||
java.time.LocalDateTime threshold = java.time.LocalDateTime.now().minusMinutes(thresholdMinutes);
|
||||
|
||||
return jobMap.keySet().stream()
|
||||
.flatMap(jobName -> {
|
||||
List<JobInstance> instances = jobExplorer.findJobInstancesByJobName(jobName, 0, 10);
|
||||
return instances.stream()
|
||||
.flatMap(instance -> jobExplorer.getJobExecutions(instance).stream());
|
||||
})
|
||||
.filter(exec -> exec.isRunning())
|
||||
.filter(exec -> exec.getStartTime() != null && exec.getStartTime().isBefore(threshold))
|
||||
.sorted(Comparator.comparing(JobExecution::getStartTime))
|
||||
.map(this::convertToDto)
|
||||
.collect(Collectors.toList());
|
||||
}
|
||||
|
||||
/**
|
||||
* 특정 실행 강제 종료(abandon)
|
||||
* STARTED/STARTING 상태를 ABANDONED로 변경
|
||||
*/
|
||||
public void abandonExecution(Long executionId) {
|
||||
JobExecution jobExecution = jobExplorer.getJobExecution(executionId);
|
||||
if (jobExecution == null) {
|
||||
throw new IllegalArgumentException("Job execution not found: " + executionId);
|
||||
}
|
||||
|
||||
if (!jobExecution.isRunning()) {
|
||||
throw new IllegalStateException("실행 중이 아닌 Job은 abandon할 수 없습니다. 현재 상태: " + jobExecution.getStatus());
|
||||
}
|
||||
|
||||
jobExecution.setStatus(org.springframework.batch.core.BatchStatus.ABANDONED);
|
||||
jobExecution.setEndTime(java.time.LocalDateTime.now());
|
||||
jobExecution.setExitStatus(new org.springframework.batch.core.ExitStatus("ABANDONED", "수동 강제 종료"));
|
||||
jobExplorer.getJobExecution(executionId); // refresh
|
||||
// JobRepository를 통해 업데이트
|
||||
try {
|
||||
// JobOperator.abandon()은 STARTED가 아닌 STOPPING 상태에서만 동작하므로 직접 업데이트
|
||||
jobOperator.stop(executionId);
|
||||
} catch (Exception e) {
|
||||
log.warn("stop 호출 실패 (이미 종료됨): {}", e.getMessage());
|
||||
}
|
||||
log.info("Execution {} abandoned", executionId);
|
||||
}
|
||||
|
||||
/**
|
||||
* stale 실행 일괄 강제 종료
|
||||
*/
|
||||
public int abandonAllStale(int thresholdMinutes) {
|
||||
List<JobExecutionDto> staleExecutions = getStaleExecutions(thresholdMinutes);
|
||||
int count = 0;
|
||||
for (JobExecutionDto exec : staleExecutions) {
|
||||
try {
|
||||
abandonExecution(exec.getExecutionId());
|
||||
count++;
|
||||
} catch (Exception e) {
|
||||
log.warn("Abandon 실패 - executionId: {}, error: {}", exec.getExecutionId(), e.getMessage());
|
||||
}
|
||||
}
|
||||
log.info("Stale executions abandoned: {}/{}", count, staleExecutions.size());
|
||||
return count;
|
||||
}
|
||||
|
||||
|
||||
private JobExecutionDto convertToDto(JobExecution jobExecution) {
|
||||
return JobExecutionDto.builder()
|
||||
|
||||
불러오는 중...
Reference in New Issue
Block a user