feat: 배치 작업 안정성 개선

src/main/kotlin/com/eodigo/batch/KamisAnnualPriceBatchConfiguration.kt

@@ -20,6 +20,7 @@ import org.springframework.beans.factory.annotation.Value
 import org.springframework.context.annotation.Bean
 import org.springframework.context.annotation.Configuration
 import org.springframework.transaction.PlatformTransactionManager
+import org.springframework.web.reactive.function.client.WebClientRequestException
 
 @Configuration
 class KamisAnnualPriceBatchConfiguration(
@@ -50,6 +51,9 @@ class KamisAnnualPriceBatchConfiguration(
             .reader(kamisAnnualPriceReader())
             .processor(kamisAnnualPriceProcessor())
             .writer(kamisAnnualPriceJpaItemWriter())
+            .faultTolerant()
+            .retryLimit(3)
+            .retry(WebClientRequestException::class.java)
             .build()
     }
 
@@ -59,7 +63,7 @@ class KamisAnnualPriceBatchConfiguration(
             .name("kamisAnnualPriceReader")
             .entityManagerFactory(entityManagerFactory)
             .pageSize(CHUNK_SIZE)
-            .queryString("SELECT p FROM Product p ORDER BY p.id ASC")
+            .queryString("SELECT p FROM Product p WHERE p.categoryCode != '500' ORDER BY p.id ASC")
             .build()
     }
 

src/main/kotlin/com/eodigo/batch/KamisDailyPriceBatchConfiguration.kt

@@ -24,6 +24,7 @@ import org.springframework.beans.factory.annotation.Value
 import org.springframework.context.annotation.Bean
 import org.springframework.context.annotation.Configuration
 import org.springframework.transaction.PlatformTransactionManager
+import org.springframework.web.reactive.function.client.WebClientRequestException
 
 @Configuration
 class KamisDailyPriceBatchConfiguration(
@@ -55,6 +56,9 @@ class KamisDailyPriceBatchConfiguration(
             .reader(kamisDailyPriceApiReader(null))
             .processor(kamisDailyPriceProcessor(null))
             .writer(kamisDailyPriceJpaWriter())
+            .faultTolerant()
+            .retryLimit(3)
+            .retry(WebClientRequestException::class.java)
             .build()
     }
 
@@ -67,34 +71,8 @@ class KamisDailyPriceBatchConfiguration(
             surveyDateString?.let { LocalDate.parse(it, DateTimeFormatter.ISO_LOCAL_DATE) }
                 ?: throw IllegalArgumentException("surveyDate job parameter is required")
 
-        val categoryCodes = listOf("100", "200", "300", "400", "500", "600")
-        val regionCodes =
-            listOf(
-                "1101",
-                "2100",
-                "2200",
-                "2300",
-                "2401",
-                "2501",
-                "2601",
-                "3111",
-                "3214",
-                "3211",
-                "3311",
-                "3511",
-                "3711",
-                "3911",
-                "3113",
-                "3613",
-                "3714",
-                "3814",
-                "3145",
-                "2701",
-                "3112",
-                "3138",
-                "3411",
-                "3818",
-            )
+        val categoryCodes = productRepository.findDistinctCategoryCodes()
+        val regionCodes = regionRepository.findAllCodes()
 
         return KamisDailyPriceApiReader(
             kamisApiClient = kamisApiClient,

src/main/kotlin/com/eodigo/batch/processor/KamisAnnualPriceProcessor.kt

@@ -4,6 +4,7 @@ import com.eodigo.domain.product.entity.AnnualNationalPrice
 import com.eodigo.domain.product.entity.Product
 import com.eodigo.domain.product.enums.MarketType
 import com.eodigo.external.kamis.KamisApiClient
+import com.eodigo.external.kamis.KamisYearlyPriceItemDto
 import com.eodigo.external.kamis.KamisYearlyPriceSectionDto
 import com.fasterxml.jackson.databind.ObjectMapper
 import com.fasterxml.jackson.module.kotlin.jacksonObjectMapper
@@ -18,62 +19,112 @@ class KamisAnnualPriceProcessor(
     private val certId: String,
 ) : ItemProcessor<Product, List<AnnualNationalPrice>> {
 
+    companion object {
+        private const val API_CALL_DELAY_MS = 200L
+        private const val RETAIL_CODE = "01"
+        private const val RANK_STANDARD = "상품"
+    }
+
     private val log = LoggerFactory.getLogger(javaClass)
     private val objectMapper: ObjectMapper = jacksonObjectMapper()
 
     override fun process(item: Product): List<AnnualNationalPrice>? {
-        val currentYear = LocalDate.now().year.toString()
         val kindCode = item.kindCode ?: return null
 
+        val currentYear = LocalDate.now().year.toString()
+
+        try {
+            // 1. API 호출
+            val responseString = fetchAnnualPriceData(item.itemCode, kindCode, currentYear)
+            if (responseString == null) {
+                return null
+            }
+
+            // 2. 응답 파싱 및 필터링
+            val priceItems = parseAndFilterPriceItems(responseString)
+            if (priceItems == null) {
+                return null
+            }
+
+            // 3. 엔티티로 변환
+            return convertToEntities(priceItems, item, currentYear)
+        } catch (e: InterruptedException) {
+            log.warn("Annual price processing was interrupted for product id: ${item.id}", e)
+            Thread.currentThread().interrupt()
+            return null
+        } catch (e: Exception) {
+            log.error("Failed to process annual price for product id: {}", item.id, e)
+            throw e
+        }
+    }
+
+    /** API를 호출하고 응답 문자열을 반환합니다. 유효하지 않은 응답은 null을 반환합니다. */
+    private fun fetchAnnualPriceData(
+        itemCode: String,
+        kindCode: String,
+        currentYear: String,
+    ): String? {
         val responseString =
             kamisApiClient.getYearlyPrices(
                 certKey = apiKey,
                 certId = certId,
                 year = currentYear,
-                itemCode = item.itemCode,
+                itemCode = itemCode,
                 kindCode = kindCode,
             )
 
+        Thread.sleep(API_CALL_DELAY_MS)
+
         if (responseString.isNullOrBlank() || responseString.contains("결과가 존재하지 않습니다.")) {
             return null
         }
+        return responseString
+    }
 
+    /** API 응답을 파싱하여 유효한 가격 아이템 리스트를 반환합니다. 유효 데이터가 없으면 null을 반환합니다. */
+    private fun parseAndFilterPriceItems(responseString: String): List<KamisYearlyPriceItemDto>? {
         val priceSections = normalizePriceSections(responseString)
-
         if (priceSections.isNullOrEmpty()) {
             return null
         }
 
-        // 3. 소매 가격/'상품' 등급인 첫 번째 섹션을 찾음
         val retailPriceSection =
-            priceSections.find { it.productClsCode == "01" && it.caption?.contains("상품") == true }
+            priceSections.find {
+                it.productClsCode == RETAIL_CODE && it.caption?.contains(RANK_STANDARD) == true
+            }
 
-        if (retailPriceSection?.items == null) {
-            return null
-        }
+        return retailPriceSection?.items
+    }
 
-        // API 응답 item을 엔티티 리스트로 변환
-        return retailPriceSection.items.mapNotNull { priceItem ->
-            val yearStr = priceItem.div
-            val priceStr = priceItem.avgData?.replace(",", "")
-
-            if (
-                yearStr != null &&
-                    yearStr.all { it.isDigit() } &&
-                    !priceStr.isNullOrBlank() &&
-                    priceStr.all { it.isDigit() }
-            ) {
-
-                AnnualNationalPrice(
-                    product = item,
-                    price = priceStr.toInt(),
-                    surveyYear = yearStr.toInt(),
-                    marketType = MarketType.RETAIL,
-                )
-            } else {
-                null
-            }
-        }
+    /** DTO 리스트를 엔티티 리스트로 변환합니다. 현재 연도의 데이터만 필터링합니다. */
+    private fun convertToEntities(
+        priceItems: List<KamisYearlyPriceItemDto>,
+        product: Product,
+        currentYear: String,
+    ): List<AnnualNationalPrice>? {
+        val entities =
+            priceItems
+                .filter { it.div == currentYear }
+                .mapNotNull { priceItem ->
+                    val yearStr = priceItem.div
+                    val priceStr = priceItem.avgData?.replace(",", "")
+
+                    if (
+                        yearStr != null &&
+                            priceStr?.all(Char::isDigit) == true &&
+                            priceStr.isNotEmpty()
+                    ) {
+                        AnnualNationalPrice(
+                            product = product,
+                            price = priceStr.toInt(),
+                            surveyYear = yearStr.toInt(),
+                            marketType = MarketType.RETAIL,
+                        )
+                    } else {
+                        null
+                    }
+                }
+        return entities.ifEmpty { null }
     }
 
     private fun normalizePriceSections(responseString: String): List<KamisYearlyPriceSectionDto>? {

src/main/kotlin/com/eodigo/batch/processor/KamisDailyPriceProcessor.kt

@@ -18,6 +18,10 @@ class KamisDailyPriceProcessor(
     private val surveyDate: LocalDate,
 ) : ItemProcessor<KamisDailyPriceApiData, DailyRegionalPrice> {
 
+    companion object {
+        private const val RANK_CODE_STANDARD = "04"
+    }
+
     private val log = LoggerFactory.getLogger(javaClass)
 
     // 1. Product와 Region 정보를 메모리에 캐싱
@@ -34,7 +38,7 @@ class KamisDailyPriceProcessor(
         // 2. 가격 정보가 유효하지 않으면 필터링
         val item = wrapper.item
 
-        if (item.rank != "상품") {
+        if (item.rankCode != RANK_CODE_STANDARD) {
             return null
         }
 

src/main/kotlin/com/eodigo/batch/reader/KamisDailyPriceApiReader.kt

@@ -4,6 +4,7 @@ import com.eodigo.batch.dto.KamisDailyPriceApiData
 import com.eodigo.external.kamis.KamisApiClient
 import java.time.LocalDate
 import java.time.format.DateTimeFormatter
+import org.slf4j.LoggerFactory
 import org.springframework.batch.item.ItemReader
 
 class KamisDailyPriceApiReader(
@@ -15,6 +16,13 @@ class KamisDailyPriceApiReader(
     private val surveyDate: LocalDate,
 ) : ItemReader<KamisDailyPriceApiData> {
 
+    companion object {
+        private const val API_CALL_DELAY_MS = 200L
+        private const val ERROR_CODE_SUCCESS = "000"
+    }
+
+    private val log = LoggerFactory.getLogger(javaClass)
+
     // 1. 처리할 모든 API 호출 결과(가격 아이템)를 담아둘 리스트
     private var allItems: MutableList<KamisDailyPriceApiData>? = null
     // 2. 현재 읽고 있는 아이템의 인덱스
@@ -45,26 +53,47 @@ class KamisDailyPriceApiReader(
 
         for (categoryCode in categoryCodes) {
             for (regionCode in regionCodes) {
-                val response =
-                    kamisApiClient.getDailyPrices(
-                        certKey = apiKey,
-                        certId = certId,
-                        regDay = regDay,
-                        categoryCode = categoryCode,
-                        countryCode = regionCode,
-                    )
+                try {
+                    val response =
+                        kamisApiClient.getDailyPrices(
+                            certKey = apiKey,
+                            certId = certId,
+                            regDay = regDay,
+                            categoryCode = categoryCode,
+                            countryCode = regionCode,
+                        )
 
-                if (response?.data?.errorCode == "000") {
-                    response.data.items?.let { items ->
-                        val wrappedItems =
-                            items.map { item ->
-                                KamisDailyPriceApiData(regionCode = regionCode, item = item)
-                            }
-                        allItems?.addAll(wrappedItems)
+                    if (response?.data?.errorCode == ERROR_CODE_SUCCESS) {
+                        response.data.items?.let { items ->
+                            val wrappedItems =
+                                items.map { item ->
+                                    KamisDailyPriceApiData(regionCode = regionCode, item = item)
+                                }
+                            allItems?.addAll(wrappedItems)
+                        }
+                    }
+                } catch (e: InterruptedException) {
+                    log.warn("API call delay was interrupted", e)
+                    Thread.currentThread().interrupt()
+                    break
+                } catch (e: Exception) {
+                    log.error(
+                        "Failed to fetch data for category: {}, region: {}. Error: {}",
+                        categoryCode,
+                        regionCode,
+                        e.message,
+                    )
+                    throw e
+                } finally {
+                    try {
+                        Thread.sleep(API_CALL_DELAY_MS)
+                    } catch (ie: InterruptedException) {
+                        log.warn("API call delay was interrupted", ie)
+                        Thread.currentThread().interrupt()
                     }
                 }
             }
         }
-        println("KamisDailyPriceApiReader: Total ${allItems?.size} items initialized.")
+        log.info("KamisDailyPriceApiReader: Total ${allItems?.size} items initialized.")
     }
 }

src/main/kotlin/com/eodigo/batch/scheduler/KamisJobScheduler.kt

@@ -2,6 +2,7 @@ package com.eodigo.batch.scheduler
 
 import java.time.LocalDate
 import java.time.LocalDateTime
+import java.time.ZoneId
 import java.time.format.DateTimeFormatter
 import org.slf4j.LoggerFactory
 import org.springframework.batch.core.Job
@@ -26,8 +27,10 @@ class KamisJobScheduler(
     fun runDailyPriceSyncJob() {
         log.info("Starting kamisDailyPriceSyncJob by scheduler...")
         try {
-            val surveyDate = LocalDate.now().minusDays(1).format(DateTimeFormatter.ISO_LOCAL_DATE)
-
+            val surveyDate =
+                LocalDate.now(ZoneId.of("Asia/Seoul"))
+                    .minusDays(1)
+                    .format(DateTimeFormatter.ISO_LOCAL_DATE)
             val jobParameters =
                 JobParametersBuilder()
                     .addLocalDateTime("run.time", LocalDateTime.now())

src/main/kotlin/com/eodigo/domain/product/repository/ProductRepository.kt

@@ -3,6 +3,7 @@ package com.eodigo.domain.product.repository
 import com.eodigo.domain.product.entity.Product
 import com.eodigo.domain.product.enums.ProductSource
 import org.springframework.data.jpa.repository.JpaRepository
+import org.springframework.data.jpa.repository.Query
 
 interface ProductRepository : JpaRepository<Product, Long> {
     fun existsBySource(source: ProductSource): Boolean
@@ -12,4 +13,7 @@ interface ProductRepository : JpaRepository<Product, Long> {
     fun findAllByOrderByIdAsc(): List<Product>
 
     fun findByNameContaining(keyword: String): List<Product>
+
+    @Query("SELECT DISTINCT p.categoryCode FROM Product p")
+    fun findDistinctCategoryCodes(): List<String>
 }

src/main/kotlin/com/eodigo/domain/product/repository/RegionRepository.kt

@@ -2,5 +2,8 @@ package com.eodigo.domain.product.repository
 
 import com.eodigo.domain.product.entity.Region
 import org.springframework.data.jpa.repository.JpaRepository
+import org.springframework.data.jpa.repository.Query
 
-interface RegionRepository : JpaRepository<Region, Long>
+interface RegionRepository : JpaRepository<Region, Long> {
+    @Query("SELECT r.code FROM Region r") fun findAllCodes(): List<String>
+}