UPLUS-105 데이터 사용량 관측 시스템 RDB 구현

build.gradle

@@ -61,6 +61,9 @@ dependencies {
     implementation 'org.springframework.boot:spring-boot-starter-data-redis'
     runtimeOnly 'org.postgresql:postgresql'
 
+    // Batch
+    implementation 'org.springframework.boot:spring-boot-starter-batch'
+
     // Kafka
     implementation 'org.apache.kafka:kafka-clients'
     implementation 'org.springframework.kafka:spring-kafka'

src/main/java/com/project/global/config/KafkaConfig.java

@@ -2,7 +2,6 @@
 
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
-import org.springframework.kafka.annotation.EnableKafka;
 import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
 import org.springframework.kafka.config.KafkaListenerContainerFactory;
 import org.springframework.kafka.core.ConsumerFactory;
@@ -11,7 +10,6 @@
 import org.springframework.util.backoff.FixedBackOff;
 
 @Configuration
-@EnableKafka
 public class KafkaConfig {
 
     @Bean
@@ -36,14 +34,13 @@ public KafkaListenerContainerFactory<?> kafkaListenerContainerFactory(
         factory.setConsumerFactory(consumerFactory);
 
         factory.setBatchListener(true);
-
-        factory.getContainerProperties().setAckMode(ContainerProperties.AckMode.MANUAL);
+        factory.setAutoStartup(false); // ⭐ 핵심
 
         factory.getContainerProperties().setAckMode(ContainerProperties.AckMode.MANUAL);
 
         DefaultErrorHandler errorHandler =
                 new DefaultErrorHandler(
-                        new FixedBackOff(1000L, 9) // 총 10번 시도(초기+9)
+                        new FixedBackOff(1000L, 2) // 총 5번 시도(초기+9)
                         );
 
         factory.setCommonErrorHandler(errorHandler);

src/main/java/com/project/global/config/RedisConfig.java

@@ -2,11 +2,13 @@
 
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
+import org.springframework.context.annotation.Profile;
 import org.springframework.data.redis.connection.RedisConnectionFactory;
 import org.springframework.data.redis.core.RedisTemplate;
 import org.springframework.data.redis.serializer.GenericJackson2JsonRedisSerializer;
 import org.springframework.data.redis.serializer.StringRedisSerializer;
 
+@Profile("!batch")
 @Configuration
 public class RedisConfig {
 

src/main/java/com/project/global/exception/code/domain/GlobalErrorCode.java

@@ -18,7 +18,13 @@ public enum GlobalErrorCode implements BaseErrorCode {
     PLAN_CHANGE_EVENT_PRODUCE_INVALID(
             HttpStatus.BAD_REQUEST, "COMMON_007", "Kafka PlanChange 이벤트 발행 과정에서 에러가 발생했습니다"),
     LUA_SCRIPT_LOAD_INVALID(HttpStatus.BAD_REQUEST, "COMMON_008", "LUA 스크립트를 불러오는 과정에서 에러가 발생했습니다"),
-    JSON_CONVERT_INVALID(HttpStatus.BAD_REQUEST, "COMMON_009", "JSON으로 변환하는 과정에서 에러가 발생했습니다");
+    JSON_CONVERT_INVALID(HttpStatus.BAD_REQUEST, "COMMON_009", "JSON으로 변환하는 과정에서 에러가 발생했습니다"),
+    USAGE_LOG_BATCH_FAILED(HttpStatus.BAD_REQUEST, "COMMON_010", "사용자 데이터 사용량 적재 배치 시스템이 실패하였습니다"),
+    USAGE_NOTIFICATION_PRODUCER_FAILED(
+            HttpStatus.BAD_REQUEST, "COMMON_011", "UsageNotification Produce 과정에서 에러가 발생하였습니다"),
+    USAGE_OUTBOX_WRITER_FAILED(
+            HttpStatus.BAD_REQUEST, "COMMON_012", "UsageOutbox Writer 배치 시스템이 실패하였습니다"),
+    PLAN_NOT_VALID(HttpStatus.BAD_REQUEST, "COMMON_013", "존재하지 않는 요금제입니다");
 
     private final HttpStatus httpStatus;
     private final String customCode;

src/main/java/com/project/rdb/BatchJobRunner.java

@@ -0,0 +1,55 @@
+// package com.project.rdb;
+//
+// import org.springframework.batch.core.Job;
+// import org.springframework.batch.core.JobParameters;
+// import org.springframework.batch.core.JobParametersBuilder;
+// import org.springframework.batch.core.launch.JobLauncher;
+// import org.springframework.boot.CommandLineRunner;
+// import org.springframework.stereotype.Component;
+//
+// import lombok.RequiredArgsConstructor;
+//
+// @Component
+// @RequiredArgsConstructor
+// public class BatchJobRunner implements CommandLineRunner {
+//
+//    private final JobLauncher jobLauncher;
+//    private final Job usageAggregationJob;
+//    private final Job usageNotificationJob;
+//    private final Job notificationSendJob;
+//
+//    //////    @Override
+//    //////    public void run(String... args) throws Exception {
+//    //////        JobParameters params = new JobParametersBuilder()
+//    //////                .addString("fromTime", "2025-12-01T00:00:00")
+//    //////                .addString("toTime",   "2025-12-31T11:59:59")
+//    //////                .addLong("run.id", System.currentTimeMillis())
+//    //////                .toJobParameters();
+//    //////
+//    //////        jobLauncher.run(usageAggregationJob, params);
+//    //////    }
+//    ////
+//    //////    @Override
+//    //////    public void run(String... args) throws Exception {
+//    //////
+//    //////        JobParameters params = new JobParametersBuilder()
+//    //////                // 월 요금제 Step용
+//    //////                .addString("period", "202512")
+//    //////                // 일 요금제 Step용
+//    //////                .addString("usageDate", "20251201")
+//    //////                .addLong("run.id", System.currentTimeMillis())
+//    //////                .toJobParameters();
+//    //////
+//    //////        jobLauncher.run(usageNotificationJob, params);
+//    //////    }
+//    //
+//    @Override
+//    public void run(String... args) throws Exception {
+//        JobParameters params =
+//                new JobParametersBuilder()
+//                        .addLong("runAt", System.currentTimeMillis()) // 중복 실행 방지
+//                        .toJobParameters();
+//
+//        jobLauncher.run(notificationSendJob, params);
+//    }
+// }

src/main/java/com/project/rdb/batch/model/BatchStepMetricsListener.java

@@ -0,0 +1,83 @@
+package com.project.rdb.batch.model;
+
+import java.math.BigDecimal;
+import java.math.RoundingMode;
+import java.time.Duration;
+import java.time.LocalDateTime;
+import java.time.ZoneId;
+import java.util.Objects;
+
+import org.springframework.batch.core.ExitStatus;
+import org.springframework.batch.core.JobExecution;
+import org.springframework.batch.core.StepExecution;
+import org.springframework.batch.core.StepExecutionListener;
+import org.springframework.stereotype.Component;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.project.rdb.batch.model.entity.BatchExecutionReport;
+import com.project.rdb.batch.model.repository.BatchExecutionReportRepository;
+
+import lombok.RequiredArgsConstructor;
+
+@Component
+@RequiredArgsConstructor
+public class BatchStepMetricsListener implements StepExecutionListener {
+    private final BatchExecutionReportRepository reportRepository;
+    private final ObjectMapper objectMapper;
+
+    @Override
+    public ExitStatus afterStep(StepExecution stepExecution) {
+
+        JobExecution jobExecution = stepExecution.getJobExecution();
+
+        LocalDateTime start =
+                Objects.requireNonNull(stepExecution.getStartTime())
+                        .atZone(ZoneId.of("Asia/Seoul"))
+                        .toLocalDateTime();
+
+        LocalDateTime end =
+                Objects.requireNonNull(stepExecution.getEndTime())
+                        .atZone(ZoneId.of("Asia/Seoul"))
+                        .toLocalDateTime();
+
+        long durationMs = Duration.between(start, end).toMillis();
+
+        long readCount = stepExecution.getReadCount();
+        BigDecimal tps =
+                durationMs > 0
+                        ? BigDecimal.valueOf(readCount)
+                                .multiply(BigDecimal.valueOf(1000))
+                                .divide(BigDecimal.valueOf(durationMs), 2, RoundingMode.HALF_UP)
+                        : BigDecimal.ZERO;
+
+        String paramsJson;
+        try {
+            paramsJson =
+                    objectMapper.writeValueAsString(
+                            jobExecution.getJobParameters().getParameters());
+        } catch (Exception e) {
+            paramsJson = "{}";
+        }
+
+        reportRepository.save(
+                BatchExecutionReport.builder()
+                        .jobName(jobExecution.getJobInstance().getJobName())
+                        .stepName(stepExecution.getStepName())
+                        .status(stepExecution.getStatus().toString())
+                        .startedAt(start)
+                        .endedAt(end)
+                        .durationMs(durationMs)
+                        .readCount(readCount)
+                        .writeCount(stepExecution.getWriteCount())
+                        .filterCount(stepExecution.getFilterCount())
+                        .skipCount(stepExecution.getSkipCount())
+                        .commitCount(stepExecution.getCommitCount())
+                        .rollbackCount(stepExecution.getRollbackCount())
+                        .tps(tps)
+                        .jobParameters(paramsJson)
+                        .createdAt(LocalDateTime.now())
+                        .build());
+
+        return stepExecution.getExitStatus();
+    }
+}

src/main/java/com/project/rdb/batch/model/dto/NotificationSendTask.java

@@ -0,0 +1,8 @@
+package com.project.rdb.batch.model.dto;
+
+import java.util.concurrent.CompletableFuture;
+
+import org.springframework.kafka.support.SendResult;
+
+public record NotificationSendTask(
+        UsageNotificationEvent event, CompletableFuture<SendResult<String, String>> future) {}

src/main/java/com/project/rdb/batch/model/dto/UsageDailyAggregation.java

@@ -0,0 +1,3 @@
+package com.project.rdb.batch.model.dto;
+
+public record UsageDailyAggregation(Long subId, String usageDate, long deltaBytes) {}

src/main/java/com/project/rdb/batch/model/dto/UsageDailyKey.java

@@ -0,0 +1,16 @@
+package com.project.rdb.batch.model.dto;
+
+import java.util.Objects;
+
+public record UsageDailyKey(Long subId, String usageDate) {
+    @Override
+    public boolean equals(Object oj) {
+        if (this == oj) {
+            return true;
+        }
+        if (!(oj instanceof UsageDailyKey that)) {
+            return false;
+        }
+        return Objects.equals(subId, that.subId) && Objects.equals(usageDate, that.usageDate);
+    }
+}

src/main/java/com/project/rdb/batch/model/dto/UsageLogRow.java

@@ -0,0 +1,5 @@
+package com.project.rdb.batch.model.dto;
+
+import java.time.LocalDateTime;
+
+public record UsageLogRow(Long subId, Long usedBytes, LocalDateTime eventTime) {}

src/main/java/com/project/rdb/batch/model/dto/UsageMonthlyAggregation.java

@@ -0,0 +1,3 @@
+package com.project.rdb.batch.model.dto;
+
+public record UsageMonthlyAggregation(Long subId, String period, long deltaBytes) {}

src/main/java/com/project/rdb/batch/model/dto/UsageMonthlyKey.java

@@ -0,0 +1,17 @@
+package com.project.rdb.batch.model.dto;
+
+import java.util.Objects;
+
+public record UsageMonthlyKey(Long subId, String period) {
+
+    @Override
+    public boolean equals(Object oj) {
+        if (this == oj) {
+            return true;
+        }
+        if (!(oj instanceof UsageMonthlyKey that)) {
+            return false;
+        }
+        return Objects.equals(subId, that.subId) && Objects.equals(period, that.period);
+    }
+}

src/main/java/com/project/rdb/batch/model/dto/UsageNotificationCandidate.java

@@ -0,0 +1,10 @@
+package com.project.rdb.batch.model.dto;
+
+public record UsageNotificationCandidate(
+        Long subId,
+        String period,
+        String unit,
+        int threshold,
+        int percent,
+        long totalUsedMb,
+        long allotmentMb) {}

src/main/java/com/project/rdb/batch/model/dto/UsageNotificationEvent.java

@@ -0,0 +1,14 @@
+package com.project.rdb.batch.model.dto;
+
+import java.util.UUID;
+
+public record UsageNotificationEvent(
+        UUID eventId,
+        Long id,
+        Long subId,
+        String period,
+        String unit,
+        int threshold,
+        int percent,
+        long totalUsedMb,
+        long allotmentMb) {}

src/main/java/com/project/rdb/batch/model/dto/UsageNotificationOutboxRow.java

@@ -0,0 +1,11 @@
+package com.project.rdb.batch.model.dto;
+
+public record UsageNotificationOutboxRow(
+        Long id,
+        Long subId,
+        String period,
+        String unit,
+        int threshold,
+        int percent,
+        long totalUsedMb,
+        long allotmentMb) {}

src/main/java/com/project/rdb/batch/model/dto/UsageNotificationSource.java

@@ -0,0 +1,4 @@
+package com.project.rdb.batch.model.dto;
+
+public record UsageNotificationSource(
+        Long subId, String period, String unit, long totalUsedBytes, long allotmentAmount) {}