Дело создается дважды путем двойного вызова API в Spring/Java [закрыто] ⇐ JAVA

[Anonymous]

У нас есть такой код, который трижды повторяет вызов API в рамках механизма опроса. Если он не завершится, он создаст дело, вызвав другой API. Происходит следующее: в журналах я вижу два потока — основной и пул потоков, создающие новые дела. Таким образом, идентификатор отслеживания из пула потоков thead принимает значение null и, таким образом, создает дублирующиеся потоки.
Вот соответствующая часть кода. LLM предлагает мне добавить тайм-аут ниже вызова отмены, однако с этим сталкивается другой API, в котором нет тайм-аута.
Основной поток
if (entity != null && entity.getId() != null) {
Future future = null;
try {
log.info("Waiting for polling result for id : {} trackingId: {}", entity.getId(), trackingId);
OrderRequest finalEntity = entity;
String finalProductOrderId = productOrderId;
Callable task = () -> checkGetStatusAPIForCeaseLine(customerId, finalProductOrderId, routeEnv, sc, trackingId, finalEntity, req);
future = executor.submit(task);
CeaseLineResponse result = future.get(futureTimeoutSecondsForCeaseLine, TimeUnit.SECONDS);
return result;
} catch (TimeoutException te) {
log.warn("Polling did not complete in time for id: {} trackingId: {}", entity.getId(), trackingId);
future.cancel(true);
return handleAmaCaseForCeaseLineTimeout(customerId, entity, productOrderId, req, httpHeaders, trackingId);
} catch (Exception e) {
log.error("Error while waiting for poll result", e);
return CeaseLineResponse.builder()
.httpCode(500)
.message("FAILED")
.orderNo(entity.getBillerOrderNo())
.orderActionId(entity.getBillerOrderActionNo())
.build();
}
}

Вызов исполнителя
@Transactional
public CeaseLineResponse checkGetStatusAPIForCeaseLine(String customerId, String productOrderId, String routeEnv, String sc, String trackingId, OrderRequest entity, CreateAndSubmitSuspendResumeCeaseOrderRequest req) {

HttpHeaders httpHeaders = new HttpHeaders();
httpHeaders.add(HEADER_TRACKING_ID, trackingId);
httpHeaders.add(HEADER_ROUTE_ENV, routeEnv);
httpHeaders.add(HEADER_SOURCE_SYSTEM_ID, "MATRIX-CEASE-POLL");
httpHeaders.add(HEADER_SOURCE_SERVER_ID, "MATRIX-CEASE-POLL");
httpHeaders.add(HEADER_SERVICE_NAME, "OrderStatus");
httpHeaders.add(HEADER_CACHE_CONTROL, HEADER_CACHE_CONTROL_NO_CACHE);
httpHeaders.add(HEADER_SERVICE_OPERATION_NAME, "MATRIX-CEASE-POLL");
httpHeaders.add(HEADER_TIMESTAMP, String.valueOf(System.currentTimeMillis()));

for (int i = 0; i < maxAttemptForCeaseLine; i++) {
// wait before polling
try {
Thread.sleep(ceaseLineSubmittedPlan * 1000);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
log.info("Polling interrupted due to timeout, trackingId: {}", trackingId);
return null;
}

try {
long start = System.nanoTime();
OrderStatusResponse orderStatus = amdocs.getOrderStatus(customerId, productOrderId, sc, trackingId, httpHeaders);
long end = System.nanoTime();
double durationSeconds = (end - start) / 1_000_000_000.0;
log.info("Polling attempt {} for getCeaseLineOrderStatus. Execution time: {} seconds, trackingId: {}", i + 1, durationSeconds, trackingId);
String status = orderStatus != null && orderStatus.getMessage() != null ? orderStatus.getMessage().getStatus() : null;
log.info("Polled status: {} for cease line, trackingId: {}", status, trackingId);

if ("CLOSED".equalsIgnoreCase(status)) {
entity.setStatus(OrderStatus.SUCCESS);
entity.setStatusUpdateDate(Instant.now());
entity.setStatusUpdateSource("API");
entity.setAttemptCount(i + 1);
final OrderRequest save = ceaseOrderRepo.save(entity);
log.info("Obtained CLOSED status for cease line. Updated row: {} trackingId: {}", save, entity.getTrackingId());

return CeaseLineResponse.builder()
.httpCode(200)
.message("SUCCESS")
.orderNo(entity.getBillerOrderNo())
.orderActionId(entity.getBillerOrderActionNo())
.build();
}
} catch (Exception ex) {
entity.setStatus(OrderStatus.FAILED);
entity.setRejectReason("Poll error: " + checkError(ex));
entity.setAttemptCount(i + 1);
final OrderRequest save = ceaseOrderRepo.save(entity);
log.info("Obtained FAILED status for cease line. Updated row: {} trackingId: {}", save, entity.getTrackingId());

return CeaseLineResponse.builder()
.httpCode(500)
.message("FAILED")
.orderNo(entity.getBillerOrderNo())
.orderActionId(entity.getBillerOrderActionNo())
.messageDescription(entity.getRejectReason())
.build();
}
}

entity.setAttemptCount(maxAttemptForReplaceDevice);
return handleAmaCaseForCeaseLineTimeout(customerId, entity, productOrderId, req, httpHeaders, trackingId);
}

private CeaseLineResponse handleAmaCaseForCeaseLineTimeout(
String customerId,
OrderRequest entity,
String productOfferId,
CreateAndSubmitSuspendResumeCeaseOrderRequest req,
HttpHeaders httpHeaders,
String trackingId
) {
log.info("Creating case in AMA for stuck cease order. Order ID: {}, Action ID: {}, trackingId: {}", entity.getBillerOrderNo(), entity.getBillerOrderActionNo(),
trackingId);
String bpmOrderId = req.getPayload().getBpmOrderInputX9() != null
? req.getPayload().getBpmOrderInputX9().getBpmOrderIdx9()
: null;
String caseNumber = maybeCreateAmaCase(customerId, entity, productOfferId, bpmOrderId, httpHeaders);

entity.setRejectReason("Order created but still processing - case Number: " + caseNumber);
entity.setStatus(OrderStatus.SUBMITTED);
final OrderRequest save = ceaseOrderRepo.save(entity);
log.info("SUBMITTED (polling finished) for cease line. Updated row: {} trackingId: {}", save, entity.getTrackingId());

return buildAcceptedOrderResponse202(entity.getBillerOrderNo(), entity.getBillerOrderActionNo(), "Order created but still processing", caseNumber);
}

Это то, что происходит в запросе и создании нового дела с использованием API
u_correlation_id\":\"null:9001962483:9001962484
Вот что LLM предложила мне добавить после вызова отмены:
// Add a small delay to ensure the polling thread has stopped
try {
Thread.sleep(100);
} catch (InterruptedException ie) {
Thread.currentThread().interrupt();
}

и прерванная проверка здесь"
for (int i = 0; i < maxAttemptForCeaseLine; i++) {
// Check if thread has been interrupted/cancelled
if (Thread.currentThread().isInterrupted()) {
log.info("Polling cancelled for trackingId: {}", trackingId);
return null;
}

Что здесь не так? Есть мысли?
Изменить: такое поведение наблюдается только при попыткеCount=3, а не для 0,1 и 2.
Изменить 2: добавление минимального примера. Однако это не может воспроизвести состояние гонки
package com.order.service.concurrent;

import org.junit.After;
import org.junit.Before;
import org.junit.Test;
import org.junit.jupiter.api.BeforeEach;
import org.junit.runner.RunWith;
import org.mockito.Mock;
import org.mockito.MockitoAnnotations;
import org.mockito.junit.MockitoJUnitRunner;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;

import static org.junit.Assert.assertEquals;
import static org.mockito.ArgumentMatchers.*;
import static org.mockito.Mockito.*;

@RunWith(MockitoJUnitRunner.class)
public class DoubleCaseCreationTest {
private static final Logger log = LoggerFactory.getLogger(DoubleCaseCreationTest.class);

@Mock
private Service Service;

@Mock
private CeaseOrderRepository ceaseOrderRepo;

private ExecutorService executor;
private AtomicInteger amaCaseCallCount;
private CeaseOrderService ceaseOrderService;

// Test configuration
private static final int TIMEOUT_SECONDS = 1;
private static final int POLLING_ATTEMPTS = 3;
private static final int[] POLLING_INTERVALS = {1,1,1}; // 1 second each = 5 seconds total

@Before
public void setUp() {
MockitoAnnotations.initMocks(this);
executor = Executors.newFixedThreadPool(5);
amaCaseCallCount = new AtomicInteger(0);

// Mock repository save
when(ceaseOrderRepo.save(any(com..matrix.order.service.entity.OrderRequest.class)))
.thenAnswer(invocation -> invocation.getArgument(0));

// Mock to return IN_PROGRESS (never completes)
OrderStatusResponse inProgressResponse = new OrderStatusResponse();
Message message = new Message();
message.setStatus("IN_PROGRESS");
inProgressResponse.setMessage(message);

when(Service.getOrderStatus(anyString(), anyString(), anyString(), anyString(), any()))
.thenReturn(inProgressResponse);

ceaseOrderService = new CeaseOrderService(
Service,
ceaseOrderRepo,
executor,
TIMEOUT_SECONDS,
POLLING_ATTEMPTS,
POLLING_INTERVALS
);
}

@After
public void tearDown() {
executor.shutdownNow();
try {
executor.awaitTermination(2, TimeUnit.SECONDS);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
}

/**
* TEST 1: Demonstrates the BUG - Two AMA cases are created
*
* Timeline:
* T=0s: Main thread submits polling task
* T=0-2s: Executor thread polls (attempts 1-2)
* T=2s: Main thread times out
* T=2s: Main thread creates AMA case #1
* T=2-5s: Executor thread continues polling (attempts 3-5)
* T=5s: Executor thread exhausts retries, creates AMA case #2
*
* Result: 2 cases created
*/
@Test
public void testBuggyVersion_CreatesTwoCases() throws Exception {
log.info("\n========================================");
log.info("TEST: BUGGY VERSION - Expects 2 Cases");
log.info("========================================\n");

// Reset counter
amaCaseCallCount.set(0);

// Track AMA case creation calls
AtomicInteger mainThreadCalls = new AtomicInteger(0);
AtomicInteger executorThreadCalls = new AtomicInteger(0);

ceaseOrderService.setAmaCaseCreationCallback((threadName) -> {
int count = amaCaseCallCount.incrementAndGet();
log.warn("AMA CASE #{} CREATED by thread: {}", count, threadName);

if (threadName.contains("main") || threadName.contains("Test worker")) {
mainThreadCalls.incrementAndGet();
} else if (threadName.contains("pool")) {
executorThreadCalls.incrementAndGet();
}
});

// Create test data
com..matrix.order.service.entity.OrderRequest entity = createTestOrder();
CreateAndSubmitSuspendResumeCeaseOrderRequest request = new CreateAndSubmitSuspendResumeCeaseOrderRequest();

// Execute BUGGY version
long startTime = System.currentTimeMillis();
CeaseLineResponse response = ceaseOrderService.createCeaseOrderBuggy(
"customer123",
entity,
"order123",
request,
"demo",
"sc123",
"track123"
);
long endTime = System.currentTimeMillis();

log.info("\n--- BUGGY VERSION Response ---");
log.info("Execution time: {}ms", (endTime - startTime));
log.info("Response code: {}", response.getHttpCode());
log.info("Response message: {}", response.getMessage());

// Wait for executor thread to finish
log.info("\nWaiting for executor thread to complete...");
//Thread.sleep(4000); // Wait for remaining polling attempts

// Results
int totalCases = amaCaseCallCount.get();
log.info("\n========================================");
log.info("BUGGY VERSION RESULTS:");
log.info("Total AMA cases created: {}", totalCases);
log.info("Cases by main thread: {}", mainThreadCalls.get());
log.info("Cases by executor thread: {}", executorThreadCalls.get());
log.info("========================================\n");

// Assertions - BUGGY version creates 2 cases
assertEquals("BUGGY: Should create 2 cases (demonstrating the bug)", 2, totalCases);
assertEquals("Main thread should create 1 case", 1, mainThreadCalls.get());
assertEquals("Executor thread should create 1 case", 1, executorThreadCalls.get());
assertEquals("Should return 202 ACCEPTED", 202, response.getHttpCode());
}

private com..matrix.order.service.entity.OrderRequest createTestOrder() {
com..matrix.order.service.entity.OrderRequest entity = new com..matrix.order.service.entity.OrderRequest();
entity.setId(1L);
entity.setBillerOrderNo("9001962483");
entity.setBillerOrderActionNo("9001962484");
entity.setTrackingId("testTrack123");
entity.setStatus(com..matrix.order.service.domain.OrderStatus.REJECTED);
return entity;
}
}

/**
* Service class that implements both BUGGY and FIXED versions
*/
class CeaseOrderService {
private static final Logger log = LoggerFactory.getLogger(CeaseOrderService.class);

private final Service Service;
private final CeaseOrderRepository ceaseOrderRepo;
private final ExecutorService executor;
private final int timeoutSeconds;
private final int maxPollingAttempts;
private final int[] pollingIntervals;

// Callback for testing
private java.util.function.Consumer amaCaseCreationCallback;

public CeaseOrderService(Service Service,
CeaseOrderRepository ceaseOrderRepo,
ExecutorService executor,
int timeoutSeconds,
int maxPollingAttempts,
int[] pollingIntervals) {
this.Service = Service;
this.ceaseOrderRepo = ceaseOrderRepo;
this.executor = executor;
this.timeoutSeconds = timeoutSeconds;
this.maxPollingAttempts = maxPollingAttempts;
this.pollingIntervals = pollingIntervals;
}

public void setAmaCaseCreationCallback(java.util.function.Consumer callback) {
this.amaCaseCreationCallback = callback;
}

/**
* BUGGY VERSION - Does not cancel future properly
*/
public CeaseLineResponse createCeaseOrderBuggy(
String customerId,
com..matrix.order.service.entity.OrderRequest entity,
String productOrderId,
CreateAndSubmitSuspendResumeCeaseOrderRequest req,
String routeEnv,
String sc,
String trackingId) {

if (entity != null && entity.getId() != null) {
Future future = null;
try {
log.info("[BUGGY] Submitting polling task - trackingId: {}", trackingId);

Callable task = new Callable() {
@Override
public CeaseLineResponse call() throws Exception {
return checkGetStatusAPIBuggy(customerId, productOrderId, routeEnv, sc, trackingId, entity, req);
}
};

future = executor.submit(task);
CeaseLineResponse result = future.get(timeoutSeconds, TimeUnit.SECONDS);
return result;

} catch (TimeoutException te) {
log.warn("[BUGGY] Timeout occurred! Main thread creating case - trackingId: {}", trackingId);
future.cancel(true);
// BUG: No future cancellation!
// Executor thread continues running and will create its own case

return handleAmaCaseForCeaseLineTimeout(customerId, entity, productOrderId, trackingId);

} catch (Exception e) {
log.error("[BUGGY] Error: {}", e.getMessage());
return buildFailedResponse(entity);
}
}
return null;
}

/**
* FIXED VERSION - Properly cancels future
*/
public CeaseLineResponse createCeaseOrderFixed(
String customerId,
com..matrix.order.service.entity.OrderRequest entity,
String productOrderId,
CreateAndSubmitSuspendResumeCeaseOrderRequest req,
String routeEnv,
String sc,
String trackingId) {

if (entity != null && entity.getId() != null) {
Future future = null;
try {
log.info("[FIXED] Submitting polling task - trackingId: {}", trackingId);

Callable task = new Callable() {
@Override
public CeaseLineResponse call() throws Exception {
return checkGetStatusAPIFixed(customerId, productOrderId, routeEnv, sc, trackingId, entity, req);
}
};

future = executor.submit(task);
CeaseLineResponse result = future.get(timeoutSeconds, TimeUnit.SECONDS);
return result;

} catch (TimeoutException te) {
log.warn("[FIXED] Timeout occurred! Cancelling future - trackingId: {}", trackingId);

// FIX: Cancel the future
if (future != null) {
boolean cancelled = future.cancel(true);
log.info("[FIXED] Future cancelled: {}", cancelled);
}

// FIX: Wait for task to stop
if (future != null) {
try {
future.get(200, TimeUnit.MILLISECONDS);
} catch (CancellationException ce) {
log.info("[FIXED] Task was successfully cancelled");
} catch (TimeoutException | ExecutionException | InterruptedException e) {
log.warn("[FIXED] Task did not stop in time: {}", e.getMessage());
}
}

return handleAmaCaseForCeaseLineTimeout(customerId, entity, productOrderId, trackingId);

} catch (Exception e) {
log.error("[FIXED] Error: {}", e.getMessage());
return buildFailedResponse(entity);
}
}
return null;
}

/**
* BUGGY polling - doesn't respect interruption
*/
private CeaseLineResponse checkGetStatusAPIBuggy(
String customerId,
String productOrderId,
String routeEnv,
String sc,
String trackingId,
com..matrix.order.service.entity.OrderRequest entity,
CreateAndSubmitSuspendResumeCeaseOrderRequest req) {

log.info("[BUGGY POLL] Starting polling in executor thread - trackingId: {}", trackingId);

for (int i = 0; i < maxPollingAttempts; i++) {
try {
log.info("[BUGGY POLL] Attempt {}/{} - sleeping {}s", i+1, maxPollingAttempts, pollingIntervals);
Thread.sleep(pollingIntervals * 1000);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
// BUG: Doesn't return, continues execution!
log.warn("[BUGGY POLL] Interrupted but continuing...");
return null;

}

try {
Thread.sleep(7000L);
OrderStatusResponse orderStatus = Service.getOrderStatus(
customerId, productOrderId, sc, trackingId, null);
String status = orderStatus != null && orderStatus.getMessage() != null
? orderStatus.getMessage().getStatus() : null;

if ("CLOSED".equalsIgnoreCase(status)) {
entity.setStatus(com..matrix.order.service.domain.OrderStatus.SUCCESS);
ceaseOrderRepo.save(entity);
log.info("[BUGGY POLL] Order completed successfully");
return buildSuccessResponse(entity);
}
} catch (Exception ex) {
log.error("[BUGGY POLL] Error polling: {}", ex.getMessage());
}
}

log.warn("[BUGGY POLL] All attempts exhausted, executor thread creating case!");
return handleAmaCaseForCeaseLineTimeout(customerId, entity, productOrderId, trackingId);
}

/**
* FIXED polling - respects interruption
*/
private CeaseLineResponse checkGetStatusAPIFixed(
String customerId,
String productOrderId,
String routeEnv,
String sc,
String trackingId,
com..matrix.order.service.entity.OrderRequest entity,
CreateAndSubmitSuspendResumeCeaseOrderRequest req) {

log.info("[FIXED POLL] Starting polling in executor thread - trackingId: {}", trackingId);

for (int i = 0; i < maxPollingAttempts; i++) {
// FIX: Check for interruption at start of loop
if (Thread.currentThread().isInterrupted()) {
log.info("[FIXED POLL] Detected interruption at loop start, exiting cleanly");
return null;
}

try {
log.info("[FIXED POLL] Attempt {}/{} - sleeping {}s", i+1, maxPollingAttempts, pollingIntervals);
Thread.sleep(pollingIntervals * 1000L);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
// FIX: Return immediately on interruption
log.info("[FIXED POLL] Interrupted during sleep, exiting immediately");
return null;
}

try {
OrderStatusResponse orderStatus = Service.getOrderStatus(
customerId, productOrderId, sc, trackingId, null);
String status = orderStatus != null && orderStatus.getMessage() != null
? orderStatus.getMessage().getStatus() : null;

if ("CLOSED".equalsIgnoreCase(status)) {
entity.setStatus(com..matrix.order.service.domain.OrderStatus.SUCCESS);
ceaseOrderRepo.save(entity);
log.info("[FIXED POLL] Order completed successfully");
return buildSuccessResponse(entity);
}
} catch (Exception ex) {
log.error("[FIXED POLL] Error polling: {}", ex.getMessage());
}
}

log.warn("[FIXED POLL] All attempts exhausted, creating case");
return handleAmaCaseForCeaseLineTimeout(customerId, entity, productOrderId, trackingId);
}

private CeaseLineResponse handleAmaCaseForCeaseLineTimeout(
String customerId,
com..matrix.order.service.entity.OrderRequest entity,
String productOrderId,
String trackingId) {

String threadName = Thread.currentThread().getName();
log.info(" Creating AMA case from thread: {}", threadName);

// Notify callback for testing
if (amaCaseCreationCallback != null) {
amaCaseCreationCallback.accept(threadName);
}

String caseNumber = "CASE-" + System.currentTimeMillis();

entity.setRejectReason("Order created but still processing - case: " + caseNumber);
entity.setStatus(com..matrix.order.service.domain.OrderStatus.SUBMITTED);
ceaseOrderRepo.save(entity);

return CeaseLineResponse.builder()
.httpCode(202)
.message("ACCEPTED")
.messageDescription("Case created: " + caseNumber)
.orderNo(entity.getBillerOrderNo())
.orderActionId(entity.getBillerOrderActionNo())
.build();
}

private CeaseLineResponse buildSuccessResponse(com..matrix.order.service.entity.OrderRequest entity) {
return CeaseLineResponse.builder()
.httpCode(200)
.message("SUCCESS")
.orderNo(entity.getBillerOrderNo())
.orderActionId(entity.getBillerOrderActionNo())
.build();
}

private CeaseLineResponse buildFailedResponse(com..matrix.order.service.entity.OrderRequest entity) {
return CeaseLineResponse.builder()
.httpCode(500)
.message("FAILED")
.orderNo(entity.getBillerOrderNo())
.orderActionId(entity.getBillerOrderActionNo())
.build();
}
}

// Supporting classes and mocks

interface Service {
OrderStatusResponse getOrderStatus(String customerId, String productOrderId,
String sc, String trackingId, Object headers);
}

interface CeaseOrderRepository {
com..matrix.order.service.entity.OrderRequest save(com..matrix.order.service.entity.OrderRequest entity);
}

class OrderRequest {
private Long id;
private String billerOrderNo;
private String billerOrderActionNo;
private String trackingId;
private OrderStatus status;
private String rejectReason;

// Getters and setters
public Long getId() { return id; }
public void setId(Long id) { this.id = id; }
public String getBillerOrderNo() { return billerOrderNo; }
public void setBillerOrderNo(String no) { this.billerOrderNo = no; }
public String getBillerOrderActionNo() { return billerOrderActionNo; }
public void setBillerOrderActionNo(String no) { this.billerOrderActionNo = no; }
public String getTrackingId() { return trackingId; }
public void setTrackingId(String id) { this.trackingId = id; }
public OrderStatus getStatus() { return status; }
public void setStatus(OrderStatus status) { this.status = status; }
public String getRejectReason() { return rejectReason; }
public void setRejectReason(String reason) { this.rejectReason = reason; }
}

enum OrderStatus { NEW, SUBMITTED, COMPLETED, REJECTED, FAILED, SUCCESS }

class OrderStatusResponse {
private Message message;
public Message getMessage() { return message; }
public void setMessage(Message message) { this.message = message; }
}

class Message {
private String status;
public String getStatus() { return status; }
public void setStatus(String status) { this.status = status; }
}

class CeaseLineResponse {
private int httpCode;
private String message;
private String messageDescription;
private String orderNo;
private String orderActionId;

public static Builder builder() { return new Builder(); }

public int getHttpCode() { return httpCode; }
public String getMessage() { return message; }
public String getMessageDescription() { return messageDescription; }
public String getOrderNo() { return orderNo; }
public String getOrderActionId() { return orderActionId; }

static class Builder {
private CeaseLineResponse response = new CeaseLineResponse();
public Builder httpCode(int code) { response.httpCode = code; return this; }
public Builder message(String msg) { response.message = msg; return this; }
public Builder messageDescription(String desc) { response.messageDescription = desc; return this; }
public Builder orderNo(String no) { response.orderNo = no; return this; }
public Builder orderActionId(String id) { response.orderActionId = id; return this; }
public CeaseLineResponse build() { return response; }
}
}

class CreateAndSubmitSuspendResumeCeaseOrderRequest {}



Подробнее здесь: https://stackoverflow.com/questions/798 ... pring-java