diff --git a/backend/coinflow-consumer-app/src/main/java/com/coinflow/aggregation/service/BatchAckWorker.java b/backend/coinflow-consumer-app/src/main/java/com/coinflow/aggregation/service/BatchAckWorker.java index 2b96bab7..e168e9e6 100644 --- a/backend/coinflow-consumer-app/src/main/java/com/coinflow/aggregation/service/BatchAckWorker.java +++ b/backend/coinflow-consumer-app/src/main/java/com/coinflow/aggregation/service/BatchAckWorker.java @@ -10,12 +10,14 @@ import java.util.List; import java.util.Map; import java.util.concurrent.BlockingQueue; -import java.util.concurrent.CompletableFuture; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.Executors; import java.util.concurrent.LinkedBlockingQueue; +import java.util.concurrent.RejectedExecutionException; import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.ScheduledFuture; import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; import lombok.RequiredArgsConstructor; import lombok.extern.slf4j.Slf4j; import org.springframework.data.redis.connection.stream.RecordId; @@ -36,6 +38,11 @@ /** * Redis Stream XACK를 배치로 처리하기 위한 워커입니다. * 틱 처리 완료 후 전달된 RecordId들을 큐에 쌓고, 일정 조건(개수 혹은 시간) 충족 시 한 번에 XACK를 호출합니다. + * + *

XACK delivery is at-least-once while this process is running: a failed in-flight batch is + * retained and retried with exponential backoff, and later batches are not drained until it + * succeeds. If the process terminates first, the records remain in the Redis PEL and require a + * separate PEL reclaim path. */ @Service @RequiredArgsConstructor @@ -49,9 +56,19 @@ public class BatchAckWorker { // 배치 설정: 10,000 ~ 100,000 TPS 대응을 위한 최적화 값 private static final int BATCH_SIZE = 500; private static final long FLUSH_INTERVAL_MS = 50; + private static final long ACK_RETRY_INITIAL_DELAY_MS = 100; + private static final long ACK_RETRY_MAX_DELAY_MS = 5_000; // 부하 분산을 위한 버퍼 큐 확장 (기존 10,000) private final BlockingQueue ackQueue = new LinkedBlockingQueue<>(50000); + private final AtomicBoolean sizeFlushScheduled = new AtomicBoolean(false); + private final Object flushScheduleMonitor = new Object(); + private ScheduledFuture intervalFlushFuture; + private long intervalScheduleGeneration; + private volatile List inFlightBatch = List.of(); + private String inFlightReason = VALUE_NA; + private int ackRetryAttempt; + private ScheduledFuture ackRetryFuture; private volatile boolean running = true; // (Point 4) 종료 상태 관리용 플래그 private final ScheduledExecutorService scheduler = Executors.newSingleThreadScheduledExecutor(r -> { Thread t = new Thread(r, "batch-ack-worker"); @@ -78,7 +95,6 @@ public void init() { TAG_FLUSH_REASON, reason)); } - scheduler.scheduleWithFixedDelay(() -> flush(VALUE_FLUSH_INTERVAL), FLUSH_INTERVAL_MS, FLUSH_INTERVAL_MS, TimeUnit.MILLISECONDS); log.info("BatchAckWorker initialized with Zero-Allocation Monitoring (BatchSize={}, Interval={}ms)", BATCH_SIZE, FLUSH_INTERVAL_MS); } @@ -88,6 +104,7 @@ public void destroy() { // 1. 새로운 요청 차단 (Point 4) this.running = false; + cancelIntervalFlush(); // 2. 종료 전 마지막 강제 Flush try { @@ -121,42 +138,182 @@ public void addAck(RecordId recordId) { log.warn("BatchAckWorker queue is full!"); } + scheduleNextFlushIfNeeded(); + } + + private void scheduleNextFlushIfNeeded() { + if (!running || !inFlightBatch.isEmpty() || ackQueue.isEmpty()) { + return; + } + if (ackQueue.size() >= BATCH_SIZE) { - CompletableFuture.runAsync(() -> flush(VALUE_FLUSH_SIZE), scheduler); + scheduleSizeFlush(); + } else { + scheduleIntervalFlush(); + } + } + + private void scheduleSizeFlush() { + if (!sizeFlushScheduled.compareAndSet(false, true)) { + return; + } + + cancelIntervalFlush(); + scheduler.execute(() -> { + try { + while (flushFullBatch()) { + // Drain every complete batch without scheduling duplicate tasks. + } + } finally { + sizeFlushScheduled.set(false); + scheduleNextFlushIfNeeded(); + } + }); + } + + private void scheduleIntervalFlush() { + synchronized (flushScheduleMonitor) { + if (!running || ackQueue.isEmpty() || sizeFlushScheduled.get() + || (intervalFlushFuture != null && !intervalFlushFuture.isDone())) { + return; + } + + long generation = ++intervalScheduleGeneration; + intervalFlushFuture = scheduler.schedule( + () -> runIntervalFlush(generation), + FLUSH_INTERVAL_MS, + TimeUnit.MILLISECONDS); + } + } + + private void runIntervalFlush(long generation) { + synchronized (flushScheduleMonitor) { + if (generation != intervalScheduleGeneration) { + return; + } + intervalFlushFuture = null; } + + flush(VALUE_FLUSH_INTERVAL); + scheduleNextFlushIfNeeded(); + } + + private void cancelIntervalFlush() { + ScheduledFuture future; + synchronized (flushScheduleMonitor) { + intervalScheduleGeneration++; + future = intervalFlushFuture; + intervalFlushFuture = null; + } + + if (future != null) { + future.cancel(false); + } + } + + private boolean flushFullBatch() { + return flush(VALUE_FLUSH_SIZE, true); } /** * 큐에 쌓인 RecordId들을 한 번에 XACK 처리합니다. */ - private synchronized void flush(String reason) { - if (ackQueue.isEmpty()) { - return; + private void flush(String reason) { + flush(reason, false); + } + + private synchronized boolean flush(String reason, boolean requireFullBatch) { + if (!inFlightBatch.isEmpty() + || ackQueue.isEmpty() + || (requireFullBatch && ackQueue.size() < BATCH_SIZE)) { + return false; } List batch = new ArrayList<>(BATCH_SIZE); ackQueue.drainTo(batch, BATCH_SIZE); - if (!batch.isEmpty()) { - String streamKey = properties.streamKey(); - String group = properties.group(); - RecordId[] ids = batch.toArray(new RecordId[0]); + if (batch.isEmpty()) { + return false; + } - try { - // 핸들을 직접 사용하여 런타임 객체 생성 최소화 - ackLatencyTimer.record(() -> { - redisTemplate.opsForStream().acknowledge(streamKey, group, ids); - ackSuccessCounter.increment(batch.size()); - - Counter commandCounter = commandCountersByReason.get(reason); - if (commandCounter != null) { - commandCounter.increment(); - } - }); - log.trace("Flushed {} ACKs in batch (reason={})", batch.size(), reason); - } catch (Exception e) { - log.error("Failed to perform Batch XACK. stream={}, group={}, reason={}", streamKey, group, reason, e); - } + inFlightBatch = List.copyOf(batch); + inFlightReason = reason; + return acknowledgeInFlight(); + } + + private boolean acknowledgeInFlight() { + List batch = inFlightBatch; + String reason = inFlightReason; + String streamKey = properties.streamKey(); + String group = properties.group(); + RecordId[] ids = batch.toArray(new RecordId[0]); + + try { + // XACK is idempotent, so retrying is safe when Redis applied the command but its response was lost. + ackLatencyTimer.record(() -> { + redisTemplate.opsForStream().acknowledge(streamKey, group, ids); + ackSuccessCounter.increment(batch.size()); + + Counter commandCounter = commandCountersByReason.get(reason); + if (commandCounter != null) { + commandCounter.increment(); + } + }); + log.trace("Flushed {} ACKs in batch (reason={})", batch.size(), reason); + + inFlightBatch = List.of(); + inFlightReason = VALUE_NA; + ackRetryAttempt = 0; + ackRetryFuture = null; + return true; + } catch (Exception e) { + log.error( + "Failed to perform Batch XACK. Preserving in-flight batch for retry. " + + "stream={}, group={}, reason={}, batchSize={}", + streamKey, group, reason, batch.size(), e); + scheduleAckRetry(); + return false; + } + } + + private void scheduleAckRetry() { + if (!running) { + log.error( + "BatchAckWorker is shutting down with an unacknowledged in-flight batch. " + + "The records remain in the Redis PEL. batchSize={}", + inFlightBatch.size()); + return; + } + + if (ackRetryFuture != null && !ackRetryFuture.isDone()) { + return; + } + + long retryDelayMs = Math.min( + ACK_RETRY_INITIAL_DELAY_MS << Math.min(ackRetryAttempt, 6), + ACK_RETRY_MAX_DELAY_MS); + ackRetryAttempt++; + + try { + ackRetryFuture = scheduler.schedule(this::retryInFlightAck, retryDelayMs, TimeUnit.MILLISECONDS); + log.warn( + "Scheduled Batch XACK retry. attempt={}, delayMs={}, batchSize={}", + ackRetryAttempt, retryDelayMs, inFlightBatch.size()); + } catch (RejectedExecutionException e) { + log.error( + "Failed to schedule Batch XACK retry. The records remain in the Redis PEL. batchSize={}", + inFlightBatch.size(), e); + } + } + + private synchronized void retryInFlightAck() { + ackRetryFuture = null; + if (inFlightBatch.isEmpty()) { + return; + } + + if (acknowledgeInFlight()) { + scheduleNextFlushIfNeeded(); } } } diff --git a/backend/coinflow-consumer-app/src/test/java/com/coinflow/aggregation/service/BatchAckWorkerTest.java b/backend/coinflow-consumer-app/src/test/java/com/coinflow/aggregation/service/BatchAckWorkerTest.java new file mode 100644 index 00000000..a17cbd2e --- /dev/null +++ b/backend/coinflow-consumer-app/src/test/java/com/coinflow/aggregation/service/BatchAckWorkerTest.java @@ -0,0 +1,201 @@ +package com.coinflow.aggregation.service; + +import com.coinflow.config.properties.TickConsumerProperties; +import com.coinflow.monitoring.MetricRecorder; +import io.micrometer.core.instrument.Counter; +import io.micrometer.core.instrument.Timer; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collections; +import java.util.List; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicLong; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.extension.ExtendWith; +import org.mockito.Mock; +import org.mockito.junit.jupiter.MockitoExtension; +import org.springframework.data.redis.connection.stream.RecordId; +import org.springframework.data.redis.core.RedisTemplate; +import org.springframework.data.redis.core.StreamOperations; + +import static com.coinflow.monitoring.constant.MetricConstants.VALUE_FLUSH_INTERVAL; +import static com.coinflow.monitoring.constant.MetricConstants.VALUE_FLUSH_SIZE; +import static org.assertj.core.api.Assertions.assertThat; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyString; +import static org.mockito.Mockito.doAnswer; +import static org.mockito.Mockito.times; +import static org.mockito.Mockito.timeout; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +@ExtendWith(MockitoExtension.class) +class BatchAckWorkerTest { + + @Mock + private RedisTemplate redisTemplate; + + @Mock + private StreamOperations streamOperations; + + @Mock + private MetricRecorder metricRecorder; + + @Mock + private Timer ackLatencyTimer; + + @Mock + private Counter defaultCounter; + + @Mock + private Counter sizeFlushCounter; + + @Mock + private Counter intervalFlushCounter; + + private BatchAckWorker worker; + + @BeforeEach + void setUp() { + TickConsumerProperties properties = new TickConsumerProperties( + "tick:raw", "tick-consumer-group", "consumer-1", 200_000L, 0.8); + + when(metricRecorder.getTimer(anyString(), any(String[].class))).thenReturn(ackLatencyTimer); + when(metricRecorder.getCounter(anyString(), any(String[].class))).thenAnswer(invocation -> { + List arguments = Arrays.asList(invocation.getArguments()); + if (arguments.contains(VALUE_FLUSH_SIZE)) { + return sizeFlushCounter; + } + if (arguments.contains(VALUE_FLUSH_INTERVAL)) { + return intervalFlushCounter; + } + return defaultCounter; + }); + doAnswer(invocation -> { + invocation.getArgument(0).run(); + return null; + }).when(ackLatencyTimer).record(any(Runnable.class)); + when(redisTemplate.opsForStream()).thenReturn(streamOperations); + + worker = new BatchAckWorker(redisTemplate, properties, metricRecorder); + worker.init(); + } + + @AfterEach + void tearDown() { + worker.destroy(); + } + + @Test + void coalescesSizeFlushRequestsAndKeepsPartialBatchForIntervalFlush() throws Exception { + CountDownLatch firstXackStarted = new CountDownLatch(1); + CountDownLatch releaseFirstXack = new CountDownLatch(1); + AtomicInteger xackCalls = new AtomicInteger(); + AtomicLong thirdXackNanos = new AtomicLong(); + List batchSizes = Collections.synchronizedList(new ArrayList<>()); + + doAnswer(invocation -> { + batchSizes.add(invocation.getArguments().length - 2); + int call = xackCalls.incrementAndGet(); + if (call == 1) { + firstXackStarted.countDown(); + assertThat(releaseFirstXack.await(1, TimeUnit.SECONDS)).isTrue(); + } else if (call == 3) { + thirdXackNanos.set(System.nanoTime()); + } + return 0L; + }).when(streamOperations).acknowledge(anyString(), anyString(), any(RecordId[].class)); + + addRecords(0, 500); + assertThat(firstXackStarted.await(1, TimeUnit.SECONDS)).isTrue(); + + // New records cross the threshold while the first size flush is still running. + addRecords(500, 600); + Thread.sleep(75); + long firstXackReleasedNanos = System.nanoTime(); + releaseFirstXack.countDown(); + + verify(sizeFlushCounter, timeout(1_000).times(2)).increment(); + verify(intervalFlushCounter, timeout(1_000).atLeastOnce()).increment(); + + assertThat(batchSizes).containsExactly(500, 500, 100); + assertThat(thirdXackNanos.get() - firstXackReleasedNanos) + .isGreaterThanOrEqualTo(TimeUnit.MILLISECONDS.toNanos(40)); + } + + @Test + void retriesFailedBatchBeforeDrainingFollowingBatches() throws Exception { + CountDownLatch firstXackFailed = new CountDownLatch(1); + CountDownLatch retryStarted = new CountDownLatch(1); + CountDownLatch releaseRetry = new CountDownLatch(1); + AtomicInteger xackCalls = new AtomicInteger(); + AtomicLong firstFailureNanos = new AtomicLong(); + AtomicLong retryStartedNanos = new AtomicLong(); + List> attemptedBatches = Collections.synchronizedList(new ArrayList<>()); + + doAnswer(invocation -> { + List ids = new ArrayList<>(); + Object[] arguments = invocation.getArguments(); + for (int index = 2; index < arguments.length; index++) { + ids.add(((RecordId) arguments[index]).getValue()); + } + attemptedBatches.add(ids); + + int call = xackCalls.incrementAndGet(); + if (call == 1) { + firstFailureNanos.set(System.nanoTime()); + firstXackFailed.countDown(); + throw new RuntimeException("Redis unavailable"); + } + if (call == 2) { + retryStartedNanos.set(System.nanoTime()); + retryStarted.countDown(); + assertThat(releaseRetry.await(1, TimeUnit.SECONDS)).isTrue(); + } + return 0L; + }).when(streamOperations).acknowledge(anyString(), anyString(), any(RecordId[].class)); + + addRecords(0, 500); + assertThat(firstXackFailed.await(1, TimeUnit.SECONDS)).isTrue(); + + addRecords(500, 600); + assertThat(retryStarted.await(1, TimeUnit.SECONDS)).isTrue(); + + assertThat(attemptedBatches).containsExactly( + expectedIds(0, 500), + expectedIds(0, 500)); + assertThat(retryStartedNanos.get() - firstFailureNanos.get()) + .isGreaterThanOrEqualTo(TimeUnit.MILLISECONDS.toNanos(80)); + + releaseRetry.countDown(); + + verify(sizeFlushCounter, timeout(2_000).times(2)).increment(); + verify(intervalFlushCounter, timeout(2_000).atLeastOnce()).increment(); + + assertThat(attemptedBatches).containsExactly( + expectedIds(0, 500), + expectedIds(0, 500), + expectedIds(500, 500), + expectedIds(1_000, 100)); + verify(defaultCounter, times(2)).increment(500.0); + verify(defaultCounter).increment(100.0); + } + + private void addRecords(long start, int count) { + for (long sequence = start; sequence < start + count; sequence++) { + worker.addAck(RecordId.of("1-" + sequence)); + } + } + + private List expectedIds(long start, int count) { + List ids = new ArrayList<>(count); + for (long sequence = start; sequence < start + count; sequence++) { + ids.add("1-" + sequence); + } + return ids; + } +}