| Line 46... |
Line 46... |
| 46 |
private OfferProcessingHelper offerProcessingHelper;
|
46 |
private OfferProcessingHelper offerProcessingHelper;
|
| 47 |
|
47 |
|
| 48 |
@Autowired
|
48 |
@Autowired
|
| 49 |
private CronBatchService cronBatchService;
|
49 |
private CronBatchService cronBatchService;
|
| 50 |
|
50 |
|
| - |
|
51 |
public boolean hasUnfinishedBatch(int offerId) {
|
| - |
|
52 |
return cronBatchService.findRunningForOffer(offerId) != null;
|
| - |
|
53 |
}
|
| - |
|
54 |
|
| 51 |
/**
|
55 |
/**
|
| 52 |
* Fire-and-forget: schedule the batch on a background worker and return immediately.
|
56 |
* Fire-and-forget: schedule the batch on a background worker and return immediately.
|
| 53 |
* Prevents duplicate concurrent runs for the same offerId at the JVM level.
|
57 |
* Prevents duplicate concurrent runs for the same offerId at the JVM level.
|
| 54 |
* Used by HTTP controllers; cron CLI keeps using the sync processOfferWithBatch.
|
58 |
* Used by HTTP controllers; cron CLI keeps using the sync processOfferWithBatch.
|
| 55 |
*/
|
59 |
*/
|
| 56 |
public String submitBatchAsync(int offerId) {
|
60 |
public String submitBatchAsync(int offerId) {
|
| 57 |
if (!inFlightOfferIds.add(offerId)) {
|
61 |
if (!inFlightOfferIds.add(offerId)) {
|
| 58 |
LOGGER.info("Offer {} batch submit ignored — already in flight", offerId);
|
62 |
LOGGER.info("Offer {} batch submit ignored — already in flight", offerId);
|
| 59 |
return "Offer " + offerId + " is already being processed. Check the batch summary for progress.";
|
63 |
return "Offer " + offerId + " is already being processed. Check the batch summary for progress.";
|
| 60 |
}
|
64 |
}
|
| - |
|
65 |
CronBatch existing = cronBatchService.findRunningForOffer(offerId);
|
| - |
|
66 |
if (existing != null) {
|
| - |
|
67 |
inFlightOfferIds.remove(offerId);
|
| - |
|
68 |
LOGGER.info("Offer {} has unfinished batch {} (started {}); blocking reprocess",
|
| - |
|
69 |
offerId, existing.getId(), existing.getStartedAt());
|
| - |
|
70 |
return "Offer " + offerId + " has an unfinished batch. Reprocessing is not allowed until the existing batch is fully processed.";
|
| - |
|
71 |
}
|
| 61 |
BATCH_EXECUTOR.submit(() -> {
|
72 |
BATCH_EXECUTOR.submit(() -> {
|
| 62 |
try {
|
73 |
try {
|
| 63 |
processOfferWithBatch(offerId);
|
74 |
processOfferWithBatch(offerId);
|
| 64 |
} catch (Exception e) {
|
75 |
} catch (Exception e) {
|
| 65 |
LOGGER.error("Offer {} batch processing failed: {}", offerId, e.getMessage(), e);
|
76 |
LOGGER.error("Offer {} batch processing failed: {}", offerId, e.getMessage(), e);
|