如何在Kogito流程实例上下文中获取CloudEvents扩展字段？

问题描述 #

我基于Kogito实现项目，使用kogito-addons-quarkus-messaging + Kafka + CloudEvents组件。向服务入站通道发送消息时，携带了自定义CloudEvents扩展字段ce_test: rest，流程实例正常启动，负载和业务键都能正常获取，但在服务任务中通过KogitoProcessContext无法找到该扩展字段。

发送的消息头部：

ce_id: 6dfde9b9-886e-4674-9936-1c6e2bfcd5dd, ce_kogitobusinesskey: asdasd, ce_source: https://github.com/cloudevents/sdk-java/tree/main/examples/kafka, ce_specversion: 1.0, ce_test: rest, ce_type: GetAccountList

负载内容：

{
   "a": "b"
}

服务任务中尝试获取字段的代码：

public void showAll (KogitoProcessContext context) {
    System.out.println("headers: " + context.getHeaders());
    System.out.println("contextData: " + context.getContextData());
    if (!context.getContextData().isEmpty()) {
        context.getContextData().forEach((s, o) -> {
            System.out.println("CD name: " + s + " val: " + o);
        });
    }
    System.out.println("vars: " + context.getProcessInstance().getVariables());
    System.out.println("process meta: " + context.getProcessInstance().getProcess().getMetaData());
    System.out.println("node meta: " + context.getNodeInstance().getNode().getMetaData());
    System.out.println("PI_headers: " + context.getProcessInstance().getHeaders());
    System.out.println("events: " + Arrays.toString(context.getProcessInstance().getEventTypes()));
}

请问是否可以在BPMN中与ce_test扩展字段交互，还是该字段会被忽略并未存储？

解决方案 #

默认情况下，Kogito只会自动处理部分预定义的CloudEvents字段（比如ce_kogitobusinesskey），自定义扩展字段不会自动存入流程上下文，需要手动配置映射才能获取：

• 通过BPMN消息事件配置头部映射
  在流程的消息启动事件（或消息捕获事件）的属性中，添加Header Mapping规则：

  • 来源头部填写ce_test
  • 目标流程变量填写自定义变量名（比如testExtension）
    配置完成后，流程启动时会自动将ce_test的值存入指定变量，在服务任务中可通过context.getProcessInstance().getVariables().get("testExtension")获取。

• 通过代码拦截器处理扩展字段
  自定义消息拦截器，在消息接收阶段提取CloudEvents的扩展字段，将其存入流程实例的上下文数据：

  public class CloudEventExtensionInterceptor implements MessageInterceptor {
      @Override
      public Message intercept(Message message) {
          CloudEvent cloudEvent = message.getPayload(CloudEvent.class);
          if (cloudEvent != null && cloudEvent.getExtension("test") != null) {
              message.addContextData("ce_test", cloudEvent.getExtension("test"));
          }
          return message;
      }
  }
  

  注册拦截器后，在服务任务中可通过context.getContextData().get("ce_test")获取对应值。

• 检查消息通道配置
  确保Kafka消息通道的配置允许传递所有头部字段，避免自定义扩展字段被过滤。

关键说明 #

自定义CloudEvents扩展字段不会被Kogito自动忽略，但需要显式配置映射才能在流程上下文中访问。通过上述任一方法，都能实现与ce_test这类扩展字段的交互。

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