Axon订阅查询抛出AxonServerException: CANCELLED: AXONIQ-5000问题求助

版本信息 #

• Axon framework 版本：4.4.7
• Axon Server 版本：4.5.7

相关代码 #

事件处理器代码 #

@Component
@ProcessingGroup("product-group")
public class ProductEventHandler {
    private final ProductRepository productsRepository;
    private final QueryUpdateEmitter queryUpdateEmitter;

    public ProductEventHandler(ProductRepository productsRepository, QueryUpdateEmitter queryUpdateEmitter) {
        this.productsRepository = productsRepository;
        this.queryUpdateEmitter = queryUpdateEmitter;
    }
    
    @EventHandler
    public void handle(ProductCreatedEvent event) throws Exception {
        ProductEntity entity = new ProductEntity();
        BeanUtils.copyProperties(event, entity);
        try {
            productsRepository.save(entity);
        } catch (IllegalArgumentException ex) {
            ex.printStackTrace();
        }
    }

    @EventHandler
    public void handle(ProductUpdatedEvent event) {
        ProductEntity entity = new ProductEntity();
        BeanUtils.copyProperties(event, entity);
        try {
            productsRepository.save(entity);
        } catch (IllegalArgumentException ex) {
            ex.printStackTrace();
        }
        
        ProductResponse productResponse = new ProductResponse();
        productResponse.setProductId(event.getProductId());
        productResponse.setStatus("Updated");

        //queryUpdateEmitter.emit(FetchProductQuery.class, query -> true, productResponse);
        queryUpdateEmitter.emit(FetchProductQuery.class, 
            query -> query.getProductId().equals(event.getProductId()), productResponse);
    }
}

控制器代码 #

@RestController
@RequestMapping("/product")
public class ProductController {

    private final CommandGateway commandGateway;
    private final QueryGateway queryGateway;
    private SubscriptionQueryResult<ProductResponse, ProductResponse> queryResult;
    
    @Autowired
    public ProductController(CommandGateway commandGateway, QueryGateway queryGateway) {
        this.queryGateway = queryGateway;
        this.commandGateway = commandGateway;
    }

    @PutMapping("/{id}")
    public ProductResponse updateProduct(@PathVariable("id") String id,
            @Valid @RequestBody ProductRequest request) {
            
        UpdateProductCommand updateProductCommand = UpdateProductCommand.builder().productId(id)
                .price(request.getPrice()).quantity(request.getQuantity()).title(request.getTitle()).build();
        commandGateway.sendAndWait(updateProductCommand);

        queryResult = queryGateway.subscriptionQuery(
                FetchProductQuery.builder().productId(id), ResponseTypes.instanceOf(ProductResponse.class),
                ResponseTypes.instanceOf(ProductResponse.class));
        try {
            // return queryResult.updates().blockFirst();
            return queryResult.initialResult().block();
        } catch (Exception e) {
            e.printStackTrace();
            throw e;
        } finally {
            queryResult.close();
        }
    }
}

查询处理器代码 #

@Component
public class ProductsQueryHandler {
    private final ProductRepository productsRepository;
    public ProductsQueryHandler(ProductRepository productsRepository) {
        this.productsRepository = productsRepository;
    }
    @QueryHandler
    public ProductResponse handler(FetchProductQuery query) {
        ProductEntity productEntity = productsRepository.findByProductId(query.getProductId());
        ProductResponse productResponse = new ProductResponse();
        BeanUtils.copyProperties(productEntity, productResponse);
        return productResponse;
    }
}

错误现象 #

1. Axon控制台无法查询到任何订阅查询记录
2. 运行时抛出如下错误：

2021-10-21 18:09:42.468 INFO 19608 --- [nio-8888-exec-2] o.a.m.interceptors.LoggingInterceptor   : Dispatched messages: [FetchProductQueryBuilder]
org.axonframework.axonserver.connector.AxonServerException: CANCELLED: AXONIQ-5000
    at org.axonframework.axonserver.connector.ErrorCode.lambda$static$24(ErrorCode.java:145)
    at org.axonframework.axonserver.connector.ErrorCode.convert(ErrorCode.java:182)
    at org.axonframework.axonserver.connector.ErrorCode.convert(ErrorCode.java:213)
    at org.axonframework.axonserver.connector.ErrorCode.convert(ErrorCode.java:202)
    at org.axonframework.axonserver.connector.event.util.GrpcExceptionParser.parse(GrpcExceptionParser.java:57)
    ...省略后续堆栈

问题原因 #

1. 核心错误：订阅查询参数未实例化 #

控制器代码中调用subscriptionQuery时，仅传入了FetchProductQuery.builder().productId(id)，没有调用.build()方法生成实际的FetchProductQuery实例，Axon Server无法识别Builder类型的查询对象，直接取消了请求，抛出AXONIQ-5000错误。

2. 版本兼容问题 #

Axon Framework 4.4.7 和 Axon Server 4.5.7存在跨小版本的兼容性问题，Axon Server 4.5.x的部分订阅查询逻辑在4.4.x版本的框架上无法正常适配。

3. 订阅查询执行顺序错误 #

当前代码先执行commandGateway.sendAndWait(updateProductCommand)等待命令执行完成，再创建订阅查询，若需要监听更新事件，此时订阅已经晚于事件触发时间，会丢失更新通知。

解决方案 #

1. 修复查询实例化错误 #

将控制器中订阅查询的代码修改为：

queryResult = queryGateway.subscriptionQuery(
        FetchProductQuery.builder().productId(id).build(), // 补充调用build()方法
        ResponseTypes.instanceOf(ProductResponse.class),
        ResponseTypes.instanceOf(ProductResponse.class));

2. 统一版本 #

将Axon Framework版本升级到4.5.x，和Axon Server版本保持一致，避免兼容性问题。

3. 调整执行顺序（若需要监听更新） #

如果业务需要获取更新后的通知，需要先创建订阅查询，再发送命令：

// 先创建订阅
queryResult = queryGateway.subscriptionQuery(
        FetchProductQuery.builder().productId(id).build(),
        ResponseTypes.instanceOf(ProductResponse.class),
        ResponseTypes.instanceOf(ProductResponse.class));
// 再发送命令
commandGateway.sendAndWait(updateProductCommand);
// 最后获取结果
try {
    return queryResult.updates().blockFirst();
} finally {
    queryResult.close();
}

内容的提问来源于stack exchange，提问作者VenkatN