Asynchronous Queue & Decoupled Workers (Kafka/RabbitMQ & S3 Storage)

package com.enterprise.report.worker;

import net.sf.jasperreports.engine.*;
import net.sf.jasperreports.engine.fill.JRSwapFileVirtualizer;
import net.sf.jasperreports.engine.util.JRSwapFile;
import net.sf.jasperreports.export.SimpleExporterInput;
import net.sf.jasperreports.export.SimpleOutputStreamExporterOutput;
import net.sf.jasperreports.pdf.JRPdfExporter;
import org.redisson.api.RLock;
import org.redisson.api.RedissonClient;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Service;
import software.amazon.awssdk.core.sync.RequestBody;
import software.amazon.awssdk.services.s3.S3Client;
import software.amazon.awssdk.services.s3.model.PutObjectRequest;

import java.io.*;
import java.nio.file.Files;
import java.util.Map;
import java.util.concurrent.TimeUnit;

@Service
public class AsynchronousReportWorkerService {

    private final RedissonClient redissonClient;
    private final S3Client s3Client;
    private final JRSwapFile sharedSwapFile;

    public record ReportTaskMessage(String jobId, String reportName, Map params, String s3Bucket, String s3Key) {}

    public AsynchronousReportWorkerService(RedissonClient redissonClient, S3Client s3Client) {
        this.redissonClient = redissonClient;
        this.s3Client = s3Client;
        // Shared Paged Swap File บน NVMe storage บล็อกละ 4KB เติบโตครั้งละ 1024 บล็อก
        this.sharedSwapFile = new JRSwapFile("/tmp/jasper_swap", 4096, 1024);
    }

    @KafkaListener(topics = "report.generation.tasks", groupId = "report-worker-cluster", concurrency = "4")
    public void processReportGenerationTask(ReportTaskMessage task) throws Exception {
        // 1. Idempotency Lock: ป้องกันการประมวลผลซ้ำเมื่อเกิด Rebalance หรือ Retry
        RLock lock = redissonClient.getLock("lock:report:job:" + task.jobId());
        boolean acquired = lock.tryLock(5, 600, TimeUnit.SECONDS);
        if (!acquired) {
            // มี Worker ตัวอื่นกำลังประมวลผลงานนี้อยู่แล้ว ข้ามทันที
            return;
        }

        // 2. ควบคุม Heap ด้วย Virtualizer: ค้างใน RAM ไม่เกิน 100 หน้า ส่วนเกินลงดิสก์
        JRSwapFileVirtualizer virtualizer = new JRSwapFileVirtualizer(100, sharedSwapFile, false);
        File tempExportFile = File.createTempFile("async_report_" + task.jobId(), ".pdf");

        try {
            task.params().put(JRParameter.REPORT_VIRTUALIZER, virtualizer);

            // 3. โหลด Pre-compiled Template จาก Classpath
            InputStream templateStream = getClass().getResourceAsStream("/reports/" + task.reportName() + ".jasper");
            JasperReport jasperReport = (JasperReport) net.sf.jasperreports.engine.util.JRLoader.loadObject(templateStream);

            // 4. Fill phase: ปลอดภัยจาก OOM แม้รายงานมีขนาดนับล้านแถว
            JRDataSource dataSource = fetchDataSourceForTask(task);
            JasperPrint jasperPrint = JasperFillManager.fillReport(jasperReport, task.params(), dataSource);

            // 5. Export phase ไปป์ลง Temporary File ก่อนส่งขึ้น Cloud
            try (OutputStream fileOut = new BufferedOutputStream(new FileOutputStream(tempExportFile))) {
                JRPdfExporter exporter = new JRPdfExporter();
                exporter.setExporterInput(new SimpleExporterInput(jasperPrint));
                exporter.setExporterOutput(new SimpleOutputStreamExporterOutput(fileOut));
                exporter.exportReport();
            }

            // 6. อัปโหลดผลลัพธ์ตรงเข้าสู่ S3 / MinIO
            s3Client.putObject(
                PutObjectRequest.builder().bucket(task.s3Bucket()).key(task.s3Key()).contentType("application/pdf").build(),
                RequestBody.fromFile(tempExportFile)
            );

            // 7. แจ้งเตือนผลลัพธ์ความสำเร็จไปยัง Notification Dispatcher
            notifyJobCompletion(task.jobId(), task.s3Key());

        } finally {
            // สำคัญที่สุด: คืนบล็อกดิสก์ใน Swap File และลบไฟล์ชั่วคราวเสมอ
            virtualizer.cleanup();
            Files.deleteIfExists(tempExportFile.toPath());
            if (lock.isHeldByCurrentThread()) {
                lock.unlock();
            }
        }
    }

    private JRDataSource fetchDataSourceForTask(ReportTaskMessage task) {
        // เชื่อมต่อไปยัง Data Lake / Read-Replica DB
        return new JREmptyDataSource(500000); 
    }

    private void notifyJobCompletion(String jobId, String s3Key) {
        // ส่งข้อความแจ้งเตือนผ่าน SSE หรือ WebSocket
    }
}

<?xml version="1.0" encoding="UTF-8"?>
<jasperReport xmlns="http://jasperreports.sourceforge.net/jasperreports"
              xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
              xsi:schemaLocation="http://jasperreports.sourceforge.net/jasperreports http://jasperreports.sourceforge.net/xsd/jasperreport.xsd"
              name="AsyncOmnibusSettlementLedger"
              pageWidth="595" pageHeight="842"
              columnWidth="555" leftMargin="20" rightMargin="20" topMargin="20" bottomMargin="20"
              uuid="e8f1c2d3-b4a5-6789-0123-abcdef456789">

    <parameter name="TASK_ID" class="java.lang.String"/>
    <parameter name="EXECUTION_CLUSTER" class="java.lang.String"/>

    <field name="transactionRef" class="java.lang.String"/>
    <field name="clearingMember" class="java.lang.String"/>
    <field name="settlementAmount" class="java.math.BigDecimal"/>
    <field name="executionStatus" class="java.lang.String"/>

    <title height="55">
        <element kind="staticText" x="0" y="5" width="555" height="25">
            <textElement textAlignment="Center" verticalAlignment="Middle">
                <font fontName="Sarabun" size="16" isBold="true"/>
            </textElement>
            <text><![CDATA[รายงานสรุปการชำระราคาแบบกระจายศูนย์ (Omnibus Settlement Run)]]></text>
        </element>
        <element kind="textField" x="0" y="30" width="555" height="20">
            <textElement textAlignment="Center">
                <font fontName="Sarabun" size="10"/>
            </textElement>
            <expression><![CDATA["Job Ticket: " + $P{TASK_ID} + " | Worker Pod Cluster: " + $P{EXECUTION_CLUSTER}]]></expression>
        </element>
    </title>

    <columnHeader height="20">
        <element kind="staticText" x="0" y="0" width="130" height="20">
            <textElement><font fontName="Sarabun" isBold="true"/></textElement>
            <text><![CDATA[TRANSACTION REF]]></text>
        </element>
        <element kind="staticText" x="130" y="0" width="190" height="20">
            <textElement><font fontName="Sarabun" isBold="true"/></textElement>
            <text><![CDATA[CLEARING MEMBER]]></text>
        </element>
        <element kind="staticText" x="320" y="0" width="135" height="20">
            <textElement textAlignment="Right"><font fontName="Sarabun" isBold="true"/></textElement>
            <text><![CDATA[SETTLED VALUE (USD)]]></text>
        </element>
        <element kind="staticText" x="455" y="0" width="100" height="20">
            <textElement textAlignment="Center"><font fontName="Sarabun" isBold="true"/></textElement>
            <text><![CDATA[STATUS]]></text>
        </element>
    </columnHeader>

    <detail>
        <band height="20">
            <element kind="textField" x="0" y="0" width="130" height="20">
                <expression><![CDATA[$F{transactionRef}]]></expression>
            </element>
            <element kind="textField" x="130" y="0" width="190" height="20">
                <expression><![CDATA[$F{clearingMember}]]></expression>
            </element>
            <element kind="textField" x="320" y="0" width="135" height="20" pattern="#,##0.00">
                <textElement textAlignment="Right"/>
                <expression><![CDATA[$F{settlementAmount}]]></expression>
            </element>
            <element kind="textField" x="455" y="0" width="100" height="20">
                <textElement textAlignment="Center"/>
                <expression><![CDATA[$F{executionStatus}]]></expression>
            </element>
        </band>
    </detail>

    <pageFooter height="20">
        <element kind="textField" x="0" y="0" width="555" height="20">
            <textElement textAlignment="Center"><font size="8"/></textElement>
            <expression><![CDATA["หน้า " + $V{PAGE_NUMBER} + " (จัดเก็บลง AWS S3 สำเร็จ)"]]></expression>
        </element>
    </pageFooter>
</jasperReport>