Merge pull request #19 from JebronLames32/kafkaSupport

Add Kafka consumer for Fineract external-events

Author: IOhacker

README.md

@@ -1 +1,23 @@
-# message-consumer
+# message-consumer
+
+## Event Consumption via Kafka or JMS
+
+###  Spring Profiles
+
+| Profile     | Behavior                                  |
+|-------------|-------------------------------------------|
+| `kafka`     | Enables Kafka-based event consumption     |
+| `jms`       | Enables JMS/ActiveMQ-based consumption    |
+| *(none)*    | Disables both consumers                   |
+
+---
+
+### Running with Kafka
+```bash
+./gradlew bootRun --args='--spring.profiles.active=kafka'
+```
+
+### Running with JMS
+```bash
+./gradlew bootRun --args='--spring.profiles.active=jms'
+```

build.gradle

@@ -25,11 +25,15 @@ dependencies {
 	implementation group: 'org.apache.avro', name: 'avro', version: '1.12.0'
 	implementation 'org.springframework.boot:spring-boot-starter-data-jpa'
 	implementation 'org.springframework.boot:spring-boot-starter-web'
+	implementation 'org.apache.kafka:kafka-clients:4.0.0'
+	implementation 'org.springframework.kafka:spring-kafka'
+	implementation 'org.springframework.integration:spring-integration-kafka'
 	compileOnly 'org.projectlombok:lombok'
 	runtimeOnly 'com.h2database:h2'
 	annotationProcessor 'org.projectlombok:lombok'
 	testImplementation 'org.springframework.boot:spring-boot-starter-test'
 	testImplementation 'org.springframework.integration:spring-integration-test'
+	testImplementation 'org.springframework.kafka:spring-kafka-test'
 
 	implementation "org.apache.fineract:fineract-avro-schemas:${fineractVersion}"
 }

src/main/java/org/test/consumer/config/KafkaConsumerConfiguration.java

@@ -0,0 +1,45 @@
+package org.test.consumer.config;
+
+import org.apache.kafka.clients.consumer.ConsumerConfig;
+import org.apache.kafka.common.serialization.ByteArrayDeserializer;
+import org.apache.kafka.common.serialization.StringDeserializer;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.context.annotation.Profile;
+import org.springframework.kafka.annotation.EnableKafka;
+import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
+import org.springframework.kafka.core.ConsumerFactory;
+import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
+
+import java.util.HashMap;
+import java.util.Map;
+
+@EnableKafka
+@Configuration
+@Profile("kafka")
+public class KafkaConsumerConfiguration {
+
+    @Value("${spring.kafka.bootstrap-servers}")
+    private String bootstrapServers;
+
+    @Bean
+    public ConsumerFactory<String, byte[]> consumerFactory() {
+        Map<String, Object> props = new HashMap<>();
+        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
+        props.put(ConsumerConfig.GROUP_ID_CONFIG, "fineract-event-ingestor");
+        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
+        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, ByteArrayDeserializer.class.getName());
+        props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
+        return new DefaultKafkaConsumerFactory<>(props);
+    }
+
+    @Bean
+    public ConcurrentKafkaListenerContainerFactory<String, byte[]> kafkaListenerContainerFactory() {
+        var factory = new ConcurrentKafkaListenerContainerFactory<String, byte[]>();
+        factory.setConsumerFactory(consumerFactory());
+        return factory;
+    }
+}
+
+

src/main/java/org/test/consumer/config/MessageConsumerJMSBrokerConfiguration.java

@@ -5,9 +5,11 @@
 import org.springframework.beans.factory.annotation.Value;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
+import org.springframework.context.annotation.Profile;
 import org.springframework.util.StringUtils;
 
 @Configuration
+@Profile("jms")
 public class MessageConsumerJMSBrokerConfiguration {
 
     @Value("${brokerurl}")

src/main/java/org/test/consumer/config/MessageConsumerJMSConfiguration.java

@@ -7,13 +7,15 @@
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
 import org.springframework.context.annotation.Import;
+import org.springframework.context.annotation.Profile;
 import org.springframework.integration.annotation.ServiceActivator;
 import org.springframework.integration.channel.DirectChannel;
 import org.springframework.integration.dsl.IntegrationFlow;
 import org.springframework.integration.jms.dsl.Jms;
 import org.test.consumer.handler.JMSMessageConsumerHandler;
 
 @Configuration
+@Profile("jms")
 @Import(value = {MessageConsumerJMSBrokerConfiguration.class})
 public class MessageConsumerJMSConfiguration {
     @Autowired

src/main/java/org/test/consumer/handler/JMSMessageConsumerHandler.java

@@ -7,6 +7,7 @@
 import lombok.extern.slf4j.Slf4j;
 import org.apache.fineract.avro.MessageV1;
 import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.context.annotation.Profile;
 import org.springframework.messaging.Message;
 import org.springframework.messaging.MessageHandler;
 import org.springframework.messaging.MessagingException;
@@ -18,6 +19,7 @@
 
 @Component
 @Slf4j
+@Profile("jms")
 public class JMSMessageConsumerHandler implements MessageHandler {
     @Autowired
     private ByteBufferConvertor byteBufferConvertor;

src/main/java/org/test/consumer/handler/KafkaMessageConsumerHandler.java

@@ -0,0 +1,72 @@
+package org.test.consumer.handler;
+
+import lombok.extern.slf4j.Slf4j;
+import org.apache.fineract.avro.MessageV1;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.context.annotation.Profile;
+import org.springframework.kafka.annotation.KafkaListener;
+import org.springframework.messaging.Message;
+import org.springframework.messaging.MessageHandler;
+import org.springframework.messaging.MessagingException;
+import org.springframework.stereotype.Component;
+import org.test.consumer.domain.EventMessage;
+import org.test.consumer.repository.EventMessageRepository;
+import org.test.consumer.utility.ByteBufferConvertor;
+
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.time.LocalDateTime;
+import java.time.format.DateTimeFormatter;
+
+/**
+ * Consumes external Fineract events from Kafka and persists them.
+ *
+ * Mirrors the structure of {@link JMSMessageConsumerHandler}.
+ */
+@Component
+@Slf4j
+@Profile("kafka")
+public class KafkaMessageConsumerHandler implements MessageHandler {
+
+    @Autowired
+    private ByteBufferConvertor byteBufferConvertor;
+
+    @Autowired
+    private EventMessageRepository repository;
+
+    @Override
+    @KafkaListener(
+            topics = "${app.kafka.topic:external-events}",
+            containerFactory = "kafkaListenerContainerFactory"
+    )
+    public void handleMessage(Message<?> springMessage) throws MessagingException {
+        byte[] rawPayload = (byte[]) springMessage.getPayload();
+        ByteBuffer wrapperBuf = byteBufferConvertor.convert(rawPayload);
+
+        try {
+            MessageV1 messagePayload = MessageV1.fromByteBuffer(wrapperBuf);
+            log.info("Received Kafka event of Category = {}, Type = {}",
+                    messagePayload.getCategory(), messagePayload.getType());
+            saveMessage(messagePayload);
+        } catch (IOException e) {
+            log.error("Unable to read message", e);
+        }
+    }
+
+    private void saveMessage(MessageV1 messagePayload) {
+        LocalDateTime createdAt =
+                LocalDateTime.parse(messagePayload.getCreatedAt(), DateTimeFormatter.ISO_DATE_TIME);
+
+        EventMessage message = new EventMessage(
+                messagePayload.getId(),
+                messagePayload.getType(),
+                messagePayload.getCategory(),
+                messagePayload.getDataschema(),
+                messagePayload.getTenantId(),
+                createdAt,
+                byteBufferConvertor.convert(messagePayload.getData()),
+                messagePayload.getBusinessDate()
+        );
+        repository.save(message);
+    }
+}

src/main/resources/application.properties

@@ -10,3 +10,12 @@ spring.datasource.driverClassName=org.h2.Driver
 spring.jpa.database-platform=org.hibernate.dialect.H2Dialect
 spring.jpa.hibernate.ddl-auto=validate
 
+#kafka
+spring.kafka.bootstrap-servers=localhost:9092
+app.kafka.topic=external-events
+spring.kafka.consumer.group-id=fineract-event-ingestor
+spring.kafka.consumer.key-deserializer=org.apache.kafka.common.serialization.StringDeserializer
+spring.kafka.consumer.value-deserializer=org.apache.kafka.common.serialization.ByteArrayDeserializer
+spring.kafka.consumer.auto-offset-reset=earliest
+spring.kafka.admin.auto-create=true
+