Problem Definition
In the Lakey project, we manage external events using the Vello module of the in-house messaging framework Prologue to store these events and utilize them in the LLM context.
The event converter provided by Prologue, 'TypeAwareEventConverter', reads the 'payloadClass' value from the header upon receiving a message and performs deserialization after checking if the class name exists in the classpath of the running service.
This approach works without issues when exchanging events within the same service or when dependent on the Event module of an external service, as the Event class already exists in the classpath of the service. However, it is a different situation with events that are published but not defined. The class of an undefined event does not exist in the Lakey classpath, leading to exceptions, and ultimately these events are lost without being processed correctly.
As the event schema of the external service changes, the internal logic of Lakey also needs to change, creating a strong coupling between services. We contemplated methods to store events without relying on the Event module of external services, and to solve this problem, we implemented 'LakeyEventConverter' ourselves.
Behavior of the existing EventConverter
1. Converts the message payload into a string with the specified encoding.
2. If the 'payloadClass' value in the header is empty, it deserializes and returns it as a 'Map' type.
3. If the 'payloadClass' value exists, it loads the class with 'Class.forName()' and then deserializes to that type.
4. If class loading fails, the 'EventConversionFailed' exception is ultimately thrown.
Solution: LakeyEventConverter
To address issue number four, we inherited from 'TypeAwareEventConverter' and overridden the 'convert()' method. This method preserves events that fail in type conversion as an interpretable raw form.
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();
}
}
}The processing flow of 'LakeyEventConverter' is divided into two cases.
1. In the case of a successful general conversion
If the conversion by the parent converter ('TypeAwareEventConverter') is successful but is for Lakey's own event or an event class it depends on, the converted object is returned as is. In this case, no further processing is required.
2. If type conversion fails like an external event
If the parent class throws 'EventConvertException', it will be caught and handled separately. In this case, the payload is deserialized into 'Map<String, Object>', and the simple class name is extracted from the FQCN in the headers to be used as 'eventType'. Based on this, an 'UnreferencedEvent' object is created and returned.
At this time, it is not just the payload that remains, but the following information is also preserved.
* Message ID, received topic (subject), message header, received timestamp, original payload
UnreferencedEvent
UnreferencedEvent is an object that contains events that have no reference class. As the name suggests, Lakey cannot know the type, but it represents events that are worth collecting.
@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;
}Here, the important values are not just the payload. The messageId is needed for deduplication, and the subject is used to check the topic from which the event arrived. The headers and receivedAt help track the source of the event and the time it was received later.
From UnreferencedEvent to RawEvent
When LakeyEventConverter returns an UnreferencedEvent, the UnreferencedEventHandler annotated with @EventHandler in Prologue receives this event. The handler does not make any particular business judgment and simply passes it to 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() checks three things.
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. Check if it is the target topic
Not all events coming from outside are stored. Only topics included in the targetTopics set in LakeyConfigProperties are processed. This filters out events that are not intended for collection, reducing unnecessary storage and follow-up processing.
2. Check if it is a duplicate event
The messageId is used as the eventId to check if the event has already been saved. The same event will only be saved once even if it comes in again.
3. Save as RawEvent
If it is the target topic and not a duplicate, we transform the UnreferencedEvent into a RawEventCdo and save it. After saving, we publish a RawEventProjectionRequestedEvent. The actual data preprocessing takes place in the handler stage that processes this event.
Completion
The reason for creating this structure is clear. It is to ensure that the incoming event itself is not lost, even if we cannot know the event type in advance.
The existing 'TypeAwareEventConverter' threw an exception and stopped processing when encountering a type that was not present in the classpath. In contrast, the 'LakeyEventConverter' preserves exceptions by containing the event in a Common Entity called 'UnreferencedEvent'. As a result, Lakey is able to collect events without directly depending on the external service's event classes.
If you need to loosen the coupling between services during event processing, I would appreciate your consideration.
Ted