nâng cấp ws tránh lỗi lúc ghi file

KittyFallWs/KittyFallWs/src/com/vas/kittyfallwsvas/wsfw/resultrequest/ResultRequestSpoolService.java

@@ -2,15 +2,17 @@ package com.vas.kittyfallwsvas.wsfw.resultrequest;
 
 import com.google.gson.Gson;
 import java.io.BufferedReader;
-import java.io.BufferedWriter;
 import java.io.File;
 import java.io.FileReader;
 import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
 import java.nio.ByteBuffer;
 import java.nio.channels.FileChannel;
 import java.nio.charset.StandardCharsets;
 import java.nio.file.AtomicMoveNotSupportedException;
 import java.nio.file.DirectoryStream;
+import java.nio.file.FileStore;
 import java.nio.file.Files;
 import java.nio.file.Path;
 import java.nio.file.StandardCopyOption;
@@ -20,18 +22,29 @@ import java.util.List;
 import java.util.Properties;
 import java.util.concurrent.ArrayBlockingQueue;
 import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.RejectedExecutionException;
+import java.util.concurrent.RejectedExecutionHandler;
 import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.ThreadPoolExecutor;
 import java.util.concurrent.ThreadFactory;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
 import java.util.concurrent.atomic.AtomicLong;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.zip.Deflater;
+import java.util.zip.GZIPOutputStream;
 import org.apache.log4j.Logger;
 import utils.Config;
 
 /**
- * Persists accepted requests before replying and drains them to the database
- * with a single, rate-limited worker.
+ * Durable local-file queue for resultRequest callbacks.
+ *
+ * The writer is deliberately independent from the database processor: a slow
+ * database must never prevent the active spool segment from being rotated.
  */
 public class ResultRequestSpoolService {
 
@@ -48,6 +61,25 @@ public class ResultRequestSpoolService {
         }
     }
 
+    private static class ProcessingResult {
+        private final ResultRequestEvent event;
+        private final boolean retryRequired;
+        private final boolean incrementRetry;
+
+        ProcessingResult(ResultRequestEvent event, boolean retryRequired, boolean incrementRetry) {
+            this.event = event;
+            this.retryRequired = retryRequired;
+            this.incrementRetry = incrementRetry;
+        }
+    }
+
+    private static class FastGzipOutputStream extends GZIPOutputStream {
+        FastGzipOutputStream(OutputStream output) throws IOException {
+            super(output, 65536);
+            def.setLevel(Deflater.BEST_SPEED);
+        }
+    }
+
     private final Logger logger;
     private final Processor processor;
     private final Gson gson = new Gson();
@@ -58,21 +90,49 @@ public class ResultRequestSpoolService {
     private final Path processingDir;
     private final Path doneDir;
     private final Path deadDir;
-    private final Path activeFile;
     private final ArrayBlockingQueue<Pending> queue;
     private final int batchSize;
+    private final long flushIntervalMs;
     private final long enqueueTimeoutMs;
     private final long writeTimeoutMs;
     private final int maxRetry;
-    private final long rateDelayMs;
+    private final int workerCount;
+    private final int workerQueueCapacity;
+    private final long workerDelayMs;
+    private final int dbFailureThreshold;
+    private final long dbPauseMs;
+    private final long dbSlowMs;
     private final long intervalSeconds;
     private final int doneRetentionDays;
     private final int deadRetentionDays;
-    private final Object activeLock = new Object();
+    private final long segmentMaxBytes;
+    private final long segmentMaxEvents;
+    private final long segmentMaxAgeMs;
+    private final int diskWarnPercent;
+    private final int diskRejectPercent;
+    private final long diskMinFreeBytes;
+    private final long shutdownTimeoutMs;
     private final AtomicLong sequence = new AtomicLong();
+    private final AtomicLong dbPausedUntil = new AtomicLong();
+    private final AtomicLong dbConsecutivePressure = new AtomicLong();
+
+    private volatile boolean accepting = true;
     private volatile boolean running = true;
+    private volatile boolean shutdownStarted;
+    private volatile long lastDiskCheckAt;
+    private volatile long lastDiskWarningAt;
+    private volatile boolean diskRejected;
     private Thread writerThread;
-    private ScheduledExecutorService scheduler;
+    private ScheduledExecutorService processorScheduler;
+    private ScheduledExecutorService maintenanceScheduler;
+    private ExecutorService[] dbWorkers;
+
+    // These fields are owned exclusively by writerThread.
+    private FileChannel activeChannel;
+    private Path activeFile;
+    private long activeBytes;
+    private long activeEvents;
+    private long activeOpenedAt;
 
     public static ResultRequestSpoolService create(Logger logger, Processor processor) throws Exception {
         Properties properties = new Properties();
@@ -96,17 +156,34 @@ public class ResultRequestSpoolService {
         this.processingDir = root.resolve("processing");
         this.doneDir = root.resolve("done");
         this.deadDir = root.resolve("dead-letter");
-        this.activeFile = activeDir.resolve("result-request.open");
-        this.batchSize = positiveInt(properties, "RESULT_REQUEST_WRITE_BATCH_SIZE", 100);
+        this.batchSize = positiveInt(properties, "RESULT_REQUEST_WRITE_BATCH_SIZE", 1000);
+        this.flushIntervalMs = positiveLong(properties, "RESULT_REQUEST_FLUSH_INTERVAL_MS", 10L);
         this.enqueueTimeoutMs = positiveLong(properties, "RESULT_REQUEST_ENQUEUE_TIMEOUT_MS", 2000L);
         this.writeTimeoutMs = positiveLong(properties, "RESULT_REQUEST_WRITE_TIMEOUT_MS", 10000L);
         this.maxRetry = positiveInt(properties, "RESULT_REQUEST_MAX_RETRY", 20);
-        this.intervalSeconds = positiveLong(properties, "RESULT_REQUEST_INTERVAL_SECONDS", 300L);
-        this.doneRetentionDays = positiveInt(properties, "RESULT_REQUEST_DONE_RETENTION_DAYS", 60);
+        this.intervalSeconds = positiveLong(properties, "RESULT_REQUEST_INTERVAL_SECONDS", 5L);
+        this.doneRetentionDays = positiveInt(properties, "RESULT_REQUEST_DONE_RETENTION_DAYS", 2);
         this.deadRetentionDays = positiveInt(properties, "RESULT_REQUEST_DEAD_RETENTION_DAYS", 150);
-        int rate = positiveInt(properties, "RESULT_REQUEST_RATE_LIMIT_PER_SECOND", 10);
-        this.rateDelayMs = Math.max(1L, 1000L / rate);
-        this.queue = new ArrayBlockingQueue<Pending>(positiveInt(properties, "RESULT_REQUEST_QUEUE_CAPACITY", 10000));
+        this.segmentMaxBytes = positiveLong(properties, "RESULT_REQUEST_SEGMENT_MAX_BYTES", 134217728L);
+        this.segmentMaxEvents = positiveLong(properties, "RESULT_REQUEST_SEGMENT_MAX_EVENTS", 100000L);
+        this.segmentMaxAgeMs = TimeUnit.SECONDS.toMillis(
+                positiveLong(properties, "RESULT_REQUEST_SEGMENT_MAX_AGE_SECONDS", 30L));
+        this.diskWarnPercent = percentage(properties, "RESULT_REQUEST_DISK_WARN_PERCENT", 70);
+        this.diskRejectPercent = percentage(properties, "RESULT_REQUEST_DISK_REJECT_PERCENT", 90);
+        this.diskMinFreeBytes = positiveLong(properties,
+                "RESULT_REQUEST_DISK_MIN_FREE_BYTES", 21474836480L);
+        this.shutdownTimeoutMs = positiveLong(properties, "RESULT_REQUEST_SHUTDOWN_TIMEOUT_MS", 30000L);
+        int rate = positiveInt(properties, "RESULT_REQUEST_RATE_LIMIT_PER_SECOND", 600);
+        this.workerCount = positiveInt(properties, "RESULT_REQUEST_WORKER_COUNT", 8);
+        this.workerQueueCapacity = positiveInt(properties, "RESULT_REQUEST_WORKER_QUEUE_CAPACITY", 1000);
+        this.workerDelayMs = Math.max(0L, (1000L * workerCount) / rate);
+        this.dbFailureThreshold = positiveInt(properties,
+                "RESULT_REQUEST_DB_FAILURE_THRESHOLD", 16);
+        this.dbPauseMs = TimeUnit.SECONDS.toMillis(positiveLong(properties,
+                "RESULT_REQUEST_DB_PAUSE_SECONDS", 60L));
+        this.dbSlowMs = positiveLong(properties, "RESULT_REQUEST_DB_SLOW_MS", 1000L);
+        this.queue = new ArrayBlockingQueue<Pending>(
+                positiveInt(properties, "RESULT_REQUEST_QUEUE_CAPACITY", 100000));
 
         if (enabled) {
             Files.createDirectories(activeDir);
@@ -117,8 +194,14 @@ public class ResultRequestSpoolService {
             recoverFiles();
             startWriter();
             startProcessor();
+            startMaintenance();
             logger.info("resultRequest SPOOL enabled. directory=" + root.toAbsolutePath()
-                    + ", intervalSeconds=" + intervalSeconds + ", dbRate=" + rate + "/s");
+                    + ", intervalSeconds=" + intervalSeconds + ", dbRate=" + rate + "/s"
+                    + ", dbWorkers=" + workerCount
+                    + ", batchSize=" + batchSize + ", flushIntervalMs=" + flushIntervalMs
+                    + ", segmentMaxBytes=" + segmentMaxBytes
+                    + ", segmentMaxEvents=" + segmentMaxEvents
+                    + ", segmentMaxAgeMs=" + segmentMaxAgeMs);
         } else {
             logger.info("resultRequest mode is DIRECT (spool worker is not started)");
         }
@@ -132,6 +215,10 @@ public class ResultRequestSpoolService {
         if (!enabled) {
             throw new IllegalStateException("resultRequest spool is disabled");
         }
+        if (!accepting) {
+            throw new IOException("resultRequest spool is stopping");
+        }
+        ensureDiskCapacity();
         Pending pending = new Pending(event);
         if (!queue.offer(pending, enqueueTimeoutMs, TimeUnit.MILLISECONDS)) {
             throw new IOException("resultRequest spool queue is full");
@@ -143,6 +230,41 @@ public class ResultRequestSpoolService {
         }
     }
 
+    private void ensureDiskCapacity() throws IOException {
+        long now = System.currentTimeMillis();
+        if (now - lastDiskCheckAt >= 1000L) {
+            synchronized (this) {
+                if (now - lastDiskCheckAt >= 1000L) {
+                    FileStore store = Files.getFileStore(root);
+                    long total = store.getTotalSpace();
+                    long usable = store.getUsableSpace();
+                    long used = total - usable;
+                    int usedPercent = total <= 0L ? 100 : (int) ((used * 100L) / total);
+                    diskRejected = usedPercent >= diskRejectPercent || usable < diskMinFreeBytes;
+                    lastDiskCheckAt = now;
+                    if ((diskRejected || usedPercent >= diskWarnPercent)
+                            && now - lastDiskWarningAt >= TimeUnit.MINUTES.toMillis(5L)) {
+                        lastDiskWarningAt = now;
+                        String level = diskRejected ? "REJECT" : "WARN";
+                        String message = "resultRequest spool disk limit reached: used="
+                                + usedPercent + "%, usableBytes=" + usable
+                                + ", warnAt=" + diskWarnPercent + "%"
+                                + ", rejectAt=" + diskRejectPercent + "%"
+                                + ", minFreeBytes=" + diskMinFreeBytes + ", state=" + level;
+                        if (diskRejected) {
+                            logger.error(message);
+                        } else {
+                            logger.warn(message);
+                        }
+                    }
+                }
+            }
+        }
+        if (diskRejected) {
+            throw new IOException("resultRequest spool disk usage reached reject threshold");
+        }
+    }
+
     private void startWriter() {
         writerThread = new Thread(new Runnable() {
             @Override
@@ -155,59 +277,157 @@ public class ResultRequestSpoolService {
     }
 
     private void writerLoop() {
-        while (running) {
-            List<Pending> batch = new ArrayList<Pending>(batchSize);
-            try {
-                Pending first = queue.take();
-                batch.add(first);
-                queue.drainTo(batch, batchSize - 1);
-                writeBatch(batch);
-                for (Pending pending : batch) {
-                    pending.durable.complete(null);
+        try {
+            while (running || !queue.isEmpty()) {
+                List<Pending> batch = new ArrayList<Pending>(batchSize);
+                try {
+                    Pending first = queue.poll(Math.min(1000L, segmentMaxAgeMs), TimeUnit.MILLISECONDS);
+                    if (first == null) {
+                        rotateIfExpired();
+                        continue;
+                    }
+                    batch.add(first);
+                    long deadline = System.nanoTime() + TimeUnit.MILLISECONDS.toNanos(flushIntervalMs);
+                    while (batch.size() < batchSize) {
+                        long remaining = deadline - System.nanoTime();
+                        if (remaining <= 0L) {
+                            break;
+                        }
+                        Pending next = queue.poll(remaining, TimeUnit.NANOSECONDS);
+                        if (next == null) {
+                            break;
+                        }
+                        batch.add(next);
+                    }
+                    writeBatch(batch);
+                    for (Pending pending : batch) {
+                        pending.durable.complete(null);
+                    }
+                } catch (InterruptedException ex) {
+                    if (running) {
+                        logger.warn("resultRequest spool writer interrupted while running", ex);
+                    }
+                } catch (Exception ex) {
+                    logger.error("Cannot write resultRequest spool batch", ex);
+                    for (Pending pending : batch) {
+                        pending.durable.completeExceptionally(ex);
+                    }
+                    accepting = false;
+                    running = false;
+                    failQueued(ex);
+                    return;
                 }
-            } catch (InterruptedException ex) {
-                Thread.currentThread().interrupt();
-                return;
+            }
+        } finally {
+            try {
+                closeAndRotateActive();
             } catch (Exception ex) {
-                logger.error("Cannot write resultRequest spool batch", ex);
-                for (Pending pending : batch) {
-                    pending.durable.completeExceptionally(ex);
-                }
+                logger.error("Cannot close active resultRequest spool segment", ex);
             }
         }
     }
 
     private void writeBatch(List<Pending> batch) throws IOException {
-        StringBuilder content = new StringBuilder();
+        StringBuilder content = new StringBuilder(batch.size() * 512);
         for (Pending pending : batch) {
             content.append(gson.toJson(pending.event)).append('\n');
         }
         byte[] bytes = content.toString().getBytes(StandardCharsets.UTF_8);
-        synchronized (activeLock) {
-            FileChannel channel = FileChannel.open(activeFile, StandardOpenOption.CREATE,
-                    StandardOpenOption.WRITE, StandardOpenOption.APPEND);
-            try {
-                ByteBuffer buffer = ByteBuffer.wrap(bytes);
-                while (buffer.hasRemaining()) {
-                    channel.write(buffer);
-                }
-                channel.force(true);
-            } finally {
-                channel.close();
-            }
+        openActiveIfRequired();
+        if (activeEvents > 0L && (activeBytes + bytes.length > segmentMaxBytes
+                || activeEvents + batch.size() > segmentMaxEvents
+                || System.currentTimeMillis() - activeOpenedAt >= segmentMaxAgeMs)) {
+            closeAndRotateActive();
+            openActiveIfRequired();
+        }
+        ByteBuffer buffer = ByteBuffer.wrap(bytes);
+        while (buffer.hasRemaining()) {
+            activeChannel.write(buffer);
+        }
+        // A request is acknowledged only after this force succeeds.
+        activeChannel.force(true);
+        activeBytes += bytes.length;
+        activeEvents += batch.size();
+        if (activeBytes >= segmentMaxBytes || activeEvents >= segmentMaxEvents
+                || System.currentTimeMillis() - activeOpenedAt >= segmentMaxAgeMs) {
+            closeAndRotateActive();
+        }
+    }
+
+    private void openActiveIfRequired() throws IOException {
+        if (activeChannel != null) {
+            return;
+        }
+        activeFile = activeDir.resolve(fileName("result-request", ".open"));
+        activeChannel = FileChannel.open(activeFile, StandardOpenOption.CREATE_NEW,
+                StandardOpenOption.WRITE, StandardOpenOption.APPEND);
+        activeBytes = 0L;
+        activeEvents = 0L;
+        activeOpenedAt = System.currentTimeMillis();
+    }
+
+    private void rotateIfExpired() throws IOException {
+        if (activeChannel != null && activeEvents > 0L
+                && System.currentTimeMillis() - activeOpenedAt >= segmentMaxAgeMs) {
+            closeAndRotateActive();
+        }
+    }
+
+    private void closeAndRotateActive() throws IOException {
+        if (activeChannel == null) {
+            return;
+        }
+        Path completed = activeFile;
+        try {
+            activeChannel.force(true);
+        } finally {
+            activeChannel.close();
+            activeChannel = null;
+        }
+        if (activeEvents > 0L && Files.exists(completed) && Files.size(completed) > 0L) {
+            String readyName = completed.getFileName().toString().replace(".open", ".ready");
+            move(completed, readyDir.resolve(readyName));
+        } else {
+            Files.deleteIfExists(completed);
+        }
+        activeFile = null;
+        activeBytes = 0L;
+        activeEvents = 0L;
+        activeOpenedAt = 0L;
+    }
+
+    private void failQueued(Exception cause) {
+        List<Pending> failed = new ArrayList<Pending>();
+        queue.drainTo(failed);
+        for (Pending pending : failed) {
+            pending.durable.completeExceptionally(cause);
         }
     }
 
     private void startProcessor() {
-        scheduler = Executors.newSingleThreadScheduledExecutor(new ThreadFactory() {
-            @Override
-            public Thread newThread(Runnable task) {
-                Thread thread = new Thread(task, "result-request-db-worker");
-                thread.setDaemon(true);
-                return thread;
-            }
-        });
-        scheduler.scheduleWithFixedDelay(new Runnable() {
+        dbWorkers = new ExecutorService[workerCount];
+        for (int i = 0; i < workerCount; i++) {
+            dbWorkers[i] = new ThreadPoolExecutor(1, 1, 0L, TimeUnit.MILLISECONDS,
+                    new ArrayBlockingQueue<Runnable>(workerQueueCapacity),
+                    newDaemonFactory("result-request-db-partition-" + i),
+                    new RejectedExecutionHandler() {
+                @Override
+                public void rejectedExecution(Runnable task, ThreadPoolExecutor executor) {
+                    if (executor.isShutdown()) {
+                        throw new RejectedExecutionException("resultRequest DB worker is stopping");
+                    }
+                    try {
+                        executor.getQueue().put(task);
+                    } catch (InterruptedException ex) {
+                        Thread.currentThread().interrupt();
+                        throw new RejectedExecutionException("Interrupted while applying DB backpressure", ex);
+                    }
+                }
+            });
+        }
+        processorScheduler = Executors.newSingleThreadScheduledExecutor(newDaemonFactory(
+                "result-request-db-dispatcher"));
+        processorScheduler.scheduleWithFixedDelay(new Runnable() {
             @Override
             public void run() {
                 try {
@@ -219,22 +439,52 @@ public class ResultRequestSpoolService {
         }, intervalSeconds, intervalSeconds, TimeUnit.SECONDS);
     }
 
+    private void startMaintenance() {
+        maintenanceScheduler = Executors.newSingleThreadScheduledExecutor(newDaemonFactory(
+                "result-request-spool-maintenance"));
+        maintenanceScheduler.scheduleWithFixedDelay(new Runnable() {
+            @Override
+            public void run() {
+                try {
+                    cleanupExpiredFiles(doneDir, "*.done", doneRetentionDays);
+                    cleanupExpiredFiles(doneDir, "*.done.gz", doneRetentionDays);
+                    cleanupExpiredFiles(deadDir, "*.dead", deadRetentionDays);
+                } catch (Exception ex) {
+                    logger.error("resultRequest spool cleanup error", ex);
+                }
+            }
+        }, 1L, 1L, TimeUnit.HOURS);
+    }
+
+    private ThreadFactory newDaemonFactory(final String name) {
+        return new ThreadFactory() {
+            @Override
+            public Thread newThread(Runnable task) {
+                Thread thread = new Thread(task, name);
+                thread.setDaemon(true);
+                return thread;
+            }
+        };
+    }
+
     private void drainOnce() throws Exception {
         recoverProcessingFiles();
-        rotateActiveFile();
+        if (isDatabasePaused()) {
+            return;
+        }
         List<Path> files = listFiles(readyDir, "*.ready");
         for (Path ready : files) {
+            if (isDatabasePaused()) {
+                return;
+            }
             Path claimed = processingDir.resolve(ready.getFileName().toString().replace(".ready", ".processing"));
             move(ready, claimed);
             processFile(claimed);
         }
-        cleanupExpiredFiles(doneDir, "*.done", doneRetentionDays);
-        cleanupExpiredFiles(deadDir, "*.dead", deadRetentionDays);
     }
 
     private void cleanupExpiredFiles(Path directory, String glob, int retentionDays) throws IOException {
-        long cutoff = System.currentTimeMillis()
-                - TimeUnit.DAYS.toMillis((long) retentionDays);
+        long cutoff = System.currentTimeMillis() - TimeUnit.DAYS.toMillis((long) retentionDays);
         for (Path file : listFiles(directory, glob)) {
             try {
                 if (Files.getLastModifiedTime(file).toMillis() < cutoff) {
@@ -248,19 +498,16 @@ public class ResultRequestSpoolService {
         }
     }
 
-    private void rotateActiveFile() throws IOException {
-        synchronized (activeLock) {
-            if (!Files.exists(activeFile) || Files.size(activeFile) == 0) {
-                return;
-            }
-            Path ready = readyDir.resolve(fileName("batch", ".ready"));
-            move(activeFile, ready);
-        }
-    }
-
     private void processFile(Path file) throws IOException {
         BufferedReader reader = Files.newBufferedReader(file, StandardCharsets.UTF_8);
-        long lineNumber = 0;
+        long lineNumber = 0L;
+        List<Future<ProcessingResult>> results = new ArrayList<Future<ProcessingResult>>();
+        List<ResultRequestEvent> retryEvents = new ArrayList<ResultRequestEvent>();
+        List<ResultRequestEvent> deadEvents = new ArrayList<ResultRequestEvent>();
+        final AtomicBoolean[] blockedPartitions = new AtomicBoolean[workerCount];
+        for (int i = 0; i < blockedPartitions.length; i++) {
+            blockedPartitions[i] = new AtomicBoolean(false);
+        }
         try {
             String line;
             while ((line = reader.readLine()) != null) {
@@ -279,45 +526,173 @@ public class ResultRequestSpoolService {
                     writeDeadLine(line, file.getFileName().toString(), lineNumber);
                     continue;
                 }
-                try {
-                    processor.process(event, file.getFileName().toString(), lineNumber);
-                } catch (Exception ex) {
-                    logger.error("resultRequest DB processing failed eventId=" + event.getEventId(), ex);
-                    event.incrementRetryCount();
-                    if (event.getRetryCount() > maxRetry) {
-                        writeEvent(deadDir, "failed", ".dead", event);
-                    } else {
-                        writeEvent(readyDir, "retry", ".ready", event);
+                final ResultRequestEvent submittedEvent = event;
+                final String sourceFile = file.getFileName().toString();
+                final long sourceLine = lineNumber;
+                final int partition = partition(submittedEvent.getMsisdn());
+                results.add(dbWorkers[partition].submit(new java.util.concurrent.Callable<ProcessingResult>() {
+                    @Override
+                    public ProcessingResult call() throws Exception {
+                        if (blockedPartitions[partition].get() || isDatabasePaused()) {
+                            return new ProcessingResult(submittedEvent, true, false);
+                        }
+                        Exception failure = null;
+                        long startedAt = System.currentTimeMillis();
+                        try {
+                            // Business processing remains delegated to the existing Processor.
+                            processor.process(submittedEvent, sourceFile, sourceLine);
+                        } catch (Exception ex) {
+                            failure = ex;
+                            blockedPartitions[partition].set(true);
+                            logger.error("resultRequest DB processing failed eventId="
+                                    + submittedEvent.getEventId(), ex);
+                        }
+                        long elapsed = System.currentTimeMillis() - startedAt;
+                        if (failure != null) {
+                            recordDatabasePressure("failure", elapsed);
+                        } else if (elapsed >= dbSlowMs) {
+                            recordDatabasePressure("slow", elapsed);
+                        } else {
+                            dbConsecutivePressure.set(0L);
+                        }
+                        if (workerDelayMs > 0L) {
+                            Thread.sleep(workerDelayMs);
+                        }
+                        return new ProcessingResult(submittedEvent,
+                                failure != null, failure != null);
                     }
-                }
-                try {
-                    Thread.sleep(rateDelayMs);
-                } catch (InterruptedException ex) {
-                    Thread.currentThread().interrupt();
-                    throw new IOException("Spool processor interrupted", ex);
+                }));
+                if (results.size() >= workerCount * workerQueueCapacity) {
+                    collectProcessingResults(results, retryEvents, deadEvents);
+                    results.clear();
                 }
             }
         } finally {
             reader.close();
         }
-        move(file, doneDir.resolve(file.getFileName().toString().replace(".processing", ".done")));
+
+        collectProcessingResults(results, retryEvents, deadEvents);
+
+        // Persist retry/dead events before marking the source segment done.
+        writeEventSegments(readyDir, "retry", ".ready", retryEvents);
+        writeEventSegments(deadDir, "failed", ".dead", deadEvents);
+        compressDoneFile(file);
+    }
+
+    private void compressDoneFile(Path processingFile) throws IOException {
+        String doneName = processingFile.getFileName().toString()
+                .replace(".processing", ".done.gz");
+        Path doneFile = doneDir.resolve(doneName);
+        Path temp = doneFile.resolveSibling(doneName + ".tmp");
+        if (Files.exists(doneFile)) {
+            Files.delete(processingFile);
+            return;
+        }
+        Files.deleteIfExists(temp);
+        try (InputStream input = Files.newInputStream(processingFile, StandardOpenOption.READ);
+                OutputStream rawOutput = Files.newOutputStream(temp, StandardOpenOption.CREATE_NEW,
+                        StandardOpenOption.WRITE);
+                GZIPOutputStream gzip = new FastGzipOutputStream(rawOutput)) {
+            byte[] buffer = new byte[65536];
+            int read;
+            while ((read = input.read(buffer)) >= 0) {
+                if (read > 0) {
+                    gzip.write(buffer, 0, read);
+                }
+            }
+            gzip.finish();
+        }
+        FileChannel channel = FileChannel.open(temp, StandardOpenOption.WRITE);
+        try {
+            channel.force(true);
+        } finally {
+            channel.close();
+        }
+        move(temp, doneFile);
+        Files.delete(processingFile);
+    }
+
+    private void collectProcessingResults(List<Future<ProcessingResult>> results,
+            List<ResultRequestEvent> retryEvents, List<ResultRequestEvent> deadEvents) throws IOException {
+        for (Future<ProcessingResult> future : results) {
+            ProcessingResult result;
+            try {
+                result = future.get();
+            } catch (InterruptedException ex) {
+                Thread.currentThread().interrupt();
+                throw new IOException("Spool processor interrupted", ex);
+            } catch (ExecutionException ex) {
+                throw new IOException("Spool partition worker failed", ex.getCause());
+            }
+            if (result.retryRequired) {
+                if (result.incrementRetry) {
+                    result.event.incrementRetryCount();
+                }
+                if (result.event.getRetryCount() > maxRetry) {
+                    deadEvents.add(result.event);
+                } else {
+                    retryEvents.add(result.event);
+                }
+            }
+        }
+    }
+
+    private boolean isDatabasePaused() {
+        return System.currentTimeMillis() < dbPausedUntil.get();
+    }
+
+    private synchronized void recordDatabasePressure(String reason, long elapsedMs) {
+        long pressure = dbConsecutivePressure.incrementAndGet();
+        if (pressure < dbFailureThreshold || isDatabasePaused()) {
+            return;
+        }
+        long pauseUntil = System.currentTimeMillis() + dbPauseMs;
+        dbPausedUntil.set(pauseUntil);
+        dbConsecutivePressure.set(0L);
+        logger.error("resultRequest DB circuit breaker opened: reason=" + reason
+                + ", elapsedMs=" + elapsedMs + ", pauseMs=" + dbPauseMs
+                + ". Spool writer remains available.");
+    }
+
+    private int partition(String msisdn) {
+        int hash = msisdn == null ? 0 : msisdn.hashCode();
+        return (hash & 0x7fffffff) % workerCount;
     }
 
     private void recoverFiles() throws IOException {
-        if (Files.exists(activeFile) && Files.size(activeFile) > 0) {
-            move(activeFile, readyDir.resolve(fileName("recovered", ".ready")));
+        for (Path open : listFiles(activeDir, "*.open")) {
+            if (Files.size(open) > 0L) {
+                move(open, readyDir.resolve(open.getFileName().toString().replace(".open", ".ready")));
+            } else {
+                Files.deleteIfExists(open);
+            }
         }
         recoverProcessingFiles();
     }
 
     private void recoverProcessingFiles() throws IOException {
         for (Path processing : listFiles(processingDir, "*.processing")) {
-            move(processing, readyDir.resolve(fileName("recovered-processing", ".ready")));
+            String originalName = processing.getFileName().toString().replace(".processing", ".ready");
+            Path target = readyDir.resolve(originalName);
+            if (Files.exists(target)) {
+                target = readyDir.resolve(fileName("recovered-processing", ".ready"));
+            }
+            move(processing, target);
         }
     }
 
-    private void writeEvent(Path directory, String prefix, String extension, ResultRequestEvent event) throws IOException {
-        writeAtomic(directory.resolve(fileName(prefix, extension)), gson.toJson(event) + "\n");
+    private void writeEventSegments(Path directory, String prefix, String extension,
+            List<ResultRequestEvent> events) throws IOException {
+        int from = 0;
+        while (from < events.size()) {
+            int to = Math.min(events.size(), from + batchSize);
+            StringBuilder content = new StringBuilder((to - from) * 512);
+            for (int i = from; i < to; i++) {
+                content.append(gson.toJson(events.get(i))).append('\n');
+            }
+            writeAtomic(directory.resolve(fileName(prefix, extension)), content.toString());
+            from = to;
+        }
     }
 
     private void writeDeadLine(String line, String source, long lineNumber) throws IOException {
@@ -341,6 +716,49 @@ public class ResultRequestSpoolService {
         move(temp, target);
     }
 
+    public synchronized void shutdown() {
+        if (!enabled || shutdownStarted) {
+            return;
+        }
+        shutdownStarted = true;
+        accepting = false;
+        running = false;
+        if (writerThread != null) {
+            writerThread.interrupt();
+            try {
+                writerThread.join(shutdownTimeoutMs);
+            } catch (InterruptedException ex) {
+                Thread.currentThread().interrupt();
+            }
+            if (writerThread.isAlive()) {
+                logger.error("Timed out stopping resultRequest spool writer; unacknowledged requests must be retried");
+            }
+        }
+        shutdownExecutor(processorScheduler);
+        if (dbWorkers != null) {
+            for (ExecutorService worker : dbWorkers) {
+                shutdownExecutor(worker);
+            }
+        }
+        shutdownExecutor(maintenanceScheduler);
+    }
+
+    private void shutdownExecutor(ExecutorService executor) {
+        if (executor == null) {
+            return;
+        }
+        executor.shutdown();
+        try {
+            if (!executor.awaitTermination(shutdownTimeoutMs, TimeUnit.MILLISECONDS)) {
+                executor.shutdownNow();
+                executor.awaitTermination(shutdownTimeoutMs, TimeUnit.MILLISECONDS);
+            }
+        } catch (InterruptedException ex) {
+            executor.shutdownNow();
+            Thread.currentThread().interrupt();
+        }
+    }
+
     private List<Path> listFiles(Path directory, String glob) throws IOException {
         List<Path> result = new ArrayList<Path>();
         DirectoryStream<Path> stream = Files.newDirectoryStream(directory, glob);
@@ -384,9 +802,14 @@ public class ResultRequestSpoolService {
     private static long positiveLong(Properties properties, String name, long defaultValue) {
         try {
             long value = Long.parseLong(get(properties, name, String.valueOf(defaultValue)));
-            return value > 0 ? value : defaultValue;
+            return value > 0L ? value : defaultValue;
         } catch (Exception ex) {
             return defaultValue;
         }
     }
+
+    private static int percentage(Properties properties, String name, int defaultValue) {
+        int value = positiveInt(properties, name, defaultValue);
+        return value > 0 && value <= 100 ? value : defaultValue;
+    }
 }

KittyFallWs/KittyFallWs/src/com/vas/webservices/KittyFallWs.java

@@ -40,6 +40,7 @@ import java.time.LocalDate;
 import java.time.LocalDateTime;
 import java.time.format.DateTimeFormatter;
 import java.util.*;
+import javax.annotation.PreDestroy;
 import javax.annotation.Resource;
 import javax.jws.WebMethod;
 import javax.jws.WebParam;
@@ -95,12 +96,17 @@ public class KittyFallWs extends WebserviceAbstract {
             String sourceFile, long sourceLine) throws Exception {
         Timestamp expireTime = getExpireTime(event.getReceivedAt(), event.getProductExpireDays());
         int result = db.processResultRequestRenew(event, expireTime, sourceFile, sourceLine);
-        if (result == 0) {
-            logger.info("Durable resultRequest processed eventId=" + event.getEventId()
+        if (result == 0 && logger.isDebugEnabled()) {
+            logger.debug("Durable resultRequest processed eventId=" + event.getEventId()
                     + ", msisdn=" + event.getMsisdn() + ", package=" + event.getProductName());
         }
     }
 
+    @PreDestroy
+    public void shutdownResultRequestSpool() {
+        resultRequestSpool.shutdown();
+    }
+
     //
     // FOR MPS
     //
@@ -732,17 +738,19 @@ public class KittyFallWs extends WebserviceAbstract {
         try {
             String ip = getIpClient();
 //            UserInfo userInfo = null;
-            StringBuilder requestLog = new StringBuilder();
-            requestLog.append("receiveresult Params:").
-                    append("\nmsisdn:").append(msisdn).
-                    append("\nparams:").append(params).
-                    append("\nchargetime:").append(chargetime).
-                    append("\nserviceid:").append(serviceid).
-                    append("\namount:").append(fee).
-                    append("\ntransactionId:").append(transactionId).
-                    append("\ncommand:").append(command).
-                    append("\nmode:").append(mode);
-            logger.info(requestLog);
+            if (logger.isDebugEnabled()) {
+                StringBuilder requestLog = new StringBuilder();
+                requestLog.append("receiveresult Params:").
+                        append("\nmsisdn:").append(msisdn).
+                        append("\nparams:").append(params).
+                        append("\nchargetime:").append(chargetime).
+                        append("\nserviceid:").append(serviceid).
+                        append("\namount:").append(fee).
+                        append("\ntransactionId:").append(transactionId).
+                        append("\ncommand:").append(command).
+                        append("\nmode:").append(mode);
+                logger.debug(requestLog);
+            }
 
             Request request = new Request();
             request.setMsisdn(msisdn);
@@ -777,7 +785,7 @@ public class KittyFallWs extends WebserviceAbstract {
                 }
                 if (params.equals("0")) {
                     String processingRoute = resultRequestSpool.isEnabled() ? "SPOOL" : "DIRECT";
-                    logger.info("resultRequest routing: route=" + processingRoute
+                    logger.debug("resultRequest routing: route=" + processingRoute
                             + ", transactionId=" + transactionId
                             + ", msisdn=" + msisdn
                             + ", serviceId=" + serviceid);
@@ -786,13 +794,13 @@ public class KittyFallWs extends WebserviceAbstract {
                                 transactionId, serviceid, msisdn, chargetime, params,
                                 mode, fee, command, productInfo);
                         resultRequestSpool.enqueue(event);
-                        logger.info("resultRequest durably queued eventId=" + event.getEventId()
+                        logger.debug("resultRequest durably queued eventId=" + event.getEventId()
                                 + ", msisdn=" + msisdn);
                     } else {
-                        logger.info("resultRequest DIRECT processing started: transactionId="
+                        logger.debug("resultRequest DIRECT processing started: transactionId="
                                 + transactionId + ", msisdn=" + msisdn);
                         renewMps(msisdn, productInfo);
-                        logger.info("resultRequest DIRECT processing completed: transactionId="
+                        logger.debug("resultRequest DIRECT processing completed: transactionId="
                                 + transactionId + ", msisdn=" + msisdn);
                     }
                     response.setErrorCode(Common.ErrorCode.SUCCESS);

KittyFallWs/etc/app.conf

@@ -24,19 +24,27 @@ NUM_THREAD_LOG_TRANS = 1
 # chuan bi quyen ghi thu muc spool, doi thanh SPOOL roi khoi dong lai KittyFallWs.
 
 # Che do xu ly: DIRECT = ghi thang DB, SPOOL = ghi file truoc va xu ly DB sau
-RESULT_REQUEST_MODE=DIRECT
+RESULT_REQUEST_MODE=SPOOL
 
 # Thu muc luu file cho, dang xu ly, da xu ly va loi
 RESULT_REQUEST_SPOOL_DIR=../data/result-request-spool
 
-# Chu ky doc file va xu ly DB, don vi giay (300 giay = 5 phut)
-RESULT_REQUEST_INTERVAL_SECONDS=300
+# Chu ky kiem tra file ready de xu ly DB, don vi giay
+RESULT_REQUEST_INTERVAL_SECONDS=5
 
 # So luong request toi da duoc giu trong hang doi ghi file
-RESULT_REQUEST_QUEUE_CAPACITY=10000
+RESULT_REQUEST_QUEUE_CAPACITY=100000
 
 # So event toi da duoc gom trong mot lan ghi va dong bo xuong dia
-RESULT_REQUEST_WRITE_BATCH_SIZE=100
+RESULT_REQUEST_WRITE_BATCH_SIZE=1000
+
+# Ghi batch khi du so luong hoac het thoi gian cho nay, don vi mili giay
+RESULT_REQUEST_FLUSH_INTERVAL_MS=10
+
+# Dong segment khi dat mot trong ba nguong: kich thuoc, so event hoac tuoi file
+RESULT_REQUEST_SEGMENT_MAX_BYTES=134217728
+RESULT_REQUEST_SEGMENT_MAX_EVENTS=100000
+RESULT_REQUEST_SEGMENT_MAX_AGE_SECONDS=30
 
 # Thoi gian toi da cho dua request vao hang doi, don vi mili giay
 RESULT_REQUEST_ENQUEUE_TIMEOUT_MS=2000
@@ -44,14 +52,38 @@ RESULT_REQUEST_ENQUEUE_TIMEOUT_MS=2000
 # Thoi gian toi da cho du lieu duoc ghi ben vung xuong dia, don vi mili giay
 RESULT_REQUEST_WRITE_TIMEOUT_MS=10000
 
-# So lan thu lai toi da khi xu ly DB loi; vuot qua se chuyen vao dead-letter
-RESULT_REQUEST_MAX_RETRY=20
+# So lan thu lai toi da khi xu ly DB loi. Voi chu ky 5 giay, 17280 lan bao ve
+# backlog trong toi thieu 24 gio neu DB ngung hoat dong lien tuc.
+RESULT_REQUEST_MAX_RETRY=17280
+
+# Xu ly song song theo hash(msisdn); cung mot thue bao luon vao cung worker
+RESULT_REQUEST_WORKER_COUNT=8
+
+# So event cho toi da tren moi worker, tao backpressure de khong day heap
+RESULT_REQUEST_WORKER_QUEUE_CAPACITY=1000
+
+# Tong muc xu ly muc tieu. 600/s du suc ly thuyet cho khoang 51,8 trieu/ngay,
+# nhung chi tang them sau khi DBA xac nhan Oracle con du tai.
+RESULT_REQUEST_RATE_LIMIT_PER_SECOND=600
+
+# Bao ve Oracle: loi hoac cham lien tiep se tam dung doc spool, writer van ghi file
+RESULT_REQUEST_DB_FAILURE_THRESHOLD=16
+RESULT_REQUEST_DB_PAUSE_SECONDS=60
+RESULT_REQUEST_DB_SLOW_MS=1000
+
+# Canh bao som va tu choi request moi neu sap het dia. Request bi tu choi tra loi
+# de doi tac retry; request da force xuong spool khong bi mat.
+RESULT_REQUEST_DISK_WARN_PERCENT=70
+RESULT_REQUEST_DISK_REJECT_PERCENT=90
+
+# Luon chua toi thieu 20 GB trong tren volume spool, ke ca khi chua cham 90%
+RESULT_REQUEST_DISK_MIN_FREE_BYTES=21474836480
 
-# So event toi da xu ly vao DB trong mot giay de tranh DB bi qua tai
-RESULT_REQUEST_RATE_LIMIT_PER_SECOND=10
+# Thoi gian cho writer flush va dong segment khi dung WS
+RESULT_REQUEST_SHUTDOWN_TIMEOUT_MS=30000
 
-# So ngay giu file da xu ly thanh cong (60 ngay = khoang 2 thang)
-RESULT_REQUEST_DONE_RETENTION_DAYS=60
+# File thanh cong duoc nen .done.gz va giu 7 ngay
+RESULT_REQUEST_DONE_RETENTION_DAYS=7
 
 # So ngay giu file xu ly that bai (150 ngay = khoang 5 thang)
 RESULT_REQUEST_DEAD_RETENTION_DAYS=150