Make sure JPA and JDBC Event Storage Engines respect Event Message type

When appending an Event Message that does not implement DomainEventMessage, reading such message out of the event store should not turn it into a DomainEventMessage. Fixes #1697
AxonFramework · Feb 8, 2021 · c61a95b · c61a95b
1 parent cfb1c41
commit c61a95b
Show file tree

Hide file tree

Showing 3 changed files with 18 additions and 3 deletions.
diff --git a/...src/main/java/org/axonframework/eventsourcing/eventstore/jdbc/JdbcEventStorageEngine.java b/...src/main/java/org/axonframework/eventsourcing/eventstore/jdbc/JdbcEventStorageEngine.java
@@ -636,11 +636,13 @@ protected TrackedEventData<?> getTrackedEventData(ResultSet resultSet,
                                                       GapAwareTrackingToken previousToken) throws SQLException {
         long globalSequence = resultSet.getLong(schema.globalIndexColumn());
 
+        String aggregateIdentifier = resultSet.getString(schema.aggregateIdentifierColumn());
+        String eventIdentifier = resultSet.getString(schema.eventIdentifierColumn());
         GenericDomainEventEntry<?> domainEvent = new GenericDomainEventEntry<>(
                 resultSet.getString(schema.typeColumn()),
-                resultSet.getString(schema.aggregateIdentifierColumn()),
+                eventIdentifier.equals(aggregateIdentifier) ? null : aggregateIdentifier,
                 resultSet.getLong(schema.sequenceNumberColumn()),
-                resultSet.getString(schema.eventIdentifierColumn()),
+                eventIdentifier,
                 readTimeStamp(resultSet, schema.timestampColumn()),
                 resultSet.getString(schema.payloadTypeColumn()),
                 resultSet.getString(schema.payloadRevisionColumn()),

diff --git a/...g/src/main/java/org/axonframework/eventsourcing/eventstore/jpa/JpaEventStorageEngine.java b/...g/src/main/java/org/axonframework/eventsourcing/eventstore/jpa/JpaEventStorageEngine.java
@@ -169,8 +169,11 @@ protected List<? extends TrackedEventData<?>> fetchTrackedEvents(TrackingToken l
         GapAwareTrackingToken token = previousToken;
         for (Object[] entry : entries) {
             long globalSequence = (Long) entry[0];
+            String aggregateIdentifier = (String) entry[2];
+            String eventIdentifier = (String) entry[4];
             GenericDomainEventEntry<?> domainEvent = new GenericDomainEventEntry<>(
-                    (String) entry[1], (String) entry[2], (long) entry[3], (String) entry[4], entry[5],
+                    (String) entry[1], eventIdentifier.equals(aggregateIdentifier) ? null : aggregateIdentifier,
+                    (long) entry[3], eventIdentifier, entry[5],
                     (String) entry[6], (String) entry[7], entry[8], entry[9]
             );
 

diff --git a/...cing/src/test/java/org/axonframework/eventsourcing/eventstore/EventStorageEngineTest.java b/...cing/src/test/java/org/axonframework/eventsourcing/eventstore/EventStorageEngineTest.java
@@ -62,6 +62,16 @@ protected void testStoreAndLoadEvents() {
         assertEquals(1, testSubject.readEvents("otherAggregate").asStream().count());
     }
 
+    @Test
+    void testAppendAndReadNonDomainEvent() {
+        testSubject.appendEvents(new GenericEventMessage<>("Hello world"));
+
+        List<? extends TrackedEventMessage<?>> actual = testSubject.readEvents(null, false)
+                                                                   .collect(toList());
+        assertEquals(1, actual.size());
+        assertFalse(actual.get(0) instanceof DomainEventMessage);
+    }
+
     @Test
     void testStoreAndLoadEventsArray() {
         testSubject.appendEvents(createEvent(0), createEvent(1));