이벤트로 시스템 잇기 (Kafka)

uEngine6 BPM과 다른 시스템을 Kafka 이벤트로 잇는 전체 구조를 설명합니다. 토픽·헤더·payload 계약, 배포 때 등록되는 이벤트 매핑, BPM이 내보내는 진행 알림, 실패와 재시도, 그리고 계정계·ERP·일반 REST 시스템의 세 가지 연동 방식을 비교합니다.

무엇을 · 왜#

업무 시스템(계정계, ERP, 구매 포털)은 거래를 기록하고 규칙을 지키는 기록 시스템입니다. 누가 결재하고 다음 일을 누가 맡으며, 막히면 어디로 되돌릴지는 BPM이 맡습니다. 둘을 REST 호출로 연결하면 서로의 주소·인증·장애에 영향을 받습니다. 이벤트로 이으면 각자 자기 일만 하고, 상대는 정해진 이름의 메시지만 압니다.

Camunda의 기능과 비교하면 다음과 같습니다.

                   ┌──────────── bpm-topic (헤더 type · corrKey · emnb, 본문 JSON) ────────────┐
업무 시스템 ───────┤                                                                          ▼
 (계정계·ERP·포털)  │                                                   process-service(Kafka 모드, 그룹 bpm)
        ▲          │                                                     │ BPM_EVENT_MAPPING 으로 정의·활동 찾기
        │          │                                                     │ → 인스턴스 시작 / 기다리던 작업 완료 / payload → 변수
        └──────────┴──── bpm-brodcast (type=TASK_COMPLETED, 본문 instanceId·tracingTag·activityName) ◀┘

process-service의 메시징은 두 방식 가운데 하나입니다(uengine.messaging.mode).

따라 하기#

예제 튜토리얼/카프카/견적연동은 이벤트로 시작해(TutQuoteRequested), 윤팀장이 「견적 검토」를 하고, 공급사의 회신 이벤트(TutQuoteReplied)가 최구매의 「공급사 회신 대기」를 끝내는 프로세스입니다. 이벤트 콘솔(18장) 한 화면에서 주고받는 메시지를 모두 확인합니다.

1. 매핑 확인. 정의를 배포하면 BPM_EVENT_MAPPING에 두 줄이 생깁니다. 콘솔을 BPM_DB_URL·MAPPING_PREFIX=튜토리얼/카프카/로 띄우면 왼쪽 아래에 보입니다. | event_name | definition_id | tracing_tag | correlation_key | 시작 | |---|---|---|---|---| | TutQuoteRequested | 튜토리얼/카프카/견적연동.bpmn | Start | requestNo | 예 | | TutQuoteReplied | 튜토리얼/카프카/견적연동.bpmn | WaitReply | requestNo | |

   

2. 정의 보기. 시작 이벤트(메시지) → 「견적 검토」(윤팀장, 사람) → 「공급사 회신 대기」(최구매, 이벤트로 끝남) → 끝.

   

3. 들어가는 이벤트. 콘솔에서 type = TutQuoteRequested, payload {"requestNo":"Q-2611-0301","item":"노트북","qty":3}를 보냅니다. 인스턴스 견적_Q-2611-0301이 시작되고 윤팀장에게 「견적 검토」가 갑니다.
4. 나오는 알림. 윤팀장이 업무 목록에서 「견적 검토」를 완료하면, 콘솔에 ← BPM 이 알림 · bpm-brodcast 줄이 생깁니다.

   {"instanceId": "343", "activityName": "견적 검토", "eventType": "TASK_COMPLETED", "taskIds": ["556"], "tracingTag": "Review"}
   

   

5. 행위자가 다르면. type = TutQuoteReplied, 헤더 emnb = kim, payload {"requestNo":"Q-2611-0301","unitPrice":1450000}을 보냅니다. 「공급사 회신 대기」의 담당자는 최구매이므로 엔진이 거절하고, 작업은 그대로 남고 알림도 없습니다.
6. 행위자가 맞으면. 같은 이벤트를 emnb = choi로 보냅니다. 「공급사 회신 대기」가 끝나고 TASK_COMPLETED 알림이 나오며, 인스턴스가 완료됩니다. 관리자 화면의 변수에 unitPrice 1450000이 들어와 있습니다.

   

   

설정 레퍼런스#

토픽#

process-service 쪽 설정(계정계 예제 config/process-service.kafka.yml과 같습니다):

uengine:
  messaging:
    mode: kafka
spring:
  cloud:
    function: { definition: bpm }
    stream:
      kafka.binder.brokers: ${KAFKA_BROKERS:localhost:9092}
      bindings:
        bpm-in-0:     { group: bpm, destination: bpm-topic, content-type: application/json }
        bpm-out:      { destination: bpm-topic, content-type: application/json }
        bpm-brodcast: { destination: bpm-brodcast, content-type: application/json }

헤더 계약 (들어오는 이벤트)#

payload 계약#

• JSON 객체 하나. 모든 최상위 필드가 같은 이름의 프로세스 변수가 됩니다(시작 이벤트는 시작 변수로, 작업 완료는 인스턴스 변수에 덮어씀).
• 상관 키 필드는 이름으로 찾습니다(순서는 상관없습니다).
• 게이트웨이에서 숫자로 비교할 값은 숫자(정수)로 보냅니다. 엔진은 조건 값의 타입으로 비교합니다.
• 필드 이름을 변수 이름과 같게 지으면 설정이 쉽습니다. 다르면 시작 이벤트·작업의 「데이터 매핑」으로 옮깁니다.

이벤트 매핑(BPM_EVENT_MAPPING)#

배포(POST /definition-changes, 화면 저장도 같음) 때 EventMappingDeployFilter가 등록합니다.

correlation_key는 아규먼트 가운데 「관계 키」(isCorrKey)의 이름입니다. 행은 (이벤트·정의·활동·시작 여부) 단위로 덮어쓰고, 지우지는 않습니다. 이벤트 이름을 바꿨다면 옛 행을 직접 지우세요.

select event_name, definition_id, tracing_tag, correlation_key, is_start_event
  from bpm_event_mapping order by definition_id, id;

진행 알림 구독#

bpm-brodcast를 자기 그룹 이름으로 구독합니다. 본문에는 인스턴스 번호만 있고 상관 키는 없으므로, 업무 문서와 짝지으려면 인스턴스 변수를 한 번 읽어 둡니다(GET /instance/{id}/variables).

import { Kafka } from 'kafkajs'
const consumer = new Kafka({ clientId: 'purchase-portal', brokers: ['localhost:9092'] }).consumer({ groupId: 'purchase-portal' })
await consumer.connect()
await consumer.subscribe({ topics: ['bpm-brodcast'] })
await consumer.run({ eachMessage: async ({ message }) => {
  if (message.headers.type?.toString() !== 'TASK_COMPLETED') return
  const { instanceId, tracingTag, activityName } = JSON.parse(message.value.toString())
  console.log(`인스턴스 ${instanceId}: ${activityName}(${tracingTag}) 끝`)
} })

실패와 재시도 (Kafka 모드)#

같은 이벤트가 두 번 와도 안전한 경우: 시작은 같은 상관 키의 실행 중 인스턴스가 있으면 건너뛰고, 작업 완료는 이미 끝난 작업에는 다시 일어나지 않습니다. 실패한 메시지를 다시 보내는 일(재처리)은 보내는 쪽에서 맡습니다. ERP 예제는 아웃박스 행에 attempts·last_error를 남기고 그 행부터 다시 보냅니다.

세 가지 연동 방식#

일반 REST 시스템의 콜백 예: 「신용 평가 요청」 서비스 작업이 외부 평가사에 POST /evaluations(본문에 요청번호)를 보내고, 다음 작업 「평가 결과 대기」가 eventSynchronization(TutEvaluationDone, 상관 키 requestNo)으로 기다립니다. 평가사가 끝나면 어댑터가 TutEvaluationDone을 bpm-topic으로 보냅니다.

알려진 제약#

• Kafka 모드에서는 폴링 받은함(POST /inbox)과 SSE(/events/stream)가 없습니다(404). 화면의 실시간 갱신도 없으므로 업무 목록은 새로 고쳐 봅니다.
• TASK_COMPLETED 외의 알림(작업 배정 TASK_ASSIGNED 등)은 Kafka로 나가지 않습니다. 외부 작업자는 업무 목록(/worklist/search/findToDo)을 읽어 할 일을 찾습니다.
• 큐잉을 켠 활동(queuingEnabled)은 Kafka 모드에서 멈춥니다(ActivityQueued를 type으로 보내고 _type으로 읽음). 시스템 작업은 외부 작업 방식으로 만드세요.
• 알림 본문에 상관 키가 없습니다. 짝짓기는 인스턴스 변수 조회로 합니다.
• require-event-actor 스위치는 시연 스택의 엔진 패치입니다. 원래 브랜치에서 사람 작업을 이벤트로 끝내려면 emnb와 선점이 필요합니다.
• 이벤트로 시작한 인스턴스의 「시작자」는 process-service가 마지막으로 받은 요청의 사용자(정적 필드)에서 옵니다. 다른 클라이언트가 동시에 API를 부르면 엉뚱한 사람이 시작자가 될 수 있으므로, 이벤트로 시작하는 정의의 레인은 그룹·역할이나 사람으로 정해 두세요.

확인 문제#