FINERACT-1694: Storing event schema along with external events

2022-09-07 14:45:51 +02:00
parent 72e78df0fc
commit 90896a6467
26 changed files with 198 additions and 30 deletions
@@ -42,6 +42,9 @@ public class ExternalEvent extends AbstractPersistableCustom {
    @Column(name = "type", nullable = false)
    private String type;

+    @Column(name = "schema", nullable = false)
+    private String schema;
+
    @Basic(fetch = FetchType.LAZY)
    @Column(name = "data", nullable = false)
    private byte[] data;
@@ -64,8 +67,9 @@ public class ExternalEvent extends AbstractPersistableCustom {
    @Column(name = "business_date", nullable = false)
    private LocalDate businessDate;

-    public ExternalEvent(String type, byte[] data, String idempotencyKey) {
+    public ExternalEvent(String type, String schema, byte[] data, String idempotencyKey) {
        this.type = type;
+        this.schema = schema;
        this.data = data;
        this.idempotencyKey = idempotencyKey;
        this.createdAt = DateUtils.getOffsetDateTimeOfTenant();
@@ -25,6 +25,7 @@ import org.apache.fineract.infrastructure.event.external.repository.ExternalEven
 import org.apache.fineract.infrastructure.event.external.repository.domain.ExternalEvent;
 import org.apache.fineract.infrastructure.event.external.service.idempotency.ExternalEventIdempotencyKeyGenerator;
 import org.apache.fineract.infrastructure.event.external.service.serialization.BusinessEventSerializerFactory;
+import org.apache.fineract.infrastructure.event.external.service.serialization.serializer.BusinessEventSerializer;
 import org.springframework.stereotype.Service;
 import org.springframework.transaction.annotation.Transactional;

@@ -45,8 +46,10 @@ public class ExternalEventService {
        String eventType = event.getType();
        String idempotencyKey = idempotencyKeyGenerator.generate(event);
        try {
-            byte[] data = serializerFactory.create(event).serialize(event);
-            ExternalEvent externalEvent = new ExternalEvent(eventType, data, idempotencyKey);
+            BusinessEventSerializer serializer = serializerFactory.create(event);
+            String schema = serializer.getSupportedSchema().getName();
+            byte[] data = serializer.serialize(event);
+            ExternalEvent externalEvent = new ExternalEvent(eventType, schema, data, idempotencyKey);

            repository.save(externalEvent);
        } catch (IOException e) {
@@ -19,6 +19,7 @@
 package org.apache.fineract.infrastructure.event.external.service.serialization.serializer;

 import java.io.IOException;
+import org.apache.avro.generic.GenericContainer;
 import org.apache.fineract.infrastructure.event.business.domain.BusinessEvent;

 public interface BusinessEventSerializer {
@@ -26,4 +27,6 @@ public interface BusinessEventSerializer {
    <T> boolean canSerialize(BusinessEvent<T> event);

    <T> byte[] serialize(BusinessEvent<T> rawEvent) throws IOException;
+
+    Class<? extends GenericContainer> getSupportedSchema();
 }
@@ -21,6 +21,7 @@ package org.apache.fineract.infrastructure.event.external.service.serialization.
 import java.io.IOException;
 import java.nio.ByteBuffer;
 import lombok.RequiredArgsConstructor;
+import org.apache.avro.generic.GenericContainer;
 import org.apache.fineract.avro.client.v1.ClientDataV1;
 import org.apache.fineract.infrastructure.event.business.domain.BusinessEvent;
 import org.apache.fineract.infrastructure.event.business.domain.client.ClientBusinessEvent;
@@ -52,4 +53,9 @@ public class ClientBusinessEventSerializer implements BusinessEventSerializer {
        ByteBuffer buffer = avroDto.toByteBuffer();
        return byteBufferConverter.convert(buffer);
    }
+
+    @Override
+    public Class<? extends GenericContainer> getSupportedSchema() {
+        return ClientDataV1.class;
+    }
 }
@@ -21,6 +21,7 @@ package org.apache.fineract.infrastructure.event.external.service.serialization.
 import java.io.IOException;
 import java.nio.ByteBuffer;
 import lombok.RequiredArgsConstructor;
+import org.apache.avro.generic.GenericContainer;
 import org.apache.fineract.avro.fixeddeposit.v1.FixedDepositAccountDataV1;
 import org.apache.fineract.infrastructure.event.business.domain.BusinessEvent;
 import org.apache.fineract.infrastructure.event.business.domain.deposit.FixedDepositAccountBusinessEvent;
@@ -53,4 +54,9 @@ public class FixedDepositAccountBusinessEventSerializer implements BusinessEvent
        ByteBuffer buffer = avroDto.toByteBuffer();
        return byteBufferConverter.convert(buffer);
    }
+
+    @Override
+    public Class<? extends GenericContainer> getSupportedSchema() {
+        return FixedDepositAccountDataV1.class;
+    }
 }
@@ -21,6 +21,7 @@ package org.apache.fineract.infrastructure.event.external.service.serialization.
 import java.io.IOException;
 import java.nio.ByteBuffer;
 import lombok.RequiredArgsConstructor;
+import org.apache.avro.generic.GenericContainer;
 import org.apache.fineract.avro.generic.v1.CommandProcessingResultV1;
 import org.apache.fineract.infrastructure.event.business.domain.BusinessEvent;
 import org.apache.fineract.infrastructure.event.business.domain.group.GroupsBusinessEvent;
@@ -48,4 +49,9 @@ public class GroupsBusinessEventSerializer implements BusinessEventSerializer {
        ByteBuffer buffer = avroDto.toByteBuffer();
        return byteBufferConverter.convert(buffer);
    }
+
+    @Override
+    public Class<? extends GenericContainer> getSupportedSchema() {
+        return CommandProcessingResultV1.class;
+    }
 }
@@ -21,6 +21,7 @@ package org.apache.fineract.infrastructure.event.external.service.serialization.
 import java.io.IOException;
 import java.nio.ByteBuffer;
 import lombok.RequiredArgsConstructor;
+import org.apache.avro.generic.GenericContainer;
 import org.apache.fineract.avro.loan.v1.LoanTransactionAdjustmentDataV1;
 import org.apache.fineract.avro.loan.v1.LoanTransactionDataV1;
 import org.apache.fineract.infrastructure.event.business.domain.BusinessEvent;
@@ -67,4 +68,9 @@ public class LoanAdjustTransactionBusinessEventSerializer implements BusinessEve
        ByteBuffer buffer = avroDto.toByteBuffer();
        return byteBufferConverter.convert(buffer);
    }
+
+    @Override
+    public Class<? extends GenericContainer> getSupportedSchema() {
+        return LoanTransactionAdjustmentDataV1.class;
+    }
 }
@@ -21,6 +21,7 @@ package org.apache.fineract.infrastructure.event.external.service.serialization.
 import java.io.IOException;
 import java.nio.ByteBuffer;
 import lombok.RequiredArgsConstructor;
+import org.apache.avro.generic.GenericContainer;
 import org.apache.fineract.avro.loan.v1.LoanAccountDataV1;
 import org.apache.fineract.infrastructure.event.business.domain.BusinessEvent;
 import org.apache.fineract.infrastructure.event.business.domain.loan.LoanBusinessEvent;
@@ -52,4 +53,9 @@ public class LoanBusinessEventSerializer implements BusinessEventSerializer {
        ByteBuffer buffer = avroDto.toByteBuffer();
        return byteBufferConverter.convert(buffer);
    }
+
+    @Override
+    public Class<? extends GenericContainer> getSupportedSchema() {
+        return LoanAccountDataV1.class;
+    }
 }
@@ -21,6 +21,7 @@ package org.apache.fineract.infrastructure.event.external.service.serialization.
 import java.io.IOException;
 import java.nio.ByteBuffer;
 import lombok.RequiredArgsConstructor;
+import org.apache.avro.generic.GenericContainer;
 import org.apache.fineract.avro.loan.v1.LoanChargeDataV1;
 import org.apache.fineract.infrastructure.event.business.domain.BusinessEvent;
 import org.apache.fineract.infrastructure.event.business.domain.loan.charge.LoanChargeBusinessEvent;
@@ -52,4 +53,9 @@ public class LoanChargeBusinessEventSerializer implements BusinessEventSerialize
        ByteBuffer buffer = avroDto.toByteBuffer();
        return byteBufferConverter.convert(buffer);
    }
+
+    @Override
+    public Class<? extends GenericContainer> getSupportedSchema() {
+        return LoanChargeDataV1.class;
+    }
 }
@@ -21,6 +21,7 @@ package org.apache.fineract.infrastructure.event.external.service.serialization.
 import java.io.IOException;
 import java.nio.ByteBuffer;
 import lombok.RequiredArgsConstructor;
+import org.apache.avro.generic.GenericContainer;
 import org.apache.fineract.avro.loan.v1.LoanProductDataV1;
 import org.apache.fineract.infrastructure.event.business.domain.BusinessEvent;
 import org.apache.fineract.infrastructure.event.business.domain.loan.product.LoanProductBusinessEvent;
@@ -52,4 +53,9 @@ public class LoanProductBusinessEventSerializer implements BusinessEventSerializ
        ByteBuffer buffer = avroDto.toByteBuffer();
        return byteBufferConverter.convert(buffer);
    }
+
+    @Override
+    public Class<? extends GenericContainer> getSupportedSchema() {
+        return LoanProductDataV1.class;
+    }
 }
@@ -21,6 +21,7 @@ package org.apache.fineract.infrastructure.event.external.service.serialization.
 import java.io.IOException;
 import java.nio.ByteBuffer;
 import lombok.RequiredArgsConstructor;
+import org.apache.avro.generic.GenericContainer;
 import org.apache.fineract.avro.loan.v1.LoanTransactionDataV1;
 import org.apache.fineract.infrastructure.event.business.domain.BusinessEvent;
 import org.apache.fineract.infrastructure.event.business.domain.loan.transaction.LoanTransactionBusinessEvent;
@@ -52,4 +53,9 @@ public class LoanTransactionBusinessEventSerializer implements BusinessEventSeri
        ByteBuffer buffer = avroDto.toByteBuffer();
        return byteBufferConverter.convert(buffer);
    }
+
+    @Override
+    public Class<? extends GenericContainer> getSupportedSchema() {
+        return LoanTransactionDataV1.class;
+    }
 }
@@ -21,6 +21,7 @@ package org.apache.fineract.infrastructure.event.external.service.serialization.
 import java.io.IOException;
 import java.nio.ByteBuffer;
 import lombok.RequiredArgsConstructor;
+import org.apache.avro.generic.GenericContainer;
 import org.apache.fineract.avro.recurringdeposit.v1.RecurringDepositAccountDataV1;
 import org.apache.fineract.infrastructure.event.business.domain.BusinessEvent;
 import org.apache.fineract.infrastructure.event.business.domain.deposit.RecurringDepositAccountBusinessEvent;
@@ -54,4 +55,9 @@ public class RecurringDepositAccountBusinessEventSerializer implements BusinessE
        ByteBuffer buffer = avroDto.toByteBuffer();
        return byteBufferConverter.convert(buffer);
    }
+
+    @Override
+    public Class<? extends GenericContainer> getSupportedSchema() {
+        return RecurringDepositAccountDataV1.class;
+    }
 }
@@ -21,6 +21,7 @@ package org.apache.fineract.infrastructure.event.external.service.serialization.
 import java.io.IOException;
 import java.nio.ByteBuffer;
 import lombok.RequiredArgsConstructor;
+import org.apache.avro.generic.GenericContainer;
 import org.apache.fineract.avro.savings.v1.SavingsAccountDataV1;
 import org.apache.fineract.infrastructure.event.business.domain.BusinessEvent;
 import org.apache.fineract.infrastructure.event.business.domain.savings.SavingsAccountBusinessEvent;
@@ -52,4 +53,9 @@ public class SavingsAccountBusinessEventSerializer implements BusinessEventSeria
        ByteBuffer buffer = avroDto.toByteBuffer();
        return byteBufferConverter.convert(buffer);
    }
+
+    @Override
+    public Class<? extends GenericContainer> getSupportedSchema() {
+        return SavingsAccountDataV1.class;
+    }
 }
@@ -21,6 +21,7 @@ package org.apache.fineract.infrastructure.event.external.service.serialization.
 import java.io.IOException;
 import java.nio.ByteBuffer;
 import lombok.RequiredArgsConstructor;
+import org.apache.avro.generic.GenericContainer;
 import org.apache.fineract.avro.savings.v1.SavingsAccountTransactionDataV1;
 import org.apache.fineract.infrastructure.event.business.domain.BusinessEvent;
 import org.apache.fineract.infrastructure.event.business.domain.savings.transaction.SavingsAccountTransactionBusinessEvent;
@@ -55,4 +56,9 @@ public class SavingsAccountTransactionBusinessEventSerializer implements Busines
        ByteBuffer buffer = avroDto.toByteBuffer();
        return byteBufferConverter.convert(buffer);
    }
+
+    @Override
+    public Class<? extends GenericContainer> getSupportedSchema() {
+        return SavingsAccountTransactionDataV1.class;
+    }
 }
@@ -21,6 +21,7 @@ package org.apache.fineract.infrastructure.event.external.service.serialization.
 import java.io.IOException;
 import java.nio.ByteBuffer;
 import lombok.RequiredArgsConstructor;
+import org.apache.avro.generic.GenericContainer;
 import org.apache.fineract.avro.share.v1.ShareAccountDataV1;
 import org.apache.fineract.infrastructure.event.business.domain.BusinessEvent;
 import org.apache.fineract.infrastructure.event.business.domain.share.ShareAccountBusinessEvent;
@@ -52,4 +53,9 @@ public class ShareAccountBusinessEventSerializer implements BusinessEventSeriali
        ByteBuffer buffer = avroDto.toByteBuffer();
        return byteBufferConverter.convert(buffer);
    }
+
+    @Override
+    public Class<? extends GenericContainer> getSupportedSchema() {
+        return ShareAccountDataV1.class;
+    }
 }
@@ -21,6 +21,7 @@ package org.apache.fineract.infrastructure.event.external.service.serialization.
 import java.io.IOException;
 import java.nio.ByteBuffer;
 import lombok.RequiredArgsConstructor;
+import org.apache.avro.generic.GenericContainer;
 import org.apache.fineract.avro.share.v1.ShareProductDataV1;
 import org.apache.fineract.infrastructure.event.business.domain.BusinessEvent;
 import org.apache.fineract.infrastructure.event.business.domain.share.ShareProductDividentsCreateBusinessEvent;
@@ -52,4 +53,9 @@ public class ShareProductDividentsCreateBusinessEventSerializer implements Busin
        ByteBuffer buffer = avroDto.toByteBuffer();
        return byteBufferConverter.convert(buffer);
    }
+
+    @Override
+    public Class<? extends GenericContainer> getSupportedSchema() {
+        return ShareProductDataV1.class;
+    }
 }
@@ -65,4 +65,5 @@
    <include file="parts/0043_add_external_event_table.xml" relativeToChangelogFile="true"/>
    <include file="parts/0044_table_report_query_fix.xml" relativeToChangelogFile="true"/>
    <include file="parts/0045_external_event_table_data_binary.xml" relativeToChangelogFile="true"/>
+    <include file="parts/0046_external_event_table_schema_info.xml" relativeToChangelogFile="true"/>
 </databaseChangeLog>
@@ -0,0 +1,35 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+
+    Licensed to the Apache Software Foundation (ASF) under one
+    or more contributor license agreements. See the NOTICE file
+    distributed with this work for additional information
+    regarding copyright ownership. The ASF licenses this file
+    to you under the Apache License, Version 2.0 (the
+    "License"); you may not use this file except in compliance
+    with the License. You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+    Unless required by applicable law or agreed to in writing,
+    software distributed under the License is distributed on an
+    "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+    KIND, either express or implied. See the License for the
+    specific language governing permissions and limitations
+    under the License.
+
+-->
+<databaseChangeLog xmlns="http://www.liquibase.org/xml/ns/dbchangelog"
+                   xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+                   xsi:schemaLocation="http://www.liquibase.org/xml/ns/dbchangelog http://www.liquibase.org/xml/ns/dbchangelog/dbchangelog-4.1.xsd">
+    <changeSet author="fineract" id="1">
+        <delete tableName="m_external_event"/>
+    </changeSet>
+    <changeSet author="fineract" id="2">
+        <addColumn tableName="m_external_event">
+            <column name="schema" type="VARCHAR(300)">
+                <constraints nullable="false"/>
+            </column>
+        </addColumn>
+    </changeSet>
+</databaseChangeLog>
@@ -29,6 +29,7 @@ import java.time.LocalDate;
 import java.time.ZoneId;
 import java.util.HashMap;
 import java.util.Map;
+import org.apache.fineract.avro.loan.v1.LoanAccountDataV1;
 import org.apache.fineract.infrastructure.businessdate.domain.BusinessDateType;
 import org.apache.fineract.infrastructure.core.domain.FineractPlatformTenant;
 import org.apache.fineract.infrastructure.core.service.ThreadLocalContextUtil;
@@ -83,6 +84,7 @@ class ExternalEventServiceTest {

        given(idempotencyKeyGenerator.generate(event)).willReturn("");
        given(serializerFactory.create(event)).willReturn(eventSerializer);
+        given(eventSerializer.getSupportedSchema()).will(invocation -> LoanAccountDataV1.class);
        given(eventSerializer.serialize(event)).willThrow(IOException.class);
        // when & then
        assertThatThrownBy(() -> underTest.postEvent(event)).isExactlyInstanceOf(RuntimeException.class);
@@ -93,6 +95,7 @@ class ExternalEventServiceTest {
        // given
        ArgumentCaptor<ExternalEvent> externalEventArgumentCaptor = ArgumentCaptor.forClass(ExternalEvent.class);

+        String eventSchema = "org.apache.fineract.avro.loan.v1.LoanAccountDataV1";
        String eventType = "TestType";
        String idempotencyKey = "key";
        BusinessEvent event = mock(BusinessEvent.class);
@@ -102,6 +105,7 @@ class ExternalEventServiceTest {
        given(event.getType()).willReturn(eventType);
        given(idempotencyKeyGenerator.generate(event)).willReturn(idempotencyKey);
        given(serializerFactory.create(event)).willReturn(eventSerializer);
+        given(eventSerializer.getSupportedSchema()).will(invocation -> LoanAccountDataV1.class);
        given(eventSerializer.serialize(event)).willReturn(data);
        // when
        underTest.postEvent(event);
@@ -111,5 +115,6 @@ class ExternalEventServiceTest {
        assertThat(externalEvent.getIdempotencyKey()).isEqualTo(idempotencyKey);
        assertThat(externalEvent.getData()).isEqualTo(data);
        assertThat(externalEvent.getType()).isEqualTo(eventType);
+        assertThat(externalEvent.getSchema()).isEqualTo(eventSchema);
    }
 }