I am creating a simple banking application using event sourcing and CQRS. The application uses postgres and kafka as an event store. The source code can be found at source. Currently, the application supports the following functionality (master branch):
- Create a new user (full name, email, pan): iam module
- Update a user (update full name, update email, update pan): iam module
- View a user & their entire update history: user-history module
Reasons for using postgres in combination with kafka as an event store: Kafka does not provide optimistic concurrency. Each event is first stored in postgres and then published to a kafka topic. Each event has an aggregate id and a revision number among other things. Whenever an update happens on an aggregate, it's revision number is incremented. The aggregate id and revision number form a unique key to provide optimistic concurrency. The key used to publish a message to kafka is the aggregate id which ensures that messages for the same aggregate will always be published to the same partition providing ordering guarantee for consumers. Is this approach correct?
Another problem that I am facing is versioning events: Let's assume the following example: Initially, creating and updating a user required a fullName, but as per new requirements, we now want to capture the first and the last name separately. After this change (implemented in the improvement/split-fullname branch), my system will have both new and old events. How do you de-serialize multiple different versions of an event cleanly. Currently, I am keeping all the event classes in the code base and using the chain of responsibility pattern as below:
Current event classes
/*
Concept of an Event - Anything that happens in the domain
*/
public interface Event<T> {
/*
Event version - Not to be confused with revision. Used mainly for deserialization.
*/
int getVersion();
/*
Which Aggregate this event happened to (users, orders, ...)
*/
String getAggregateType();
/*
Id of the aggregate on which this event happened (order-12345, user-12345, ...)
*/
AggregateId getAggregateId();
/*
Revision no of this aggregate. Mainly used for optimistic concurrency/locking.
*/
int getRevision();
/*
Type of the event (OrderCreated, UserCreated, ...)
*/
String getEventType();
void applyOn(T aggregate);
}
@Getter
@Setter(AccessLevel.PRIVATE)
public abstract class UserEvent implements Event<User> {
public static final String TYPE = "users";
private UserId aggregateId;
private int revision;
private int version = 1;
private String eventType = UserEvent.TYPE;
UserEvent() {}
UserEvent(UserId aggregateId, int revision) {
this.aggregateId = aggregateId;
this.revision = revision;
}
@Override
public String getAggregateType() {
return UserEvent.TYPE;
}
}
@Getter
@Setter(AccessLevel.PRIVATE)
public class UserCreated extends UserEvent {
private String firstName;
private String lastName;
private String email;
private String pan;
private String eventType = "UserCreated";
private UserCreated() {}
public UserCreated(UserId aggregateId, int revision,
String firstName, String lastName, String email, String pan) {
super(aggregateId, revision);
this.firstName = firstName;
this.lastName = lastName;
this.email = email;
this.pan = pan;
}
@Override
public void applyOn(User user) {
user.apply(this);
}
@Override
public int getVersion() {
return 2;
}
}
@Getter
@Setter(AccessLevel.PRIVATE)
public class UserNameUpdated extends UserEvent {
private String newFirstName;
private String newLastName;
private String eventType = "UserNameUpdated";
private UserNameUpdated() {}
public UserNameUpdated(UserId userId, int revision, String newFirstName, String newLastName) {
super(userId, revision);
this.newFirstName = newFirstName;
this.newLastName = newLastName;
}
@Override
public void applyOn(User user) {
user.apply(this);
}
@Override
public int getVersion() {
return 2;
}
}
Old event classes
@Getter
@Setter(AccessLevel.PRIVATE)
public class UserCreatedV1 extends UserEvent {
private String fullName;
private String email;
private String pan;
private String eventType = "UserCreated";
private UserCreatedV1() {}
public UserCreatedV1(UserId aggregateId, int revision, String fullName, String email, String pan) {
super(aggregateId, revision);
this.fullName = fullName;
this.email = email;
this.pan = pan;
}
@Override
public void applyOn(User user) {
user.apply(this);
}
@Override
public int getVersion() {
return 1;
}
}
@Getter
@Setter(AccessLevel.PRIVATE)
public class UserNameUpdatedV1 extends UserEvent {
private String newFullName;
private String eventType = "UserNameUpdated";
private UserNameUpdatedV1() {}
public UserNameUpdatedV1(UserId userId, int revision, String newFullName) {
super(userId, revision);
this.newFullName = newFullName;
}
@Override
public void applyOn(User user) {
user.apply(this);
}
@Override
public int getVersion() {
return 1;
}
}
Handler chains
public interface UserEventParser {
void setNext(UserEventParser userEventParser);
UserEvent parse(Fact fact, ObjectMapper mapper);
}
@Slf4j
public class UserEventParserV1 implements UserEventParser {
private UserEventParser next;
@Override
public void setNext(UserEventParser userEventParser) {
this.next = userEventParser;
}
@Override
public UserEvent parse(Fact fact, ObjectMapper mapper) {
String type = fact.getFactType();
if (fact.getVersion() == 1 && ("UserCreated".equals(type)
|| "UserNameUpdated".equals(type))) {
UserEvent userEvent = null;
try {
if ("UserCreated".equals(type)){
userEvent = mapper.readValue(fact.getFact(), UserCreatedV1.class);
} else {
userEvent = mapper.readValue(fact.getFact(), UserNameUpdatedV1.class);
}
} catch (Exception e) {
log.error("Exception: Cannot parse fact -> event", e);
}
return userEvent;
}
return next.parse(fact, mapper);
}
}
@Slf4j
public class DefaultUserEventParser implements UserEventParser {
private UserEventParser next;
@Override
public void setNext(UserEventParser userEventParser) {
}
@Override
public UserEvent parse(Fact fact, ObjectMapper mapper) {
UserEvent userEvent = null;
try {
String type = fact.getFactType();
if ("UserCreated".equals(type)){
userEvent = mapper.readValue(fact.getFact(), UserCreated.class);
} else if ("UserNameUpdated".equals(type)) {
userEvent = mapper.readValue(fact.getFact(), UserNameUpdated.class);
} else if ("UserEmailUpdated".equals(type)) {
userEvent = mapper.readValue(fact.getFact(), UserEmailUpdated.class);
} else if ("UserPanUpdated".equals(type)) {
userEvent = mapper.readValue(fact.getFact(), UserPanUpdated.class);
}
} catch (Exception e) {
log.error("Exception: Cannot parse fact -> event", e);
}
return userEvent;
}
}
User aggregate
/*
The User Aggregate
*/
public class User {
private UserId userId;
private String firstName;
private String lastName;
private String email;
private String pan;
private int revision;
private List<UserEvent> changes = new ArrayList<>();
public UserId getUserId() { return this.userId; }
public String getFirstName() { return this.firstName; }
public String getLastName() { return this.lastName; }
public String getEmail() {
return this.email;
}
public String getPan() { return this.pan; }
public List<UserEvent> getChanges() { return this.changes; }
public UserId create(String firstName, String lastName, String email, String pan) {
this.userId = UserId.generate();
this.firstName = Objects.requireNonNull(firstName);
this.lastName = Objects.requireNonNull(lastName);
this.email = email;
this.pan = Objects.requireNonNull(pan);
this.revision = 1;
this.changes.add(
new UserCreated(this.userId, this.revision, this.firstName, this.lastName, this.email, this.pan)
);
return this.userId;
}
public void updateFullName(String newFirstName, String newLastName) {
this.firstName = newFirstName;
this.lastName = newLastName;
++this.revision;
this.changes.add(
new UserNameUpdated(this.userId, this.revision, this.firstName, this.lastName)
);
}
public void updateEmail(String newEmail) {
this.email = newEmail;
++this.revision;
this.changes.add(
new UserEmailUpdated(this.userId, this.revision, this.email)
);
}
public void updatePan(String newPan) {
this.pan = newPan;
++this.revision;
this.changes.add(
new UserPanUpdated(this.userId, this.revision, this.pan)
);
}
public void apply(UserCreated userCreated) {
this.userId = userCreated.getAggregateId();
this.firstName = userCreated.getFirstName();
this.lastName = userCreated.getLastName();
this.email = userCreated.getEmail();
this.pan = userCreated.getPan();
this.revision = userCreated.getRevision();
}
public void apply(UserNameUpdated userNameUpdated) {
this.firstName = userNameUpdated.getNewFirstName();
this.lastName = userNameUpdated.getNewLastName();
this.revision = userNameUpdated.getRevision();
}
public void apply(UserEmailUpdated userEmailUpdated) {
this.email = userEmailUpdated.getNewEmail();
this.revision = userEmailUpdated.getRevision();
}
public void apply(UserPanUpdated userPanUpdated) {
this.pan = userPanUpdated.getNewPan();
this.revision = userPanUpdated.getRevision();
}
public void apply(UserCreatedV1 userCreated) {
this.userId = userCreated.getAggregateId();
String fullName = userCreated.getFullName();
String[] name = fullName.split(" ");
this.firstName = name[0];
this.lastName = name.length > 1 ? name[1] : null;
this.email = userCreated.getEmail();
this.pan = userCreated.getPan();
this.revision = userCreated.getRevision();
}
public void apply(UserNameUpdatedV1 userNameUpdated) {
String newFullName = userNameUpdated.getNewFullName();
String[] name = newFullName.split(" ");
this.firstName = name[0];
this.lastName = name.length > 1 ? name[1] : null;
this.revision = userNameUpdated.getRevision();
}
}
The above solution works but I feel as an application evolves, this solution will end up with a lot of classes for older events even if I keep my aggregates small. Is there a better alternative to this problem?
Also please look at the entire source code on github (master and improvement/split-fullname branches) and provide any necessary feedback.
