문제 정의
Lakey 프로젝트에서는 사내 메시징 프레임워크인 Prologue의 Vello모듈을 사용하여 외부 이벤트를 저장하고 이를 LLM 컨텍스트에 사용할 수 있도록 외부 이벤트를 관리하고 있습니다.
Prologue가 기본으로 제공하는 이벤트 변환기인 'TypeAwareEventConverter'는 메시지를 수신하면 헤더의 'payloadClass' 값을 읽고, 클래스 이름이 구동중인 서비스의 클래스패스에 존재하는지 확인 후에 역직렬화를 수행합니다.
이 방식은 동일 서비스 내부에서나, 외부 서비스의 Event 모듈을 의존한 상태에서 이벤트를 주고받을 때는 문제가 없습니다. 이미 Event 클래스가 서비스의 클래스패스에 존재하기 때문입니다. 하지만 정의되지 않고 발행된 이벤트는 상황이 다릅니다. 정의되지 않은 이벤트의 클래스는 Lakey의 클래스패스에 없기 때문에 예외를 발생시키며, 결국 해당 이벤트는 정상 처리되지 못한 채 소실됩니다.
외부 서비스의 이벤트 스키마가 바뀔 때마다 Lakey 내부 로직도 변경되기 때문에 서비스 간 강한 결합이 생기게 됩니다. 외부 서비스의 Event모듈에 의존하지 않고 이벤트를 저장하는 방법을 고민했으며, 이 문제를 해결하기 위해 'LakeyEventConverter'를 직접 구현했습니다.
기존 EventConverter의 동작 방식
1. 메시지 페이로드를 지정된 인코딩으로 문자열로 변환합니다.
2. 헤더의 'payloadClass' 값이 비어 있으면 'Map' 타입으로 역직렬화해 반환합니다.
3. 'payloadClass' 값이 존재하면 'Class.forName()'으로 클래스를 로드한 뒤, 해당 타입으로 역직렬화합니다.
4. 클래스 로딩에 실패하면 최종적으로 'EventConversionFailed' 예외가 발생합니다.
해결 방법: LakeyEventConverter
4번 문제를 해결하기 위해 'TypeAwareEventConverter'를 상속한 뒤 'convert()' 메서드를 오버라이드했습니다. 타입 변환에 실패한 이벤트를 버리지 않고, 해석 가능한 원시 형태로 보존하는 방법입니다.
public class LakeyEventConverter extends TypeAwareEventConverter {
private final String encoding;
public LakeyEventConverter(
@Value("${vizend.prologue.vello.encoding:UTF-8}") String encoding) {
super(encoding);
this.encoding = encoding;
}
@Override
public Object convert(VelloMessage message) {
try {
Object object = super.convert(message);
if (object instanceof LakeyOutgoingDomainEvent out) {
return out.toIncomingEvent();
}
return object;
} catch (EventConvertException e) {
String payloadJson = new String(message.getPayload(), Charset.forName(encoding));
Map<String, Object> payload = JsonUtil.fromJson(payloadJson, Map.class);
String fqcn = message.getPayloadClass();
String simpleClassName = fqcn.substring(fqcn.lastIndexOf('.') + 1);
return UnreferencedEvent.builder()
.eventType(simpleClassName)
.headers(message.getHeaders())
.payload(payload)
.messageId(message.getId())
.subject(message.getSubject())
.receivedAt(LocalDateTime.now())
.build();
}
}
}'LakeyEventConverter'의 처리 흐름은 두 가지 경우로 나뉩니다.
1. 일반적인 변환 성공인 경우
부모 컨버터('TypeAwareEventConverter')의 변환이 성공했지만 Lakey 자체 이벤트나 의존하고 있는 이벤트 클래스인 경우에는, 변환된 객체를 그대로 반환합니다.이 경우에는 별도의 후처리가 필요하지 않습니다.
2. 외부 이벤트처럼 타입 변환에 실패한 경우
부모 클래스가 'EventConvertException'을 던지면 이를 잡아 별도 처리합니다. 이 경우 페이로드를 'Map<String, Object>'로 역직렬화하고, 헤더에 담긴 FQCN에서 단순 클래스 이름만 추출해 'eventType'으로 사용합니다. 이후 이를 기반으로 'UnreferencedEvent' 객체를 생성해 반환합니다.
이때 단순히 페이로드만 남기는 것이 아니라, 아래 정보도 함께 보존합니다.
* 메시지 ID, 수신 토픽(subject), 메시지 헤더, 수신 시각, 원본 페이로드
UnreferencedEvent
UnreferencedEvent는 참조할 클래스가 없는 이벤트를 담는 객체입니다. 이름 그대로 Lakey가 타입을 알 수는 없지만, 수집할 가치는 있는 이벤트를 표현합니다.
@Getter
@Builder
public class UnreferencedEvent {
private String eventType;
private Map<String, String> headers;
private Map<String, Object> payload;
private LocalDateTime receivedAt;
private String messageId;
private String subject;
}여기서 중요한 값은 페이로드만이 아닙니다. messageId는 중복 처리를 위해 필요하고, subject는 이벤트가 들어온 토픽을 확인하는 데 사용됩니다. headers와 receivedAt은 나중에 이벤트 출처와 수신 시점을 추적하는 데 도움이 됩니다.
UnreferencedEvent에서 RawEvent까지
LakeyEventConverter가 UnreferencedEvent를 반환하면, Prologue의 @EventHandler가 붙은 UnreferencedEventHandler가 이 이벤트를 받습니다. 핸들러는 별다른 비즈니스 판단을 하지 않고 RawEventFlow.register()로 넘깁니다.
@Component
@RequiredArgsConstructor
public class UnreferencedEventHandler {
private final RawEventFlow rawEventFlow;
@EventHandler
public void handleUnreferencedEvent(UnreferencedEvent unreferencedEvent) {
if (unreferencedEvent == null) {
log.warn("UnreferencedEvent is null");
return;
}
rawEventFlow.register(unreferencedEvent);
}
}RawEventFlow.register()에서는 세 가지를 확인합니다.
public void register(UnreferencedEvent unreferencedEvent) {
if (!rawEventAction.isTargetEvent(unreferencedEvent)) {
return;
}
RawEventCdo rawEventCdo = rawEventAction.convertToRawEventCdo(unreferencedEvent);
if (rawEventLogic.existsByEventId(rawEventCdo.getEventId())) {
return;
}
String rawEventId = rawEventLogic.registerRawEvent(rawEventCdo);
eventProxy.produceEvent(RawEventProjectionRequestedEvent.newInstance(rawEventId));
}1. 대상 토픽인지 확인
외부에서 들어오는 모든 이벤트를 저장하지는 않습니다. LakeyConfigProperties에 설정된 targetTopics에 포함된 토픽만 처리합니다. 수집 대상이 아닌 이벤트를 초기에 걸러 불필요한 저장과 후속 처리를 줄입니다.
2. 중복 이벤트인지 확인
messageId를 eventId로 사용해 이미 저장된 이벤트인지 확인합니다. 같은 이벤트가 다시 들어와도 한 번만 저장되도록 합니다.
3. RawEvent로 저장
대상 토픽이고 중복도 아니라면 UnreferencedEvent를 RawEventCdo로 변환해 저장합니다. 저장 이후에는 RawEventProjectionRequestedEvent를 발행합니다. 실제 데이터 전처리는 이 이벤트를 처리하는 핸들러 단계에서 진행됩니다.
마무리
이 구조를 만든 이유는 명확합니다. 이벤트 타입을 미리 알 수 없더라도, 들어온 이벤트 자체는 잃지 않기 위해서입니다.
기존의 'TypeAwareEventConverter'는 클래스패스에 없는 타입을 만나면 예외를 던지고 처리를 중단했습니다. 반면 'LakeyEventConverter'는 예외 처리된 이벤트를 'UnreferencedEvent'라는 Common Entity에 담아 보존합니다. 그 결과 Lakey는 외부 서비스의 이벤트 클래스를 직접 의존하지 않아도 이벤트를 수집할 수 있게 되었습니다.
이벤트 처리를 할 때 서비스 간 결합도를 느슨하게 할 필요가 있다면 참고해주시면 감사하겠습니다.
Ted