java-project/baiyue-firstleg
4
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

优化

Browse Source
This commit is contained in:
ouyangxq
2026-09-29 18:06:18 +08:00
parent 54ad26c607
commit 2c60a29e0f
30 changed files with 384 additions and 139 deletions
Download Patch File Download Diff File Expand all files Collapse all files
common-core/AGENTS.md
+1
View File
@@ -14,6 +14,7 @@
| 业务异常 | `src/main/java/com/baiyue/common/core/exception/BizException.java` |
| 通用审计字段 | `src/main/java/com/baiyue/common/core/model/BaseEntity.java` |
| MyBatis-Plus 与审计人 | `src/main/java/com/baiyue/common/core/mybatis/` |
| 有限重试与退避 | `src/main/java/com/baiyue/common/core/retry/BoundedRetryPolicy.java`；供多个服务的 Outbox 复用，不拥有持久化或投递逻辑 |
| 响应契约测试 | `src/test/java/com/baiyue/common/core/ApiResponseTest.java` |
## 修改与验证
common-event-contract/src/main/java/com/baiyue/common/event/EventDependencyNotReadyException.java
+7
View File
@@ -0,0 +1,7 @@
package com.baiyue.common.event;
/** Required business state has not arrived yet; retry the same event with backoff. */
public class EventDependencyNotReadyException extends IllegalArgumentException {
public EventDependencyNotReadyException(String message) { super(message); }
public EventDependencyNotReadyException(String message, Throwable cause) { super(message, cause); }
}
common-event-contract/src/main/java/com/baiyue/common/event/EventEnvelopeValidator.java
+1 -1
View File
@@ -75,7 +75,7 @@ public final class EventEnvelopeValidator {
public static void requireValid(EventEnvelope envelope) {
List<String> errors = validate(envelope);
if (!errors.isEmpty()) {
throw new IllegalArgumentException(String.join("；", errors));
throw new EventProtocolException(String.join("；", errors));
}
}
common-event-contract/src/main/java/com/baiyue/common/event/EventProtocolException.java
+7
View File
@@ -0,0 +1,7 @@
package com.baiyue.common.event;
/** An immutable event violates the consumption protocol; retrying unchanged data cannot fix it. */
public class EventProtocolException extends IllegalArgumentException {
public EventProtocolException(String message) { super(message); }
public EventProtocolException(String message, Throwable cause) { super(message, cause); }
}
common-web/AGENTS.md
+1
View File
@@ -13,6 +13,7 @@
| 请求 ID 过滤器 | `src/main/java/com/baiyue/common/web/CorrelationIdFilter.java` |
| W3C Trace Context | `src/main/java/com/baiyue/common/web/TraceParent.java`；严格接收 version 00，响应 `traceparent`、`X-Trace-Id` 与 MDC `traceId` 使用同一 trace ID |
| Feign Trace 传播 | `FeignTraceRequestInterceptor.java`、`FeignTracePropagationAutoConfiguration.java`；仅 Feign 存在时装配，optional 依赖不向无 Feign 服务传递 |
| 可靠事件 HTTP 客户端配置 | `EventDeliveryFeignConfiguration.java`；由业务事件 Feign 客户端显式选择，在最终构建阶段固定连接/读取超时、禁止重定向和客户端内部重试，不作为全局自动配置 |
| RestClient Trace 传播 | `RestClientTraceInterceptor.java`、`RestClientTracePropagationAutoConfiguration.java`；Spring 管理的 Builder 自动装配，原始 Builder 可显式复用拦截器 |
| 自动配置入口 | `src/main/java/com/baiyue/common/web/CommonWebAutoConfiguration.java`；为只扫描自身包的服务注册异常处理器和 Correlation 过滤器 |
| 全局异常映射与业务失败指标 | `src/main/java/com/baiyue/common/web/GlobalExceptionHandler.java`；指标 `baiyue.api.business.failures` 按有限 `code` 标记 |
common-web/src/main/java/com/baiyue/common/web/EventDeliveryFailure.java
+55
View File
@@ -0,0 +1,55 @@
package com.baiyue.common.web;
import com.baiyue.common.core.api.ApiResponse;
import com.fasterxml.jackson.core.JsonProcessingException;
import feign.FeignException;
import java.util.Collections;
import java.util.IdentityHashMap;
import java.util.Set;
/** Explicit failure classification and bounded, credential-free delivery diagnostics. */
public class EventDeliveryFailure extends IllegalStateException {
private static final Set<String> PERMANENT_CODES = Set.of("A0401", "A0403", "EVT_PROTOCOL");
private final boolean permanent;
public EventDeliveryFailure(boolean permanent, String message) {
super(message);
this.permanent = permanent;
}
public static void requireSuccessful(ApiResponse<?> response) {
requireSuccessful(response, Set.of());
}
public static void requireSuccessful(ApiResponse<?> response, Set<String> additionalPermanentCodes) {
if (response == null) throw new EventDeliveryFailure(false, "Consumer returned an empty response");
if (!response.successful()) {
throw new EventDeliveryFailure(response.code() != null && (PERMANENT_CODES.contains(response.code()) || additionalPermanentCodes.contains(response.code())),
"Consumer rejected event: code=" + response.code() + ", message=" + response.message()
+ ", requestId=" + response.requestId() + ", traceId=" + response.traceId());
}
if (response.data() == null) throw new EventDeliveryFailure(false, "Consumer ACK is missing");
}
public static boolean isPermanent(Throwable failure) {
Set<Throwable> seen = Collections.newSetFromMap(new IdentityHashMap<>());
for (Throwable current = failure; current != null && seen.add(current); current = current.getCause()) {
if (current instanceof EventDeliveryFailure delivery && delivery.permanent) return true;
if (current instanceof FeignException http && (http.status() == 401 || http.status() == 403)) return true;
}
return false;
}
public static String diagnostic(Throwable failure, String internalToken) {
Throwable current = failure;
Set<Throwable> seen = Collections.newSetFromMap(new IdentityHashMap<>());
while (current.getCause() != null && seen.add(current)
&& ("NoFallbackAvailableException".equals(current.getClass().getSimpleName())
|| "No fallback available.".equals(current.getMessage()))) current = current.getCause();
String message;
if (current instanceof FeignException http) message = "HTTP " + http.status();
else if (current instanceof JsonProcessingException) message = "Invalid event JSON";
else message = current.getMessage() == null ? "unknown" : current.getMessage();
String text = current.getClass().getSimpleName() + ": " + message;
if (internalToken != null && !internalToken.isBlank()) text = text.replace(internalToken, "***");
text = text.replaceAll("(?i)Bearer\\s+[^\\s,;\\\"']+", "Bearer ***")
.replaceAll("(?i)((?:X-Internal-Service-Token|Authorization|access[_-]?token|refresh[_-]?token)\\s*[=:]\\s*)[^\\s,;]+", "$1***")
.replaceAll("[\\r\\n\\t]+", " ");
return text.length() > 500 ? text.substring(0, 500) : text;
}
}
common-web/src/test/java/com/baiyue/common/web/EventDeliveryFailureTest.java
+55
View File
@@ -0,0 +1,55 @@
package com.baiyue.common.web;
import com.baiyue.common.core.api.ApiResponse;
import com.fasterxml.jackson.databind.ObjectMapper;
import feign.FeignException;
import feign.Request;
import feign.Response;
import org.junit.jupiter.api.Test;
import java.nio.charset.StandardCharsets;
import java.time.Instant;
import java.util.Map;
import static org.assertj.core.api.Assertions.*;
class EventDeliveryFailureTest {
@Test void onlyExplicitProtocolOrAuthRejectionIsPermanent() {
for (String code : new String[]{"EVT_PROTOCOL", "A0401", "A0403"}) {
Throwable error = catchThrowable(() -> EventDeliveryFailure.requireSuccessful(response(code)));
assertThat(EventDeliveryFailure.isPermanent(error)).as(code).isTrue();
}
for (String code : new String[]{"EVT_DEPENDENCY", "A0400", "A0409", "A0429", "S0503", "unknown", null}) {
Throwable error = catchThrowable(() -> EventDeliveryFailure.requireSuccessful(response(code)));
assertThat(error).isInstanceOf(EventDeliveryFailure.class);
assertThat(EventDeliveryFailure.isPermanent(error)).as(code).isFalse();
}
assertThat(EventDeliveryFailure.isPermanent(new IllegalArgumentException("ACK mismatch"))).isFalse();
}
@Test void httpAuthenticationFailureIsPermanentBut429And503Retry() {
for (int status : new int[]{401, 403, 429, 503}) {
var http = http(status);
assertThat(EventDeliveryFailure.isPermanent(new RuntimeException("wrapper", http)))
.isEqualTo(status == 401 || status == 403);
assertThat(EventDeliveryFailure.diagnostic(http, "secret")).contains("HTTP " + status)
.doesNotContain("remote-body", "secret");
}
}
@Test void diagnosticsRemoveTokenHeadersSecretsAndLineBreaks() {
var failure = new IllegalStateException("secret-value\nAuthorization=Bearer abc.def.ghi; X-Internal-Service-Token=another-secret, access_token=third-secret");
String value = EventDeliveryFailure.diagnostic(failure, "secret-value");
assertThat(value).doesNotContain("secret-value", "abc.def.ghi", "another-secret", "third-secret", "\n");
assertThat(EventDeliveryFailure.diagnostic(new RuntimeException("x".repeat(1000)), "")).hasSize(500);
}
@Test void malformedJsonDiagnosticNeverCopiesPersistedPayload() {
Throwable failure = catchThrowable(() -> new ObjectMapper().readTree("private-payload-value"));
assertThat(EventDeliveryFailure.diagnostic(failure, ""))
.contains("Invalid event JSON").doesNotContain("private-payload-value");
}
private ApiResponse<Object> response(String code) {
return new ApiResponse<>(code, "consumer-message", null, "request-id", "trace-id", Instant.now());
}
private FeignException http(int status) {
var request = Request.create(Request.HttpMethod.POST, "http://consumer/events?token=secret", Map.of(), new byte[0], StandardCharsets.UTF_8, null);
return FeignException.errorStatus("receive", Response.builder().request(request).status(status).reason("failure")
.headers(Map.of()).body("remote-body-secret", StandardCharsets.UTF_8).build());
}
}
finance/src/main/java/com/baiyue/finance/service/BusinessFeeSourceEventService.java
+3
View File
@@ -40,6 +40,9 @@ public class BusinessFeeSourceEventService {
|| envelope.targetRegion() != RegionCode.CN) {
throw error("FIN-EVT-001", "跨区费用事件类型、版本或区域无效");
}
if (!EventPayloadHash.matches(EventPayloadCanonicalizer.canonicalize(envelope.payload()), envelope.payloadHash())) {
throw new FinanceEventException("FIN-EVT-002", "事件信封摘要与实际载荷不一致");
}
BusinessFeeSourceChangedV1 source = payload(envelope);
validate(envelope, source);
String sourceSystem = source.sourceSystem().trim().toUpperCase();
finance/src/main/java/com/baiyue/finance/service/WalletCommandEventService.java
+3
View File
@@ -112,6 +112,9 @@ public class WalletCommandEventService {
|| envelope.sourceRegion() != RegionCode.OVERSEAS || envelope.targetRegion() != RegionCode.CN) {
throw new FinanceEventException("FIN-EVT-001", "资金指令类型、版本或区域无效");
}
if (!EventPayloadHash.matches(EventPayloadCanonicalizer.canonicalize(envelope.payload()), envelope.payloadHash())) {
throw new FinanceEventException("FIN-EVT-002", "事件信封摘要与实际载荷不一致");
}
try {
WalletCommandRequestedV1 value = objectMapper.treeToValue(envelope.payload(), WalletCommandRequestedV1.class);
String operation = value.operation() == null ? "" : value.operation().trim().toUpperCase(Locale.ROOT);
finance/src/main/java/com/baiyue/finance/service/WalletResultOutboxRelay.java
+12 -6
View File
@@ -1,6 +1,8 @@
package com.baiyue.finance.service;
import com.baiyue.common.event.EventEnvelope;
import com.baiyue.common.web.EventDeliveryFailure;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.baiyue.common.core.retry.BoundedRetryPolicy;
import com.baiyue.finance.client.WalletResultLastlegClient;
import com.fasterxml.jackson.databind.ObjectMapper;
@@ -50,21 +52,25 @@ public class WalletResultOutboxRelay {
"自动投递次数已耗尽（含租约超时）", eventId, worker);
return;
}
boolean requestStarted = false;
try {
EventEnvelope envelope = objectMapper.readValue(String.valueOf(row.get("payload_json")), EventEnvelope.class).requireValid();
requestStarted = true;
var response = client.receive(internalToken, envelope);
if (response == null || !response.successful() || response.data() == null)
throw new IllegalStateException("Lastleg 未确认资金结果事件");
EventDeliveryFailure.requireSuccessful(response);
response.data().requireMatches(envelope);
jdbc.update("UPDATE fin_wallet_event_outbox SET status='PUBLISHED',published_at=NOW(3),locked_by=NULL,locked_at=NULL,last_error=NULL,updated_at=NOW(3) WHERE event_id=? AND status='SENDING' AND locked_by=?",
eventId, worker);
} catch (Exception exception) {
String error = EventDeliveryFailure.diagnostic(exception, internalToken);
boolean permanent = EventDeliveryFailure.isPermanent(exception)
|| (!requestStarted && (exception instanceof JsonProcessingException || exception instanceof IllegalArgumentException));
int retries = failures + 1;
String status = policy.exhausted(retries) ? "DEAD" : "PENDING";
LocalDateTime next = policy.exhausted(retries) ? LocalDateTime.now() : LocalDateTime.now().plus(policy.delayAfterFailure(retries));
String status = permanent || policy.exhausted(retries) ? "DEAD" : "PENDING";
LocalDateTime next = permanent || policy.exhausted(retries) ? LocalDateTime.now() : LocalDateTime.now().plus(policy.delayAfterFailure(retries));
jdbc.update("UPDATE fin_wallet_event_outbox SET status=?,retry_count=?,next_retry_at=?,locked_by=NULL,locked_at=NULL,last_error=?,updated_at=NOW(3) WHERE event_id=? AND status='SENDING' AND locked_by=?",
status, retries, next, safe(exception.getMessage()), eventId, worker);
log.warn("资金结果事件投递失败 eventId={}, reason={}", eventId, exception.getMessage());
status, retries, next, safe(error), eventId, worker);
log.warn("资金结果事件投递失败 eventId={}, reason={}", eventId, error);
}
}
finance/src/test/java/com/baiyue/finance/WalletCommandEventServiceTest.java
+12
View File
@@ -40,6 +40,18 @@ class WalletCommandEventServiceTest {
when(transactions.getTransaction(any())).thenAnswer(invocation -> mock(TransactionStatus.class));
}
@Test
void rejectsTamperedEnvelopeBeforeOpeningWalletTransaction() {
EventEnvelope valid = envelope("FREEZE", 1L, BigDecimal.TEN);
EventEnvelope tampered = new EventEnvelope(valid.eventId(), valid.eventType(), valid.eventVersion(),
valid.occurredAt(), valid.sourceRegion(), valid.targetRegion(), valid.tenantId(), valid.aggregateType(),
valid.aggregateId(), valid.aggregateVersion(), valid.handoverVersion(), valid.traceId(),
valid.correlationId(), valid.causationId(), valid.payload(), "sha256:" + "0".repeat(64));
assertThatThrownBy(() -> service.consume(tampered)).isInstanceOf(FinanceEventException.class)
.hasMessageContaining("摘要与实际载荷不一致");
verifyNoInteractions(jdbc, wallet);
}
@Test
void freezesWalletAndPublishesCompletedEvent() {
EventEnvelope envelope = envelope("FREEZE", 31L, new BigDecimal("88.50"));
firstleg/src/main/java/com/baiyue/firstleg/controller/EventConsumptionExceptionHandler.java
+21
View File
@@ -0,0 +1,21 @@
package com.baiyue.firstleg.controller;
import com.baiyue.common.core.api.ApiResponse;
import com.baiyue.common.event.EventProtocolException;
import com.baiyue.common.event.EventDependencyNotReadyException;
import org.slf4j.MDC;
import org.springframework.core.Ordered;
import org.springframework.core.annotation.Order;
import org.springframework.web.bind.annotation.ExceptionHandler;
import org.springframework.web.bind.annotation.RestControllerAdvice;
import java.time.Instant;
@RestControllerAdvice
@Order(Ordered.HIGHEST_PRECEDENCE)
public class EventConsumptionExceptionHandler {
@ExceptionHandler({EventProtocolException.class, EventDependencyNotReadyException.class})
public ApiResponse<Void> event(IllegalArgumentException failure) {
String code = failure instanceof EventDependencyNotReadyException ? "EVT_DEPENDENCY" : "EVT_PROTOCOL";
return new ApiResponse<>(code, failure.getMessage(), null, MDC.get("requestId"), MDC.get("traceId"), Instant.now());
}
}
firstleg/src/main/java/com/baiyue/firstleg/controller/InternalOverseasContainerEventController.java
+1 -1
View File
@@ -37,7 +37,7 @@ public class InternalOverseasContainerEventController {
case EventTypes.OVERSEAS_CONTAINER_UNLOADED -> consumer.consume(event);
case EventTypes.OVERSEAS_INBOUND_ORDER_DETAINED_CHANGED -> detainedConsumer.consume(event);
case EventTypes.OVERSEAS_DELIVERY_COMPLETED -> deliveryCompletedConsumer.consume(event);
default -> throw new IllegalArgumentException("不支持的 Firstleg 业务事件类型");
default -> throw new com.baiyue.common.event.EventProtocolException("不支持的 Firstleg 业务事件类型");
};
return ApiResponse.success(result);
}
firstleg/src/main/java/com/baiyue/firstleg/service/LoadingCompletedOutboxRelay.java
+9 -28
View File
@@ -2,6 +2,8 @@ package com.baiyue.firstleg.service;
import com.baiyue.common.core.api.ApiResponse;
import com.baiyue.common.event.EventEnvelope;
import com.baiyue.common.web.EventDeliveryFailure;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.baiyue.common.core.retry.BoundedRetryPolicy;
import com.baiyue.common.event.EventEnvelopeValidator;
import com.baiyue.firstleg.client.LastlegEventClient;
@@ -50,15 +52,19 @@ public class LoadingCompletedOutboxRelay {
repository.dead(row.getId(), workerId, failures, "自动投递次数已耗尽（含租约超时）");
continue;
}
boolean requestStarted = false;
try {
EventEnvelope event=objectMapper.readValue(row.getPayloadJson(),EventEnvelope.class);
EventEnvelopeValidator.requireValid(event);
requestStarted = true;
var response=client.receive(internalToken,event);
requireAcknowledgement(event, response);
repository.published(row.getId(),workerId,LocalDateTime.now());
} catch(Exception exception) {
String error=safeError(exception);
if (policy.exhausted(failures + 1)) repository.dead(row.getId(), workerId, failures + 1, error);
boolean permanent = EventDeliveryFailure.isPermanent(exception)
|| (!requestStarted && (exception instanceof JsonProcessingException || exception instanceof IllegalArgumentException));
String error = EventDeliveryFailure.diagnostic(exception, internalToken);
if (permanent || policy.exhausted(failures + 1)) repository.dead(row.getId(), workerId, failures + 1, error);
else repository.retry(row.getId(), workerId, LocalDateTime.now().plus(policy.delayAfterFailure(failures + 1)), error);
log.warn("头程业务事件投递失败: eventType={}, eventKey={}, reason={}",
row.getEventType(), row.getEventKey(), error);
@@ -68,33 +74,8 @@ public class LoadingCompletedOutboxRelay {
private void requireAcknowledgement(EventEnvelope event,
ApiResponse<com.baiyue.common.event.DeliveryAck> response) {
if(response==null) throw new IllegalStateException("Lastleg未返回响应");
if(!response.successful()) {
throw new IllegalStateException("Lastleg拒绝头程业务事件: code=" + response.code()
+ ", message=" + response.message() + ", requestId=" + response.requestId()
+ ", traceId=" + response.traceId());
}
if (response.data() == null) throw new IllegalStateException("Lastleg ACK 缺失");
EventDeliveryFailure.requireSuccessful(response);
response.data().requireMatches(event);
}
private String safeError(Exception exception) {
Throwable diagnostic=diagnosticCause(exception);
String message=diagnostic.getMessage();
String value=diagnostic.getClass().getSimpleName()
+ ((message==null || message.isBlank())?"":": " + message);
if(internalToken!=null && !internalToken.isBlank()) value=value.replace(internalToken,"***");
value=value.replaceAll("[\\r\\n\\t]+"," ");
return value.length()>500?value.substring(0,500):value;
}
private Throwable diagnosticCause(Throwable failure) {
Throwable current=failure;
while(current.getCause()!=null && current.getCause()!=current
&& ("NoFallbackAvailableException".equals(current.getClass().getSimpleName())
|| "No fallback available.".equals(current.getMessage()))) {
current=current.getCause();
}
return current;
}
}
firstleg/src/main/java/com/baiyue/firstleg/service/OverseasContainerEventConsumer.java
+14 -12
View File
@@ -1,6 +1,8 @@
package com.baiyue.firstleg.service;
import com.baiyue.common.event.EventEnvelope;
import com.baiyue.common.event.EventProtocolException;
import com.baiyue.common.event.EventDependencyNotReadyException;
import com.baiyue.common.event.DeliveryAck;
import com.baiyue.common.event.EventPayloadCanonicalizer;
import com.baiyue.common.event.EventPayloadHash;
@@ -33,23 +35,23 @@ public class OverseasContainerEventConsumer {
OverseasContainerUnloadedV1 payload = payload(event);
if (payload.sourceContainerBookingId() == null || payload.sourceContainerBookingId().isBlank()
|| !payload.sourceContainerBookingId().equals(event.aggregateId())) {
throw new IllegalArgumentException("事件 aggregateId 与 sourceContainerBookingId 不一致");
throw new EventProtocolException("事件 aggregateId 与 sourceContainerBookingId 不一致");
}
if (payload.containerNo() == null || payload.containerNo().isBlank()) {
throw new IllegalArgumentException("货柜号不能为空");
throw new EventProtocolException("货柜号不能为空");
}
if (payload.unloadingCompletedAt() == null || payload.unloadingPhotoUploadedAt() == null) {
throw new IllegalArgumentException("卸柜完成时间和卸柜照片上传时间不能为空");
throw new EventProtocolException("卸柜完成时间和卸柜照片上传时间不能为空");
}
if (payload.actualEmptyReportedAt() != null
&& payload.actualEmptyReportedAt().isBefore(payload.unloadingPhotoUploadedAt())) {
throw new IllegalArgumentException("实际报空时间不能早于卸柜照片上传时间");
throw new EventProtocolException("实际报空时间不能早于卸柜照片上传时间");
}
ContainerEventInboxEntity existing = inbox.selectById(event.eventId());
if (existing != null) {
if (!existing.getPayloadHash().equalsIgnoreCase(event.payloadHash())) {
throw new IllegalArgumentException("相同 eventId 对应的事件摘要冲突");
throw new EventProtocolException("相同 eventId 对应的事件摘要冲突");
}
if (!"PROCESSED".equals(existing.getStatus())) throw new IllegalStateException("事件尚未完成消费");
return DeliveryAck.of(event, DeliveryAck.Outcome.DUPLICATE);
@@ -79,20 +81,20 @@ public class OverseasContainerEventConsumer {
private void validateEnvelope(EventEnvelope event) {
event.requireValid();
if (!EventTypes.OVERSEAS_CONTAINER_UNLOADED.equals(event.eventType()) || event.eventVersion() != 1) {
throw new IllegalArgumentException("不支持的海外货柜事件");
throw new EventProtocolException("不支持的海外货柜事件");
}
if (event.sourceRegion() != RegionCode.OVERSEAS || event.targetRegion() != RegionCode.CN
|| !"CONTAINER_BOOKING".equals(event.aggregateType())) {
throw new IllegalArgumentException("海外货柜事件区域或聚合类型无效");
throw new EventProtocolException("海外货柜事件区域或聚合类型无效");
}
if (!EventPayloadHash.matches(EventPayloadCanonicalizer.canonicalize(event.payload()), event.payloadHash())) {
throw new IllegalArgumentException("payloadHash 与事件负载不一致");
throw new EventProtocolException("payloadHash 与事件负载不一致");
}
}
private OverseasContainerUnloadedV1 payload(EventEnvelope event) {
try { return objectMapper.treeToValue(event.payload(), OverseasContainerUnloadedV1.class); }
catch (Exception exception) { throw new IllegalArgumentException("海外货柜事件负载无效", exception); }
catch (Exception exception) { throw new EventProtocolException("海外货柜事件负载无效", exception); }
}
private ContainerEventInboxEntity inbox(EventEnvelope event) {
@@ -110,12 +112,12 @@ public class OverseasContainerEventConsumer {
try {
bookingId = Long.parseLong(sourceContainerBookingId);
} catch (NumberFormatException exception) {
throw new IllegalArgumentException("国内货柜订舱 ID 无效", exception);
throw new EventProtocolException("国内货柜订舱 ID 无效", exception);
}
ContainerBookingEntity value = containers.selectActiveBookingForUpdate(bookingId);
if (value == null) throw new IllegalArgumentException("头程货柜不存在或不可用");
if (value == null) throw new EventDependencyNotReadyException("头程货柜不存在或不可用");
if (value.getContainerNo() == null || !value.getContainerNo().equals(containerNo)) {
throw new IllegalArgumentException("事件货柜号与国内订舱记录不一致");
throw new EventProtocolException("事件货柜号与国内订舱记录不一致");
}
return value;
}
firstleg/src/main/java/com/baiyue/firstleg/service/OverseasDeliveryCompletedConsumer.java
+13 -10
View File
@@ -2,6 +2,8 @@ package com.baiyue.firstleg.service;
import com.baomidou.mybatisplus.core.toolkit.IdWorker;
import com.baiyue.common.event.EventEnvelope;
import com.baiyue.common.event.EventProtocolException;
import com.baiyue.common.event.EventDependencyNotReadyException;
import com.baiyue.common.event.DeliveryAck;
import com.baiyue.common.event.EventPayloadCanonicalizer;
import com.baiyue.common.event.EventPayloadHash;
@@ -45,12 +47,12 @@ public class OverseasDeliveryCompletedConsumer {
long inboundId = positive(payload.sourceInboundId(), "来源入仓单 ID 无效");
long tenantId = positive(event.tenantId(), "客户 ID 无效");
if (!Objects.equals(event.aggregateId(), payload.sourceInboundId())) {
throw new IllegalArgumentException("事件 aggregateId 与来源入仓单 ID 不一致");
throw new EventProtocolException("事件 aggregateId 与来源入仓单 ID 不一致");
}
ContainerEventInboxEntity existing = inbox.selectById(event.eventId());
if (existing != null) {
if (!existing.getPayloadHash().equalsIgnoreCase(event.payloadHash())) {
throw new IllegalArgumentException("相同 eventId 对应的事件摘要冲突");
throw new EventProtocolException("相同 eventId 对应的事件摘要冲突");
}
if (!"PROCESSED".equals(existing.getStatus())) throw new IllegalStateException("事件尚未完成消费");
return DeliveryAck.of(event, DeliveryAck.Outcome.DUPLICATE);
@@ -65,7 +67,7 @@ public class OverseasDeliveryCompletedConsumer {
}
InboundOrderEntity inbound = required(inboundId, tenantId, payload);
if (!"LOADED".equals(inbound.getProcessStatus()) && !"DELIVERED".equals(inbound.getProcessStatus())) {
throw new IllegalArgumentException("来源头程入仓单尚未装柜完成");
throw new EventDependencyNotReadyException("来源头程入仓单尚未装柜完成");
}
if ("LOADED".equals(inbound.getProcessStatus())) {
LocalDateTime completedAt = LocalDateTime.ofInstant(payload.completedAt(), ZoneOffset.UTC);
@@ -102,10 +104,10 @@ public class OverseasDeliveryCompletedConsumer {
if (!EventTypes.OVERSEAS_DELIVERY_COMPLETED.equals(event.eventType()) || event.eventVersion() != 1
|| event.sourceRegion() != RegionCode.OVERSEAS || event.targetRegion() != RegionCode.CN
|| !"INBOUND_ORDER".equals(event.aggregateType())) {
throw new IllegalArgumentException("不支持的海外派送完成事件");
throw new EventProtocolException("不支持的海外派送完成事件");
}
if (!EventPayloadHash.matches(EventPayloadCanonicalizer.canonicalize(event.payload()), event.payloadHash())) {
throw new IllegalArgumentException("payloadHash 与事件负载不一致");
throw new EventProtocolException("payloadHash 与事件负载不一致");
}
}
@@ -114,22 +116,23 @@ public class OverseasDeliveryCompletedConsumer {
OverseasDeliveryCompletedV1 value = objectMapper.treeToValue(event.payload(), OverseasDeliveryCompletedV1.class);
if (value.completedAt() == null || !StringUtils.hasText(value.deliveryNo())
|| !StringUtils.hasText(value.sourceOrderId()) || !StringUtils.hasText(value.sourceOrderNo())
|| !StringUtils.hasText(value.inboundCode())) throw new IllegalArgumentException("字段不完整");
|| !StringUtils.hasText(value.inboundCode())) throw new EventProtocolException("字段不完整");
return value;
} catch (Exception exception) {
throw new IllegalArgumentException("海外派送完成事件负载无效", exception);
throw new EventProtocolException("海外派送完成事件负载无效", exception);
}
}
private InboundOrderEntity required(long inboundId, long tenantId, OverseasDeliveryCompletedV1 payload) {
InboundOrderEntity inbound = inboundOrders.selectActiveByIdForUpdate(inboundId);
if (inbound == null || !Objects.equals(inbound.getTenantId(), tenantId)
if (inbound == null) throw new EventDependencyNotReadyException("来源头程入仓单不存在");
if (!Objects.equals(inbound.getTenantId(), tenantId)
|| !Objects.equals(inbound.getInboundCode(), payload.inboundCode())
|| !Objects.equals(String.valueOf(inbound.getOrderId()), payload.sourceOrderId())
|| !Objects.equals(inbound.getOrderNo(), payload.sourceOrderNo())
|| (StringUtils.hasText(payload.fbaWarehouseCode()) && StringUtils.hasText(inbound.getDestinationCode())
&& !Objects.equals(inbound.getDestinationCode(), payload.fbaWarehouseCode()))) {
throw new IllegalArgumentException("事件与来源头程入仓单归属不一致");
throw new EventProtocolException("事件与来源头程入仓单归属不一致");
}
return inbound;
}
@@ -146,6 +149,6 @@ public class OverseasDeliveryCompletedConsumer {
private static long positive(String value, String message) {
try { long parsed = Long.parseLong(value); if (parsed > 0) return parsed; }
catch (RuntimeException ignored) { }
throw new IllegalArgumentException(message);
throw new EventProtocolException(message);
}
}
firstleg/src/main/java/com/baiyue/firstleg/service/OverseasInboundOrderDetainedConsumer.java
+11 -9
View File
@@ -2,6 +2,8 @@ package com.baiyue.firstleg.service;
import com.baomidou.mybatisplus.core.toolkit.IdWorker;
import com.baiyue.common.event.EventEnvelope;
import com.baiyue.common.event.EventProtocolException;
import com.baiyue.common.event.EventDependencyNotReadyException;
import com.baiyue.common.event.DeliveryAck;
import com.baiyue.common.event.EventPayloadCanonicalizer;
import com.baiyue.common.event.EventPayloadHash;
@@ -41,17 +43,17 @@ public class OverseasInboundOrderDetainedConsumer {
long inboundId = parsePositive(payload.sourceInboundId(), "来源入仓单 ID 无效");
long tenantId = parsePositive(event.tenantId(), "客户 ID 无效");
if (!payload.sourceInboundId().equals(event.aggregateId())) {
throw new IllegalArgumentException("事件 aggregateId 与来源入仓单 ID 不一致");
throw new EventProtocolException("事件 aggregateId 与来源入仓单 ID 不一致");
}
if (payload.changedAt() == null || payload.inboundCode() == null || payload.inboundCode().isBlank()
|| payload.reason() == null || payload.reason().isBlank()) {
throw new IllegalArgumentException("暂扣事件缺少入仓码、原因或变更时间");
throw new EventProtocolException("暂扣事件缺少入仓码、原因或变更时间");
}
ContainerEventInboxEntity existing = inbox.selectById(event.eventId());
if (existing != null) {
if (!existing.getPayloadHash().equalsIgnoreCase(event.payloadHash())) {
throw new IllegalArgumentException("相同 eventId 对应的事件摘要冲突");
throw new EventProtocolException("相同 eventId 对应的事件摘要冲突");
}
if (!"PROCESSED".equals(existing.getStatus())) throw new IllegalStateException("事件尚未完成消费");
return DeliveryAck.of(event, DeliveryAck.Outcome.DUPLICATE);
@@ -87,23 +89,23 @@ public class OverseasInboundOrderDetainedConsumer {
if (!EventTypes.OVERSEAS_INBOUND_ORDER_DETAINED_CHANGED.equals(event.eventType())
|| event.eventVersion() != 1 || event.sourceRegion() != RegionCode.OVERSEAS
|| event.targetRegion() != RegionCode.CN || !"INBOUND_ORDER".equals(event.aggregateType())) {
throw new IllegalArgumentException("不支持的海外入仓单暂扣事件");
throw new EventProtocolException("不支持的海外入仓单暂扣事件");
}
if (!EventPayloadHash.matches(EventPayloadCanonicalizer.canonicalize(event.payload()), event.payloadHash())) {
throw new IllegalArgumentException("payloadHash 与事件负载不一致");
throw new EventProtocolException("payloadHash 与事件负载不一致");
}
}
private OverseasInboundOrderDetainedChangedV1 payload(EventEnvelope event) {
try { return objectMapper.treeToValue(event.payload(), OverseasInboundOrderDetainedChangedV1.class); }
catch (Exception exception) { throw new IllegalArgumentException("海外入仓单暂扣事件负载无效", exception); }
catch (Exception exception) { throw new EventProtocolException("海外入仓单暂扣事件负载无效", exception); }
}
private InboundOrderEntity requiredInbound(long inboundId, long tenantId, String inboundCode) {
InboundOrderEntity value = inboundOrders.selectActiveByIdForUpdate(inboundId);
if (value == null) throw new IllegalArgumentException("来源头程入仓单不存在");
if (value == null) throw new EventDependencyNotReadyException("来源头程入仓单不存在");
if (!Objects.equals(value.getTenantId(), tenantId) || !Objects.equals(value.getInboundCode(), inboundCode)) {
throw new IllegalArgumentException("事件与来源头程入仓单归属不一致");
throw new EventProtocolException("事件与来源头程入仓单归属不一致");
}
return value;
}
@@ -130,7 +132,7 @@ public class OverseasInboundOrderDetainedConsumer {
private long parsePositive(String value, String message) {
try { long parsed = Long.parseLong(value); if (parsed > 0) return parsed; }
catch (RuntimeException ignored) { }
throw new IllegalArgumentException(message);
throw new EventProtocolException(message);
}
private long parseOperatorId(String value) {
firstleg/src/test/java/com/baiyue/firstleg/controller/EventConsumptionExceptionHandlerTest.java
+22
View File
@@ -0,0 +1,22 @@
package com.baiyue.firstleg.controller;
import com.baiyue.common.event.EventProtocolException;
import com.baiyue.common.event.EventDependencyNotReadyException;
import org.junit.jupiter.api.Test;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RestController;
import org.springframework.test.web.servlet.setup.MockMvcBuilders;
import static org.springframework.test.web.servlet.request.MockMvcRequestBuilders.get;
import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.*;
class EventConsumptionExceptionHandlerTest {
@RestController static class Probe {
@GetMapping("/protocol") void protocol() { throw new EventProtocolException("invalid hash"); }
@GetMapping("/dependency") void dependency() { throw new EventDependencyNotReadyException("source not ready"); }
}
@Test void protocolAndDependencyHaveDistinctWireCodes() throws Exception {
var mvc = MockMvcBuilders.standaloneSetup(new Probe()).setControllerAdvice(new EventConsumptionExceptionHandler()).build();
mvc.perform(get("/protocol")).andExpect(status().isOk()).andExpect(jsonPath("$.code").value("EVT_PROTOCOL"));
mvc.perform(get("/dependency")).andExpect(status().isOk()).andExpect(jsonPath("$.code").value("EVT_DEPENDENCY"));
}
}
lastleg-service/src/main/java/com/baiyue/lastleg/client/DirectEventClient.java
+12 -3
View File
@@ -5,12 +5,17 @@ import com.baiyue.common.event.DeliveryAck;
import com.baiyue.common.event.EventEnvelope;
import com.baiyue.common.event.EventTypes;
import lombok.RequiredArgsConstructor;
import com.baiyue.common.web.EventDeliveryFailure;
import java.util.Set;
import org.springframework.stereotype.Component;
/** Dispatches each supported business event directly to its owning service. */
@Component
@RequiredArgsConstructor
public class DirectEventClient {
private static final Set<String> FINANCE_PERMANENT_CODES = Set.of(
"FIN-EVT-001", "FIN-EVT-002", "FIN-EVT-004", "FIN-EVT-007", "FIN-EVT-008",
"FIN-EVT-012", "FIN-EVT-015", "FIN-EVT-016", "FIN-EVT-401");
private final FirstlegEventClient firstleg;
private final FinanceBusinessEventClient finance;
@@ -20,9 +25,13 @@ public class DirectEventClient {
case EventTypes.OVERSEAS_CONTAINER_UNLOADED,
EventTypes.OVERSEAS_INBOUND_ORDER_DETAINED_CHANGED,
EventTypes.OVERSEAS_DELIVERY_COMPLETED -> firstleg.receive(token, event);
case EventTypes.WALLET_COMMAND_REQUESTED -> finance.wallet(token, event);
case EventTypes.BUSINESS_FEE_SOURCE_CHANGED -> finance.fee(token, event);
default -> throw new IllegalArgumentException("Unsupported outbound event type: " + event.eventType());
case EventTypes.WALLET_COMMAND_REQUESTED -> checkedFinance(finance.wallet(token, event));
case EventTypes.BUSINESS_FEE_SOURCE_CHANGED -> checkedFinance(finance.fee(token, event));
default -> throw new EventDeliveryFailure(true, "Unsupported outbound event type: " + event.eventType());
};
}
private ApiResponse<DeliveryAck> checkedFinance(ApiResponse<DeliveryAck> response) {
EventDeliveryFailure.requireSuccessful(response, FINANCE_PERMANENT_CODES);
return response;
}
}
lastleg-service/src/main/java/com/baiyue/lastleg/controller/EventConsumptionExceptionHandler.java
+21
View File
@@ -0,0 +1,21 @@
package com.baiyue.lastleg.controller;
import com.baiyue.common.core.api.ApiResponse;
import com.baiyue.common.event.EventProtocolException;
import com.baiyue.common.event.EventDependencyNotReadyException;
import org.slf4j.MDC;
import org.springframework.core.Ordered;
import org.springframework.core.annotation.Order;
import org.springframework.web.bind.annotation.ExceptionHandler;
import org.springframework.web.bind.annotation.RestControllerAdvice;
import java.time.Instant;
@RestControllerAdvice
@Order(Ordered.HIGHEST_PRECEDENCE)
public class EventConsumptionExceptionHandler {
@ExceptionHandler({EventProtocolException.class, EventDependencyNotReadyException.class})
public ApiResponse<Void> event(IllegalArgumentException failure) {
String code = failure instanceof EventDependencyNotReadyException ? "EVT_DEPENDENCY" : "EVT_PROTOCOL";
return new ApiResponse<>(code, failure.getMessage(), null, MDC.get("requestId"), MDC.get("traceId"), Instant.now());
}
}
lastleg-service/src/main/java/com/baiyue/lastleg/controller/LastlegOrderController.java
+1 -1
View File
@@ -73,7 +73,7 @@ public class LastlegOrderController {
} else if (EventTypes.FIRST_LEG_LOADING_PLAN_COMPLETED.equals(event.eventType())) {
consumed = consumer.consume(event);
} else {
throw new ResponseStatusException(HttpStatus.BAD_REQUEST, "Lastleg 不消费该事件类型: " + event.eventType());
throw new com.baiyue.common.event.EventProtocolException("Lastleg 不消费该事件类型: " + event.eventType());
}
return ApiResponse.success(consumed);
}
lastleg-service/src/main/java/com/baiyue/lastleg/financefee/service/OverseasFeeOutboxRelay.java
+10 -4
View File
@@ -1,6 +1,8 @@
package com.baiyue.lastleg.financefee.service;
import com.baiyue.common.event.EventEnvelope;
import com.baiyue.common.web.EventDeliveryFailure;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.baiyue.common.core.retry.BoundedRetryPolicy;
import com.baiyue.lastleg.client.DirectEventClient;
import com.baiyue.lastleg.entity.LastlegEventOutbox;
@@ -46,17 +48,21 @@ public class OverseasFeeOutboxRelay {
repository.dead(row.getId(), worker, failures, "自动投递次数已耗尽（含租约超时）");
continue;
}
boolean requestStarted = false;
try {
EventEnvelope envelope = objectMapper.readValue(row.getPayloadJson(), EventEnvelope.class).requireValid();
requestStarted = true;
var response = client.receive(token, envelope);
if (response == null || !response.successful() || response.data() == null)
throw new IllegalStateException("Finance未确认费用事件");
EventDeliveryFailure.requireSuccessful(response);
response.data().requireMatches(envelope);
repository.published(row.getId(), worker, LocalDateTime.now(ZoneOffset.UTC));
} catch (Exception exception) {
String error = EventDeliveryFailure.diagnostic(exception, token);
boolean permanent = EventDeliveryFailure.isPermanent(exception)
|| (!requestStarted && (exception instanceof JsonProcessingException || exception instanceof IllegalArgumentException));
int retries = failures + 1;
if (policy.exhausted(retries)) repository.dead(row.getId(), worker, retries, exception.getMessage());
else repository.retry(row.getId(), worker, retries, LocalDateTime.now(ZoneOffset.UTC).plus(policy.delayAfterFailure(retries)), exception.getMessage());
if (permanent || policy.exhausted(retries)) repository.dead(row.getId(), worker, retries, error);
else repository.retry(row.getId(), worker, retries, LocalDateTime.now(ZoneOffset.UTC).plus(policy.delayAfterFailure(retries)), error);
}
}
}
lastleg-service/src/main/java/com/baiyue/lastleg/inboundplan/pda/service/InboundUnloadCompletedOutboxRelay.java
+11 -9
View File
@@ -1,6 +1,8 @@
package com.baiyue.lastleg.inboundplan.pda.service;
import com.baiyue.common.event.EventEnvelope;
import com.baiyue.common.web.EventDeliveryFailure;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.baiyue.common.core.retry.BoundedRetryPolicy;
import com.baiyue.common.event.DeliveryAck;
import com.baiyue.common.event.EventEnvelopeValidator;
@@ -51,29 +53,29 @@ public class InboundUnloadCompletedOutboxRelay {
repository.dead(row.getId(), workerId, failures, "自动投递次数已耗尽（含租约超时）");
continue;
}
boolean requestStarted = false;
try {
EventEnvelope event = objectMapper.readValue(row.getPayloadJson(), EventEnvelope.class);
EventEnvelopeValidator.requireValid(event);
requestStarted = true;
var response = client.receive(internalToken, event);
requireAcknowledgement(event, response);
repository.published(row.getId(), workerId, LocalDateTime.now());
} catch (Exception exception) {
if (policy.exhausted(failures + 1)) repository.dead(row.getId(), workerId, failures + 1, exception.getMessage());
String error = EventDeliveryFailure.diagnostic(exception, internalToken);
boolean permanent = EventDeliveryFailure.isPermanent(exception)
|| (!requestStarted && (exception instanceof JsonProcessingException || exception instanceof IllegalArgumentException));
if (permanent || policy.exhausted(failures + 1)) repository.dead(row.getId(), workerId, failures + 1, error);
else repository.retry(row.getId(), workerId, LocalDateTime.now().plus(policy.delayAfterFailure(failures + 1)),
exception.getMessage());
log.warn("海外到国内事件投递失败: eventKey={}, reason={}", row.getEventKey(), exception.getMessage());
error);
log.warn("海外到国内事件投递失败: eventKey={}, reason={}", row.getEventKey(), error);
}
}
}
private void requireAcknowledgement(EventEnvelope event,
com.baiyue.common.core.api.ApiResponse<DeliveryAck> response) {
if (response == null) throw new IllegalStateException("消费服务返回空响应");
if (!response.successful()) {
throw new IllegalStateException("消费服务拒绝海外事件: code=" + response.code()
+ ", message=" + response.message() + ", requestId=" + response.requestId());
}
if (response.data() == null) throw new IllegalStateException("消费服务成功响应缺少确认数据");
EventDeliveryFailure.requireSuccessful(response);
response.data().requireMatches(event);
}
}
lastleg-service/src/main/java/com/baiyue/lastleg/service/LoadingCompletedConsumer.java
+40 -40
View File
@@ -55,23 +55,23 @@ public class LoadingCompletedConsumer {
envelope.requireValid();
if (!EventTypes.FIRST_LEG_LOADING_PLAN_COMPLETED.equals(envelope.eventType())
|| (envelope.eventVersion() != 1 && envelope.eventVersion() != 2 && envelope.eventVersion() != 3))
throw new IllegalArgumentException("不支持的尾程交接事件");
throw new EventProtocolException("不支持的尾程交接事件");
if (envelope.sourceRegion() != RegionCode.CN || envelope.targetRegion() != RegionCode.OVERSEAS)
throw new IllegalArgumentException("尾程交接事件区域无效");
throw new EventProtocolException("尾程交接事件区域无效");
if (!"LOADING_PLAN".equals(envelope.aggregateType()))
throw new IllegalArgumentException("尾程交接事件聚合类型无效");
throw new EventProtocolException("尾程交接事件聚合类型无效");
if (!EventPayloadHash.matches(EventPayloadCanonicalizer.canonicalize(envelope.payload()),
envelope.payloadHash())) throw new IllegalArgumentException("payloadHash 与尾程事件负载不一致");
envelope.payloadHash())) throw new EventProtocolException("payloadHash 与尾程事件负载不一致");
FirstLegLoadingPlanCompletedV1 payload = parsePayload(envelope);
String containerNo = ContainerNumbers.requireCanonical(payload.containerNo());
String sourceContainerBookingId = sourceContainerBookingId(envelope, payload);
if(payload.orders()==null||payload.orders().isEmpty()) throw new IllegalArgumentException("事件没有入仓单");
if(payload.orders()==null||payload.orders().isEmpty()) throw new EventProtocolException("事件没有入仓单");
OwnerSnapshot owner = requireOwnerSnapshot(payload.orders(), envelope.tenantId());
if (envelope.eventVersion() >= 2) validateCartonSkuMappings(payload, envelope.payload());
freezeContainerIdentity(envelope, sourceContainerBookingId, containerNo);
LastlegEventInbox existing=inboxMapper.selectById(envelope.eventId());
if(existing!=null) {
if(!existing.getPayloadHash().equalsIgnoreCase(envelope.payloadHash())) throw new IllegalArgumentException("事件摘要冲突");
if(!existing.getPayloadHash().equalsIgnoreCase(envelope.payloadHash())) throw new EventProtocolException("事件摘要冲突");
if (!"PROCESSED".equals(existing.getStatus())) throw new IllegalStateException("事件尚未完成消费");
return DeliveryAck.of(envelope, DeliveryAck.Outcome.DUPLICATE);
}
@@ -84,7 +84,7 @@ public class LoadingCompletedConsumer {
LastlegEventInbox raced = inboxMapper.selectById(envelope.eventId());
if (raced == null || raced.getPayloadHash() == null
|| !raced.getPayloadHash().equalsIgnoreCase(envelope.payloadHash())) {
throw new IllegalArgumentException("事件摘要冲突", exception);
throw new EventProtocolException("事件摘要冲突", exception);
}
if (!"PROCESSED".equals(raced.getStatus())) throw new IllegalStateException("事件尚未完成消费");
return DeliveryAck.of(envelope, DeliveryAck.Outcome.DUPLICATE);
@@ -111,7 +111,7 @@ public class LoadingCompletedConsumer {
if (envelope.eventVersion() < 3) return;
JsonNode values = envelope.payload().path("loadingEvidence");
if (!values.isArray() || values.size() != 5) {
throw new IllegalArgumentException("V3 尾程交接事件必须包含五类国内装柜照片");
throw new EventProtocolException("V3 尾程交接事件必须包含五类国内装柜照片");
}
Set<String> expected = Set.of("PHOTO_STEP_1", "PHOTO_STEP_2", "PHOTO_STEP_3", "PHOTO_STEP_4", "SEAL");
Set<String> actual = new HashSet<>();
@@ -121,7 +121,7 @@ public class LoadingCompletedConsumer {
String fileName = value.path("fileName").asText("").trim();
if (!expected.contains(type) || !actual.add(type) || fileId <= 0
|| fileName.isBlank() || fileName.length() > 255) {
throw new IllegalArgumentException("V3 尾程交接事件的国内装柜照片引用无效");
throw new EventProtocolException("V3 尾程交接事件的国内装柜照片引用无效");
}
ContainerLoadingEvidence existing = loadingEvidence.find(sourceContainerBookingId, type);
if (existing != null) {
@@ -143,7 +143,7 @@ public class LoadingCompletedConsumer {
|| !Objects.equals(raced.getFileName(), fileName)) throw duplicate;
}
}
if (!actual.equals(expected)) throw new IllegalArgumentException("V3 尾程交接事件的国内装柜照片类型不完整");
if (!actual.equals(expected)) throw new EventProtocolException("V3 尾程交接事件的国内装柜照片类型不完整");
}
private FirstLegLoadingPlanCompletedV1 parsePayload(EventEnvelope envelope) {
@@ -154,7 +154,7 @@ public class LoadingCompletedConsumer {
// identity is validated separately from the original immutable payload.
return compatibleMapper.treeToValue(envelope.payload(), FirstLegLoadingPlanCompletedV1.class);
}
catch(Exception exception) { throw new IllegalArgumentException("尾程事件负载无效",exception); }
catch(Exception exception) { throw new EventProtocolException("尾程事件负载无效",exception); }
}
private int countOrders(EventEnvelope envelope,String containerNo) {
@@ -184,7 +184,7 @@ public class LoadingCompletedConsumer {
? envelope.payload().path("sourceContainerBookingId").asText(null)
: payload.containerBookingId();
if (value == null || value.isBlank() || value.trim().length() > 128) {
throw new IllegalArgumentException("事件缺少稳定来源柜预订 ID sourceContainerBookingId");
throw new EventProtocolException("事件缺少稳定来源柜预订 ID sourceContainerBookingId");
}
return value.trim();
}
@@ -194,12 +194,12 @@ public class LoadingCompletedConsumer {
String customerId = requiredOwnerValue(orders.get(0).customerId(), "customerId");
String customerName = requiredOwnerValue(orders.get(0).customerName(), "customerName");
if (!trustedCustomerId.equals(customerId)) {
throw new IllegalArgumentException("Order customerId does not match the trusted event customer");
throw new EventProtocolException("Order customerId does not match the trusted event customer");
}
for (FirstLegLoadingPlanCompletedV1.Order order : orders) {
if (!customerId.equals(requiredOwnerValue(order.customerId(), "customerId"))
|| !customerName.equals(requiredOwnerValue(order.customerName(), "customerName"))) {
throw new IllegalArgumentException("同一装柜完成事件的货主快照不一致");
throw new EventProtocolException("同一装柜完成事件的货主快照不一致");
}
}
return new OwnerSnapshot(customerId, customerName);
@@ -207,11 +207,11 @@ public class LoadingCompletedConsumer {
private String requiredOwnerValue(String value, String field) {
if (value == null || value.isBlank()) {
throw new IllegalArgumentException("尾程事件缺少不可变货主快照 " + field);
throw new EventProtocolException("尾程事件缺少不可变货主快照 " + field);
}
String normalized = value.trim();
if (normalized.length() > 128) {
throw new IllegalArgumentException("尾程事件货主快照过长 " + field);
throw new EventProtocolException("尾程事件货主快照过长 " + field);
}
return normalized;
}
@@ -231,11 +231,11 @@ public class LoadingCompletedConsumer {
}
int manifestPackages = rawPallets.path(palletIndex).path("manifestPackageCount").asInt(-1);
if (palletCartonMarks.size() != manifestPackages) {
throw new IllegalArgumentException("一件代发托盘物理箱数与清单箱数不一致: " + pallet.palletNo());
throw new EventProtocolException("一件代发托盘物理箱数与清单箱数不一致: " + pallet.palletNo());
}
for (String mark : palletCartonMarks) {
if (!orderCartonMarks.add(mark)) {
throw new IllegalArgumentException("一件代发物理箱不能跨托盘重复: " + mark);
throw new EventProtocolException("一件代发物理箱不能跨托盘重复: " + mark);
}
}
}
@@ -257,7 +257,7 @@ public class LoadingCompletedConsumer {
Set<String> orderCartonMarks) {
List<String> marks = cartonMarks == null ? List.of() : cartonMarks;
if (skuCode == null || skuCode.isBlank()) {
throw new IllegalArgumentException("V2 尾程交接事件缺少 SKU");
throw new EventProtocolException("V2 尾程交接事件缺少 SKU");
}
if (marks.isEmpty()) return;
requireSkuCartonMapping(skuCode, loadedPackages, marks, orderCartonMarks);
@@ -266,15 +266,15 @@ public class LoadingCompletedConsumer {
private void requireSkuCartonMapping(String skuCode, int loadedPackages, List<String> cartonMarks,
Set<String> orderCartonMarks) {
if (skuCode == null || skuCode.isBlank()) {
throw new IllegalArgumentException("V2 尾程交接事件的箱唛映射缺少 SKU");
throw new EventProtocolException("V2 尾程交接事件的箱唛映射缺少 SKU");
}
List<String> marks = cartonMarks == null ? List.of() : cartonMarks;
if (marks.size() != loadedPackages) {
throw new IllegalArgumentException("V2 尾程交接事件的 SKU 箱唛数量必须等于实装箱数: " + skuCode);
throw new EventProtocolException("V2 尾程交接事件的 SKU 箱唛数量必须等于实装箱数: " + skuCode);
}
for (String mark : marks) {
if (mark == null || mark.isBlank() || !orderCartonMarks.add(mark)) {
throw new IllegalArgumentException("V2 尾程交接事件包含空值或重复箱唛: " + skuCode);
throw new EventProtocolException("V2 尾程交接事件包含空值或重复箱唛: " + skuCode);
}
}
}
@@ -298,7 +298,7 @@ public class LoadingCompletedConsumer {
List<DropshipRequests.Pallet> dropshipPallets = "ONE_CLICK_DROPSHIP".equals(orderType)
? pallets(source, rawSource) : List.of();
if ("ONE_CLICK_DROPSHIP".equals(orderType) && dropshipPallets.isEmpty()) {
throw new IllegalArgumentException("One-click dropship event has no pallet snapshot");
throw new EventProtocolException("One-click dropship event has no pallet snapshot");
}
order.setId(IdWorker.getId()); order.setOrderNo(orderNo); order.setCustomerId(customerId); order.setOrderType(orderType);
order.setDestinationType(destinationType);
@@ -346,12 +346,12 @@ public class LoadingCompletedConsumer {
for (int palletIndex = 0; palletIndex < pallets.size(); palletIndex++) {
String palletNo = pallets.get(palletIndex).palletNo();
if (palletNo == null || palletNo.isBlank() || palletNo.length() > 128 || !palletNos.add(palletNo)) {
throw new IllegalArgumentException("散货来源托盘码为空、重复或过长");
throw new EventProtocolException("散货来源托盘码为空、重复或过长");
}
sourcePallets.insert(IdWorker.getId(), customerId, order.getId(), palletNo.trim(), palletIndex + 1, event.eventId());
}
if (!pallets.isEmpty() && order.getPalletCount() > 0 && pallets.size() != order.getPalletCount()) {
throw new IllegalArgumentException("散货来源托盘码数量与托盘数不一致");
throw new EventProtocolException("散货来源托盘码数量与托盘数不一致");
}
}
for(var item:items) {
@@ -379,7 +379,7 @@ public class LoadingCompletedConsumer {
if ("ONE_CLICK_DROPSHIP".equals(orderType)) {
long warehouseId;
try { warehouseId = Long.parseLong(payload.warehouseId()); }
catch (Exception exception) { throw new IllegalArgumentException("One-click dropship warehouseId is invalid", exception); }
catch (Exception exception) { throw new EventProtocolException("One-click dropship warehouseId is invalid", exception); }
dropshipOrders.receive(new DropshipRequests.Receive(customerId, order.getId(), warehouseId,
event.eventId() + ":DROPSHIP:" + source.sourceInboundId(), dropshipPallets));
}
@@ -390,7 +390,7 @@ public class LoadingCompletedConsumer {
try {
return LocalDateTime.parse(value.trim());
} catch (java.time.format.DateTimeParseException exception) {
throw new IllegalArgumentException("事件计划到仓时间 estimatedWarehouseArrivalAt 无效", exception);
throw new EventProtocolException("事件计划到仓时间 estimatedWarehouseArrivalAt 无效", exception);
}
}
@@ -447,7 +447,7 @@ public class LoadingCompletedConsumer {
String detailId = sku.path("sourceInboundDetailId").asText(null);
var item = itemsByDetail.get(detailId);
if (item == null || !Objects.equals(item.skuCode(), sku.path("skuCode").asText(null))) {
throw new IllegalArgumentException("一件代发托盘来源明细缺失或与 SKU 不一致");
throw new EventProtocolException("一件代发托盘来源明细缺失或与 SKU 不一致");
}
List<String> marks = marksByDetail.computeIfAbsent(detailId, ignored -> new java.util.ArrayList<>());
sku.path("cartonMarks").forEach(mark -> marks.add(mark.asText()));
@@ -455,7 +455,7 @@ public class LoadingCompletedConsumer {
}
for (var item : itemsByDetail.values()) {
if (marksByDetail.getOrDefault(item.sourceInboundDetailId(), List.of()).size() != item.packageCount()) {
throw new IllegalArgumentException("一件代发托盘箱唛与来源明细箱数不一致");
throw new EventProtocolException("一件代发托盘箱唛与来源明细箱数不一致");
}
}
return marksByDetail;
@@ -466,7 +466,7 @@ public class LoadingCompletedConsumer {
Map<String, List<FirstLegLoadingPlanCompletedV1.Item>> byDetail = new java.util.LinkedHashMap<>();
for (var item : source == null ? List.<FirstLegLoadingPlanCompletedV1.Item>of() : source) {
if (item.sourceInboundDetailId() == null || item.sourceInboundDetailId().isBlank()) {
throw new IllegalArgumentException("一件代发来源明细标识缺失");
throw new EventProtocolException("一件代发来源明细标识缺失");
}
byDetail.computeIfAbsent(item.sourceInboundDetailId(), ignored -> new ArrayList<>()).add(item);
}
@@ -476,7 +476,7 @@ public class LoadingCompletedConsumer {
if (fragments.stream().anyMatch(item -> !Objects.equals(first.skuCode(), item.skuCode())
|| !Objects.equals(first.destinationWarehouseId(), item.destinationWarehouseId())
|| !Objects.equals(first.destinationWarehouseCode(), item.destinationWarehouseCode()))) {
throw new IllegalArgumentException("同一来源明细的尾程事件分片快照不一致");
throw new EventProtocolException("同一来源明细的尾程事件分片快照不一致");
}
List<String> marks = fragments.stream().flatMap(item -> item.cartonMarks().stream()).toList();
result.add(new FirstLegLoadingPlanCompletedV1.Item(first.sourceInboundDetailId(), first.skuCode(),
@@ -512,7 +512,7 @@ public class LoadingCompletedConsumer {
List<String> marks = sourceMarks == null ? List.of() : sourceMarks;
if (marks.isEmpty()) return;
if (marks.stream().anyMatch(mark -> mark == null || mark.isBlank()) || marks.stream().distinct().count() != marks.size()) {
throw new IllegalArgumentException("装柜事件的箱唛池不可包含空值或重复箱唛");
throw new EventProtocolException("装柜事件的箱唛池不可包含空值或重复箱唛");
}
for (String mark : marks) {
InboundPlanCarton carton = new InboundPlanCarton();
@@ -536,18 +536,18 @@ public class LoadingCompletedConsumer {
boolean overseasWarehouseIdentity = hasText(payload.warehouseId())
&& Objects.equals(payload.warehouseId(), order.destinationWarehouseId());
if (fbaEvidence && overseasWarehouseIdentity) {
throw new IllegalArgumentException("历史尾程事件目的类型证据冲突: " + order.sourceInboundId());
throw new EventProtocolException("历史尾程事件目的类型证据冲突: " + order.sourceInboundId());
}
if (fbaEvidence) return "FBA";
if (overseasWarehouseIdentity) return "OVERSEAS";
throw new IllegalArgumentException("历史尾程事件缺少 destinationType 且无法安全推断: " + order.sourceInboundId());
throw new EventProtocolException("历史尾程事件缺少 destinationType 且无法安全推断: " + order.sourceInboundId());
}
private String normalizeDestinationType(String value) {
if (!hasText(value)) return null;
String normalized = value.trim().toUpperCase(java.util.Locale.ROOT);
if (!"FBA".equals(normalized) && !"OVERSEAS".equals(normalized)) {
throw new IllegalArgumentException("尾程事件 destinationType 必须为 FBA 或 OVERSEAS");
throw new EventProtocolException("尾程事件 destinationType 必须为 FBA 或 OVERSEAS");
}
return normalized;
}
@@ -577,7 +577,7 @@ public class LoadingCompletedConsumer {
List<String> marks = new ArrayList<>(left.cartonMarks()); marks.addAll(right.cartonMarks());
Map<String, Long> quantities = new java.util.LinkedHashMap<>(left.cartonQuantities());
right.cartonQuantities().forEach((mark, quantity) -> {
if (quantities.putIfAbsent(mark, quantity) != null) throw new IllegalArgumentException("Duplicate carton quantity: " + mark);
if (quantities.putIfAbsent(mark, quantity) != null) throw new EventProtocolException("Duplicate carton quantity: " + mark);
});
return new DropshipRequests.SkuPackages(left.skuCode(), left.productName(),
Math.addExact(left.packageCount(), right.packageCount()),
@@ -620,25 +620,25 @@ public class LoadingCompletedConsumer {
private int positiveManifest(JsonNode source, String field) {
int value = source.path(field).asInt(0);
if (value <= 0) throw new IllegalArgumentException("One-click dropship pallet is missing " + field);
if (value <= 0) throw new EventProtocolException("One-click dropship pallet is missing " + field);
return value;
}
private String requiredManifestHash(JsonNode source) {
String value = source.path("manifestHash").asText(null);
if (value == null || !value.matches("(?i)[0-9a-f]{64}")) {
throw new IllegalArgumentException("One-click dropship pallet manifestHash is invalid");
throw new EventProtocolException("One-click dropship pallet manifestHash is invalid");
}
return value.toLowerCase(java.util.Locale.ROOT);
}
private Map<String, Long> cartonQuantities(JsonNode sku) {
JsonNode node = sku.path("cartonQuantities");
if (!node.isObject()) throw new IllegalArgumentException("One-click dropship cartonQuantities is required");
if (!node.isObject()) throw new EventProtocolException("One-click dropship cartonQuantities is required");
Map<String, Long> result = new java.util.LinkedHashMap<>();
node.fields().forEachRemaining(entry -> {
if (!entry.getValue().isIntegralNumber() || !entry.getValue().canConvertToLong() || entry.getValue().longValue() <= 0) {
throw new IllegalArgumentException("Invalid carton quantity: " + entry.getKey());
throw new EventProtocolException("Invalid carton quantity: " + entry.getKey());
}
result.put(entry.getKey(), entry.getValue().longValue());
});
lastleg-service/src/main/java/com/baiyue/lastleg/shipping/service/ContainerShippingNodeConsumer.java
+11 -8
View File
@@ -1,6 +1,8 @@
package com.baiyue.lastleg.shipping.service;
import com.baiyue.common.event.EventEnvelope;
import com.baiyue.common.event.EventProtocolException;
import com.baiyue.common.event.EventDependencyNotReadyException;
import com.baiyue.common.event.DeliveryAck;
import com.baiyue.common.event.EventPayloadCanonicalizer;
import com.baiyue.common.event.EventPayloadHash;
@@ -36,7 +38,7 @@ public class ContainerShippingNodeConsumer {
LastlegEventInbox existing = inboxes.selectById(envelope.eventId());
if (existing != null) {
if (!existing.getPayloadHash().equalsIgnoreCase(envelope.payloadHash())) {
throw new IllegalArgumentException("船节点事件摘要冲突");
throw new EventProtocolException("船节点事件摘要冲突");
}
if (!"PROCESSED".equals(existing.getStatus())) throw new IllegalStateException("事件尚未完成消费");
return DeliveryAck.of(envelope, DeliveryAck.Outcome.DUPLICATE);
@@ -46,7 +48,7 @@ public class ContainerShippingNodeConsumer {
catch (DuplicateKeyException raced) {
LastlegEventInbox concurrent = inboxes.selectById(envelope.eventId());
if (concurrent == null || !concurrent.getPayloadHash().equalsIgnoreCase(envelope.payloadHash())) {
throw new IllegalArgumentException("船节点事件摘要冲突", raced);
throw new EventProtocolException("船节点事件摘要冲突", raced);
}
if (!"PROCESSED".equals(concurrent.getStatus())) throw new IllegalStateException("事件尚未完成消费");
return DeliveryAck.of(envelope, DeliveryAck.Outcome.DUPLICATE);
@@ -55,8 +57,9 @@ public class ContainerShippingNodeConsumer {
gates.initialize(payload.containerBookingId(), payload.oceanBookingId(),
payload.containerNo(), 0, 1);
ContainerShippingGate gate = gates.lock(payload.containerBookingId());
if (gate == null || !payload.containerNo().equals(gate.getContainerNo())) {
throw new IllegalArgumentException("船节点柜号快照与装柜交接不一致");
if (gate == null) throw new EventDependencyNotReadyException("船节点柜级投影尚未就绪");
if (!payload.containerNo().equals(gate.getContainerNo())) {
throw new EventProtocolException("船节点柜号快照与装柜交接不一致");
}
boolean applied = apply(gate, envelope, payload);
gates.updateById(gate);
@@ -98,7 +101,7 @@ public class ContainerShippingNodeConsumer {
envelope.requireValid();
if (!EventPayloadHash.matches(EventPayloadCanonicalizer.canonicalize(envelope.payload()),
envelope.payloadHash())) {
throw new IllegalArgumentException("payloadHash 与船节点事件负载不一致");
throw new EventProtocolException("payloadHash 与船节点事件负载不一致");
}
if (!EventTypes.FIRST_LEG_CONTAINER_SHIPPING_NODE_RECORDED.equals(envelope.eventType())
|| envelope.eventVersion() != 1 || envelope.sourceRegion() != RegionCode.CN
@@ -106,19 +109,19 @@ public class ContainerShippingNodeConsumer {
|| !"CONTAINER_BOOKING".equals(envelope.aggregateType())
|| envelope.handoverVersion() == null
|| envelope.handoverVersion() != envelope.aggregateVersion() + 1) {
throw new IllegalArgumentException("船节点交接信封无效");
throw new EventProtocolException("船节点交接信封无效");
}
}
private void validateEnvelopePayload(EventEnvelope envelope, FirstLegContainerShippingNodeRecordedV1 payload) {
if (!"0".equals(envelope.tenantId()) || !envelope.aggregateId().equals(payload.containerBookingId())) {
throw new IllegalArgumentException("船节点交接平台作用域或聚合标识不一致");
throw new EventProtocolException("船节点交接平台作用域或聚合标识不一致");
}
}
private FirstLegContainerShippingNodeRecordedV1 payload(EventEnvelope envelope) {
try { return objectMapper.treeToValue(envelope.payload(), FirstLegContainerShippingNodeRecordedV1.class); }
catch (Exception exception) { throw new IllegalArgumentException("船节点交接负载无效", exception); }
catch (Exception exception) { throw new EventProtocolException("船节点交接负载无效", exception); }
}
private LastlegEventInbox inbox(EventEnvelope envelope) {
lastleg-service/src/main/java/com/baiyue/lastleg/wallet/WalletCommandCompletedConsumer.java
+4 -4
View File
@@ -14,11 +14,11 @@ import org.springframework.dao.DuplicateKeyException;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.web.server.ResponseStatusException;
import com.baiyue.common.event.EventProtocolException;
import java.util.List;
import static org.springframework.http.HttpStatus.BAD_REQUEST;
@Service
@RequiredArgsConstructor
@@ -67,12 +67,12 @@ public class WalletCommandCompletedConsumer {
|| result.businessType() == null || result.businessType().isBlank()
|| result.businessNo() == null || result.businessNo().isBlank()) throw bad("资金结果事件字段无效");
return result;
} catch (ResponseStatusException exception) { throw exception; }
} catch (EventProtocolException exception) { throw exception; }
catch (Exception exception) { throw bad("资金结果事件载荷无法解析"); }
}
private String json(Object value) {
try { return objectMapper.writeValueAsString(value); }
catch (Exception exception) { throw bad("资金结果事件序列化失败"); }
}
private static ResponseStatusException bad(String message) { return new ResponseStatusException(BAD_REQUEST, message); }
private static EventProtocolException bad(String message) { return new EventProtocolException(message); }
}
lastleg-service/src/test/java/com/baiyue/lastleg/client/DirectEventClientTest.java
+2 -1
View File
@@ -27,6 +27,7 @@ class DirectEventClientTest {
@Test void feeIsSentOnlyToFinance() {
var event = event(EventTypes.BUSINESS_FEE_SOURCE_CHANGED);
when(finance.fee("token", event)).thenReturn(ApiResponse.success(DeliveryAck.of(event, DeliveryAck.Outcome.APPLIED)));
client.receive("token", event);
verify(finance).fee("token", event);
verifyNoMoreInteractions(finance);
@@ -45,7 +46,7 @@ class DirectEventClientTest {
@Test void unsupportedEventsHaveNoDefaultDestination() {
assertThatThrownBy(() -> client.receive("token", event(EventTypes.WALLET_COMMAND_COMPLETED)))
.isInstanceOf(IllegalArgumentException.class).hasMessageContaining("Unsupported outbound");
.isInstanceOf(com.baiyue.common.web.EventDeliveryFailure.class).hasMessageContaining("Unsupported outbound");
verifyNoInteractions(firstleg, finance);
}
lastleg-service/src/test/java/com/baiyue/lastleg/controller/EventConsumptionExceptionHandlerTest.java
+22
View File
@@ -0,0 +1,22 @@
package com.baiyue.lastleg.controller;
import com.baiyue.common.event.EventProtocolException;
import com.baiyue.common.event.EventDependencyNotReadyException;
import org.junit.jupiter.api.Test;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RestController;
import org.springframework.test.web.servlet.setup.MockMvcBuilders;
import static org.springframework.test.web.servlet.request.MockMvcRequestBuilders.get;
import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.*;
class EventConsumptionExceptionHandlerTest {
@RestController static class Probe {
@GetMapping("/protocol") void protocol() { throw new EventProtocolException("invalid hash"); }
@GetMapping("/dependency") void dependency() { throw new EventDependencyNotReadyException("source not ready"); }
}
@Test void protocolAndDependencyHaveDistinctWireCodes() throws Exception {
var mvc = MockMvcBuilders.standaloneSetup(new Probe()).setControllerAdvice(new EventConsumptionExceptionHandler()).build();
mvc.perform(get("/protocol")).andExpect(status().isOk()).andExpect(jsonPath("$.code").value("EVT_PROTOCOL"));
mvc.perform(get("/dependency")).andExpect(status().isOk()).andExpect(jsonPath("$.code").value("EVT_DEPENDENCY"));
}
}
lastleg-service/src/test/java/com/baiyue/lastleg/controller/LastlegOrderControllerTest.java
+1 -1
View File
@@ -155,7 +155,7 @@ class LastlegOrderControllerTest {
when(event.eventType()).thenReturn(EventTypes.IAM_USER_STATUS_CHANGED);
LoadingCompletedConsumer loading = (LoadingCompletedConsumer) ReflectionTestUtils.getField(controller, "consumer");
assertThrows(ResponseStatusException.class, () -> controller.receive("internal-token", event));
assertThrows(com.baiyue.common.event.EventProtocolException.class, () -> controller.receive("internal-token", event));
verify(loading, never()).consume(event);
}
lastleg-service/src/test/java/com/baiyue/lastleg/inboundplan/pda/service/InboundUnloadCompletedOutboxRelayTest.java
+1 -1
View File
@@ -75,7 +75,7 @@ class InboundUnloadCompletedOutboxRelayTest {
relay.relay();
verify(repository).retry(eq(9001L), anyString(), any(LocalDateTime.class), eq("consumer unavailable"));
verify(repository).retry(eq(9001L), anyString(), any(LocalDateTime.class), contains("consumer unavailable"));
verify(repository, never()).published(anyLong(), anyString(), any());
}