[Holliverse] 이탈률 구현 (3) - Burst Test와 첫 번째 정합성 깨짐

holliverse=> SELECT COALESCE(SUM(maf.comparison_cnt), 0) AS compare_total,
holliverse-> COALESCE(SUM(maf.change_mobile_cnt), 0) AS change_total,
holliverse-> COALESCE(SUM(maf.checked_penalty_fee_cnt), 0) AS penalty_total,
holliverse-> COALESCE(SUM(maf.comparison_cnt + maf.change_mobile_cnt + 
holliverse-> maf.checked_penalty_fee_cnt), 0) AS total_events
holliverse-> FROM feature_snapshot_store fss
holliverse-> JOIN member_action_feature maf
holliverse-> ON maf.feature_snapshot_id = fss.feature_snapshot_id
holliverse-> JOIN burst_members bm
holliverse-> ON bm.member_id = fss.member_id
holliverse-> WHERE fss.feature_type = 'MEMBER_ACTION_FEATURE';
 compare_total | change_total | penalty_total | total_events 
---------------+--------------+---------------+--------------
          9569 |         5763 |          3836 |        19168
(1 row)

@Component
@Profile("customer")
public class CustomerMetrics {

    private final MeterRegistry meterRegistry;

    public CustomerMetrics(MeterRegistry meterRegistry) {
        this.meterRegistry = meterRegistry;
    }

    public Timer.Sample startSample() {
        return Timer.start(meterRegistry);
    }

    public void recordRecommendationRequest(String outcome) {
        ...
    }

    public void stopRecommendationDuration(Timer.Sample sample, String outcome, String source) {
        ...
    }

    public void stopRecommendationWaitDuration(Timer.Sample sample, String outcome) {
        ...
    }

    public void recordRecommendationTrigger(String status) {
        ...
    }

    public void stopRecommendationKafkaConsume(Timer.Sample sample, String outcome) {
        ...
    }

    public void recordUserLogBatchSize(int size) {
        DistributionSummary.builder("holliverse.userlog.batch.size")
                .description("User-log batch payload size")
                .baseUnit("requests")
                .register(meterRegistry)
                .record(size);
    }

    public void recordUserLogPublish(String eventName, String result) {
        Counter.builder("holliverse.userlog.publish")
                .description("User-log publish results")
                .tag("event_name", eventName)
                .tag("result", result)
                .register(meterRegistry)
                .increment();
    }

    public void stopAdminLogFeatureDuration(Timer.Sample sample, String result) {
        sample.stop(Timer.builder("holliverse.userlog.admin_log_feature.duration")
                .description("Admin log-feature API call duration")
                .tag("result", result)
                .register(meterRegistry));
    }
}

@Async("userLogTaskExecutor")
public void publishBatch(Long memberId, List<UserLogRequest> requests) {
    if (requests == null || requests.isEmpty()) {
        return;
    }
    customerMetrics.recordUserLogBatchSize(requests.size());
    batchSizeSummary.record(requests.size());
    requestCounter("batch").increment();
    for (UserLogRequest request : requests) {
        doPublish(memberId, request);
    }

    requests.forEach(request -> sendAdminTarget(memberId, request));
}

@Async("userLogTaskExecutor")
public void publish(Long memberId, UserLogRequest request) {
    requestCounter("single").increment();
    doPublish(memberId, request);
    sendAdminTarget(memberId, request);
}

private void doPublish(Long memberId, UserLogRequest request) {
    UserLogEventName eventName = UserLogEventName.from(request.eventName());

    long eventId;
    try {
        eventId = decodeTsidToLong(request.tsid());
    } catch (IllegalArgumentException e) {
        customerMetrics.recordUserLogPublish(request.eventName(), "invalid_tsid");
        resultCounter("invalid_tsid").increment();
        throw new CustomerException(CustomerErrorCode.INVALID_USER_LOG_EVENT_ID);
    }

    UserLogPayload payload = new UserLogPayload(
            eventId,
            request.timestamp(),
            request.event(),
            eventName.value(),
            memberId,
            request.eventProperties()
    );

    String json;
    try {
        json = objectMapper.writeValueAsString(payload);
    } catch (JsonProcessingException e) {
        customerMetrics.recordUserLogPublish(eventName.value(), "serialization_error");
        resultCounter("serialization_error").increment();
        log.warn("[UserLog] 직렬화 실패 memberId={}", memberId, e);
        return;
    }

    kafkaTemplate.send(topic, String.valueOf(memberId), json)
            .whenComplete((result, ex) -> {
                if (ex != null) {
                    customerMetrics.recordUserLogPublish(eventName.value(), "kafka_error");
                    resultCounter("kafka_error").increment();
                    log.warn("[UserLog] Kafka 전송 실패 memberId={} eventName={}",
                            memberId, eventName.value(), ex);
                    return;
                }
                customerMetrics.recordUserLogPublish(eventName.value(), "success");
                resultCounter("kafka_success").increment();
            });
}

private Counter requestCounter(String mode) {
    return requestCounters.computeIfAbsent(mode, ignored ->
            Counter.builder("holliverse.userlog.requests")
                    .description("User log request count by mode")
                    .tag("mode", mode)
                    .register(meterRegistry));
}

private Counter resultCounter(String result) {
    return resultCounters.computeIfAbsent(result, ignored ->
            Counter.builder("holliverse.userlog.publish")
                    .description("User log publish result count")
                    .tag("result", result)
                    .register(meterRegistry));
}

holliverse=> SELECT COALESCE(SUM(maf.comparison_cnt), 0) AS compare_total,
holliverse-> COALESCE(SUM(maf.change_mobile_cnt), 0) AS change_total,
holliverse-> COALESCE(SUM(maf.checked_penalty_fee_cnt), 0) AS penalty_total,
holliverse-> COALESCE(SUM(maf.comparison_cnt + maf.change_mobile_cnt + 
holliverse-> maf.checked_penalty_fee_cnt), 0) AS total_events
holliverse-> FROM feature_snapshot_store fss
holliverse-> JOIN member_action_feature maf
holliverse-> ON maf.feature_snapshot_id = fss.feature_snapshot_id
holliverse-> JOIN burst_members bm
holliverse-> ON bm.member_id = fss.member_id
holliverse-> WHERE fss.feature_type = 'MEMBER_ACTION_FEATURE';
 compare_total | change_total | penalty_total | total_events 
---------------+--------------+---------------+--------------
          3295 |         1996 |          1344 |        6635
(1 row)

/**
 * admin log-feature 별도 executor로 분리
 */
@Service
@Profile("customer")
@RequiredArgsConstructor
public class AdminLogFeatureDispatchService {

    private final AdminLogFeaturesClient adminLogFeaturesClient;

    @Async("adminLogFeatureTaskExecutor")
    public void dispatch(Long memberId, UserLogEventName eventName, String timestamp) {
        adminLogFeaturesClient.sendLogFeature(memberId, eventName, timestamp);
    }
}

 /**
     * Admin 이벤트 변환.
     */
    private void sendAdminTarget(Long memberId, UserLogRequest request) {
        UserLogEventName eventName = UserLogEventName.from(request.eventName());
        if (!isAdminTarget(eventName)) {
            return;
        }

        adminLogFeatureDispatchService.dispatch(
                memberId,
                eventName,
                request.timestamp()
        );
    }

holliverse=> SELECT COALESCE(SUM(maf.comparison_cnt), 0) AS compare_total,
holliverse-> COALESCE(SUM(maf.change_mobile_cnt), 0) AS change_total,
holliverse-> COALESCE(SUM(maf.checked_penalty_fee_cnt), 0) AS penalty_total,
holliverse-> COALESCE(SUM(maf.comparison_cnt + maf.change_mobile_cnt + 
holliverse-> maf.checked_penalty_fee_cnt), 0) AS total_events
holliverse-> FROM feature_snapshot_store fss
holliverse-> JOIN member_action_feature maf
holliverse-> ON maf.feature_snapshot_id = fss.feature_snapshot_id
holliverse-> JOIN burst_members bm
holliverse-> ON bm.member_id = fss.member_id
holliverse-> WHERE fss.feature_type = 'MEMBER_ACTION_FEATURE';
 compare_total | change_total | penalty_total | total_events 
---------------+--------------+---------------+--------------
          4684 |         2834 |          1859 |        9377
(1 row)

@Service
@Profile("customer")
@RequiredArgsConstructor
public class AdminLogFeatureDispatchService {

    private final AdminLogFeaturesClient adminLogFeaturesClient;

    @Async("adminLogFeatureTaskExecutor")
    public void dispatch(Long memberId, UserLogEventName eventName, String timestamp) {
        adminLogFeaturesClient.sendLogFeature(memberId, eventName, timestamp);
    }
}

@Bean(name = "adminLogFeatureTaskExecutor")
public ThreadPoolTaskExecutor adminLogFeatureTaskExecutor() {
    ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
    executor.setCorePoolSize(4);
    executor.setMaxPoolSize(16);
    executor.setQueueCapacity(512);
    executor.setThreadNamePrefix("admin-log-feature-");
    executor.setRejectedExecutionHandler(
    		new java.util.concurrent.ThreadPoolExecutor.CallerRunsPolicy());
    executor.initialize();
    return executor;
}

@Bean(name = "adminLogFeatureTaskExecutor")
public ThreadPoolTaskExecutor adminLogFeatureTaskExecutor() {
    ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
    executor.setCorePoolSize(4);
    executor.setMaxPoolSize(16);
    executor.setQueueCapacity(512);
    executor.setThreadNamePrefix("admin-log-feature-");
    executor.setRejectedExecutionHandler(
    	new java.util.concurrent.ThreadPoolExecutor.AbortPolicy());
    executor.initialize();
    return executor;
}

public void recordAdminLogFeatureDispatch(String result) {
    Counter.builder("holliverse.userlog.admin_log_feature.dispatch")
            .description("Admin log-feature async dispatch enqueue results")
            .tag("result", result)
            .register(meterRegistry)
            .increment();
}

    /**
     * Admin 이벤트 변환.
     */
    private void sendAdminTarget(Long memberId, UserLogRequest request) {
        UserLogEventName eventName = UserLogEventName.from(request.eventName());
        if (!isAdminTarget(eventName)) {
            return;
        }

        try {
            adminLogFeatureDispatchService.dispatch(
                    memberId,
                    eventName,
                    request.timestamp()
            );
            customerMetrics.recordAdminLogFeatureDispatch("enqueued");
        } catch (TaskRejectedException e) {
            log.warn("[UserLog] Admin log-feature dispatch rejected. memberId={},
            	eventName={}", memberId, eventName);
            customerMetrics.recordAdminLogFeatureDispatch("rejected");
        }
    }

holliverse=> SELECT COALESCE(SUM(maf.comparison_cnt), 0) AS compare_total,COALESCE(SUM(maf.change_mobile_cnt), 0) 
holliverse-> AS change_total, COALESCE(SUM(maf.checked_penalty_fee_cnt), 0) AS penalty_total,
holliverse-> COALESCE(SUM(maf.comparison_cnt + maf.change_mobile_cnt + maf.checked_penalty_fee_cnt), 0) 
holliverse-> AS total_events FROM feature_snapshot_store fss
holliverse-> JOIN member_action_feature maf ON maf.feature_snapshot_id = fss.feature_snapshot_id 
holliverse-> JOIN burst_members bm ON bm.member_id = fss.member_id WHERE fss.feature_type = 'MEMBER_ACTION_FEATURE';
 compare_total | change_total | penalty_total | total_events 
---------------+--------------+---------------+--------------
          5623 |         3467 |          2351 |        11441
(1 row)