FINERACT-1824

- Kafka support for remote batch jobs: cob
 - Kafka support for external events

fineract-provider/src/main/java/org/apache/fineract/infrastructure/event/external/config/ExternalEventKafkaConfiguration.java

@@ -0,0 +1,61 @@
+/**
+ * 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.
+ */
+package org.apache.fineract.infrastructure.event.external.config;
+
+import static org.apache.kafka.clients.producer.ProducerConfig.BOOTSTRAP_SERVERS_CONFIG;
+import static org.apache.kafka.clients.producer.ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG;
+import static org.apache.kafka.clients.producer.ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG;
+
+import java.util.HashMap;
+import java.util.Map;
+import org.apache.fineract.infrastructure.core.config.FineractProperties;
+import org.apache.kafka.common.serialization.ByteArraySerializer;
+import org.apache.kafka.common.serialization.LongSerializer;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.kafka.core.DefaultKafkaProducerFactory;
+import org.springframework.kafka.core.KafkaTemplate;
+import org.springframework.kafka.core.ProducerFactory;
+
+@Configuration
+@ConditionalOnProperty(value = "fineract.events.external.producer.kafka.enabled", havingValue = "true")
+public class ExternalEventKafkaConfiguration {
+
+    @Autowired
+    private FineractProperties fineractProperties;
+
+    @Bean
+    public ProducerFactory<Long, byte[]> externalEventsProducerFactory() {
+        FineractProperties.FineractExternalEventsProducerKafkaProperties kafkaProp = fineractProperties.getEvents().getExternal()
+                .getProducer().getKafka();
+        Map<String, Object> props = new HashMap<>(kafkaProp.getProducer().getExtraPropertiesMap());
+        props.put(BOOTSTRAP_SERVERS_CONFIG, kafkaProp.getBootstrapServers());
+        props.put(KEY_SERIALIZER_CLASS_CONFIG, LongSerializer.class);
+        props.put(VALUE_SERIALIZER_CLASS_CONFIG, ByteArraySerializer.class);
+        return new DefaultKafkaProducerFactory<>(props);
+    }
+
+    @Bean
+    public KafkaTemplate<Long, byte[]> externalEventsKafkaTemplate(ProducerFactory<Long, byte[]> externalEventsProducerFactory) {
+        return new KafkaTemplate<>(externalEventsProducerFactory);
+    }
+
+}

fineract-provider/src/main/java/org/apache/fineract/infrastructure/event/external/config/ExternalEventsKafkaTopicAutoCreateCondition.java

@@ -0,0 +1,36 @@
+/**
+ * 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.
+ */
+package org.apache.fineract.infrastructure.event.external.config;
+
+import org.springframework.boot.autoconfigure.condition.AllNestedConditions;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
+
+public class ExternalEventsKafkaTopicAutoCreateCondition extends AllNestedConditions {
+
+    public ExternalEventsKafkaTopicAutoCreateCondition() {
+        super(ConfigurationPhase.PARSE_CONFIGURATION);
+    }
+
+    @ConditionalOnProperty(value = "fineract.events.external.producer.kafka.enabled", havingValue = "true")
+    static class ExternalEventsKafkaCondition {}
+
+    @ConditionalOnProperty(value = "fineract.events.external.producer.kafka.topic.auto-create", havingValue = "true")
+    static class TopicAutoCreateCondition {}
+
+}

fineract-provider/src/main/java/org/apache/fineract/infrastructure/event/external/config/KafkaExternalEventTopicConfig.java

@@ -0,0 +1,57 @@
+/**
+ * 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.
+ */
+package org.apache.fineract.infrastructure.event.external.config;
+
+import static org.apache.kafka.clients.producer.ProducerConfig.BOOTSTRAP_SERVERS_CONFIG;
+
+import java.util.HashMap;
+import java.util.Map;
+import org.apache.fineract.infrastructure.core.config.FineractProperties;
+import org.apache.kafka.clients.admin.NewTopic;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Conditional;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.kafka.config.TopicBuilder;
+import org.springframework.kafka.core.KafkaAdmin;
+
+@Configuration
+@Conditional(ExternalEventsKafkaTopicAutoCreateCondition.class)
+public class KafkaExternalEventTopicConfig {
+
+    @Autowired
+    private FineractProperties fineractProperties;
+
+    @Bean
+    public KafkaAdmin admin() {
+        Map<String, Object> props = new HashMap<>(
+                fineractProperties.getEvents().getExternal().getProducer().getKafka().getAdmin().getExtraPropertiesMap());
+        props.put(BOOTSTRAP_SERVERS_CONFIG, fineractProperties.getRemoteJobMessageHandler().getKafka().getBootstrapServers());
+        return new KafkaAdmin(props);
+    }
+
+    @Bean
+    public NewTopic externalEventsTopic() {
+        FineractProperties.KafkaTopicProperties topicProperties = fineractProperties.getEvents().getExternal().getProducer().getKafka()
+                .getTopic();
+        return TopicBuilder.name(topicProperties.getName()).partitions(topicProperties.getPartitions())
+                .replicas(topicProperties.getReplicas()).build();
+    }
+
+}

fineract-provider/src/main/java/org/apache/fineract/infrastructure/event/external/producer/kafka/KafkaExternalEventProducer.java

@@ -0,0 +1,82 @@
+/**
+ * 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.
+ */
+package org.apache.fineract.infrastructure.event.external.producer.kafka;
+
+import static org.apache.fineract.infrastructure.core.diagnostics.performance.MeasuringUtil.measure;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.TimeUnit;
+import lombok.AllArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.fineract.infrastructure.core.config.FineractProperties;
+import org.apache.fineract.infrastructure.event.external.exception.AcknowledgementTimeoutException;
+import org.apache.fineract.infrastructure.event.external.producer.ExternalEventProducer;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
+import org.springframework.kafka.core.KafkaTemplate;
+import org.springframework.kafka.support.SendResult;
+import org.springframework.stereotype.Component;
+
+@Component
+@Slf4j
+@ConditionalOnProperty(value = "fineract.events.external.producer.kafka.enabled", havingValue = "true")
+@AllArgsConstructor
+public class KafkaExternalEventProducer implements ExternalEventProducer {
+
+    @Autowired
+    private KafkaTemplate<Long, byte[]> externalEventsKafkaTemplate;
+
+    @Autowired
+    private FineractProperties fineractProperties;
+
+    @Override
+    public void sendEvents(Map<Long, List<byte[]>> partitions) throws AcknowledgementTimeoutException {
+        FineractProperties.FineractExternalEventsProducerKafkaProperties kafkaProperties = fineractProperties.getEvents().getExternal()
+                .getProducer().getKafka();
+        String topicName = kafkaProperties.getTopic().getName();
+        List<CompletableFuture<SendResult<Long, byte[]>>> sendResults = new ArrayList<>();
+        measure(() -> {
+            Set<Long> keys = partitions.keySet();
+            for (Long key : keys) {
+                List<byte[]> messages = partitions.get(key);
+                for (byte[] message : messages) {
+                    sendResults.add(externalEventsKafkaTemplate.send(topicName, key, message));
+                }
+            }
+
+            try {
+                CompletableFuture<Void> allOf = CompletableFuture.allOf(sendResults.toArray(new CompletableFuture[0]));
+                allOf.get(kafkaProperties.getTimeoutInSeconds(), TimeUnit.SECONDS);
+            } catch (Exception exception) {
+                throw new RuntimeException("Could not send the messages", exception);
+            }
+        }, timeTaken -> {
+            if (log.isDebugEnabled()) {
+                int eventCount = partitions.values().stream().map(Collection::size).reduce(0, Integer::sum);
+                int msgPerSec = (int) (((double) eventCount / timeTaken.toMillis()) * 1000);
+                log.debug("Sent messages with {} msg/s", msgPerSec);
+            }
+        });
+    }
+}

fineract-provider/src/main/java/org/apache/fineract/infrastructure/springbatch/InputChannelInterceptor.java

@@ -37,9 +37,12 @@ public class InputChannelInterceptor implements ExecutorChannelInterceptor {
     }
 
     public Message<StepExecutionRequest> beforeHandleMessage(Message<?> message) {
-        ContextualMessage castedMessage = (ContextualMessage) message.getPayload();
-        ThreadLocalContextUtil.init(castedMessage.getContext());
+        return new GenericMessage<>(beforeHandleMessage((ContextualMessage) message.getPayload()));
+    }
+
+    public StepExecutionRequest beforeHandleMessage(ContextualMessage contextualMessage) {
+        ThreadLocalContextUtil.init(contextualMessage.getContext());
         ThreadLocalContextUtil.setActionContext(ActionContext.COB);
-        return new GenericMessage<>(castedMessage.getStepExecutionRequest());
+        return contextualMessage.getStepExecutionRequest();
     }
 }

fineract-provider/src/main/java/org/apache/fineract/infrastructure/springbatch/messagehandler/MessageHandlerConfig.java

@@ -0,0 +1,36 @@
+/**
+ * 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.
+ */
+package org.apache.fineract.infrastructure.springbatch.messagehandler;
+
+import org.springframework.batch.core.step.StepLocator;
+import org.springframework.batch.integration.partition.BeanFactoryStepLocator;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+
+@Configuration
+@ConditionalOnProperty(value = "fineract.mode.batch-worker-enabled", havingValue = "true")
+public class MessageHandlerConfig {
+
+    @Bean
+    public StepLocator stepLocator() {
+        return new BeanFactoryStepLocator();
+    }
+
+}

fineract-provider/src/main/java/org/apache/fineract/infrastructure/springbatch/messagehandler/StepExecutionRequestHandler.java

@@ -19,7 +19,6 @@
 package org.apache.fineract.infrastructure.springbatch.messagehandler;
 
 import lombok.RequiredArgsConstructor;
-import org.apache.fineract.infrastructure.springbatch.messagehandler.conditions.JmsWorkerCondition;
 import org.springframework.batch.core.BatchStatus;
 import org.springframework.batch.core.JobInterruptedException;
 import org.springframework.batch.core.Step;
@@ -28,13 +27,13 @@ import org.springframework.batch.core.explore.JobExplorer;
 import org.springframework.batch.core.repository.JobRepository;
 import org.springframework.batch.core.step.StepLocator;
 import org.springframework.batch.integration.partition.StepExecutionRequest;
-import org.springframework.context.annotation.Conditional;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
 import org.springframework.dao.OptimisticLockingFailureException;
 import org.springframework.stereotype.Component;
 
 @Component
 @RequiredArgsConstructor
-@Conditional(JmsWorkerCondition.class)
+@ConditionalOnProperty(value = "fineract.mode.batch-worker-enabled", havingValue = "true")
 public class StepExecutionRequestHandler {
 
     private final JobRepository jobRepository;

fineract-provider/src/main/java/org/apache/fineract/infrastructure/springbatch/messagehandler/conditions/jms/JmsManagerCondition.java

@@ -16,7 +16,7 @@
  * specific language governing permissions and limitations
  * under the License.
  */
-package org.apache.fineract.infrastructure.springbatch.messagehandler.conditions;
+package org.apache.fineract.infrastructure.springbatch.messagehandler.conditions.jms;
 
 import org.springframework.boot.autoconfigure.condition.AllNestedConditions;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;

fineract-provider/src/main/java/org/apache/fineract/infrastructure/springbatch/messagehandler/conditions/jms/JmsWorkerCondition.java

@@ -16,7 +16,7 @@
  * specific language governing permissions and limitations
  * under the License.
  */
-package org.apache.fineract.infrastructure.springbatch.messagehandler.conditions;
+package org.apache.fineract.infrastructure.springbatch.messagehandler.conditions.jms;
 
 import org.springframework.boot.autoconfigure.condition.AllNestedConditions;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
@@ -28,8 +28,8 @@ public class JmsWorkerCondition extends AllNestedConditions {
     }
 
     @ConditionalOnProperty(value = "fineract.mode.batch-worker-enabled", havingValue = "true")
-    static class WorkerCondition {}
+    public static class WorkerCondition {}
 
     @ConditionalOnProperty(value = "fineract.remote-job-message-handler.jms.enabled", havingValue = "true")
-    static class JmsCondition {}
+    public static class JmsCondition {}
 }

fineract-provider/src/main/java/org/apache/fineract/infrastructure/springbatch/messagehandler/conditions/kafka/KafkaManagerCondition.java

@@ -0,0 +1,35 @@
+/**
+ * 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.
+ */
+package org.apache.fineract.infrastructure.springbatch.messagehandler.conditions.kafka;
+
+import org.springframework.boot.autoconfigure.condition.AllNestedConditions;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
+
+public class KafkaManagerCondition extends AllNestedConditions {
+
+    public KafkaManagerCondition() {
+        super(ConfigurationPhase.PARSE_CONFIGURATION);
+    }
+
+    @ConditionalOnProperty(value = "fineract.mode.batch-manager-enabled", havingValue = "true")
+    static class ManagerCondition {}
+
+    @ConditionalOnProperty(value = "fineract.remote-job-message-handler.kafka.enabled", havingValue = "true")
+    static class KafkaCondition {}
+}

fineract-provider/src/main/java/org/apache/fineract/infrastructure/springbatch/messagehandler/conditions/kafka/KafkaRemoteJobTopicAutoCreateCondition.java

@@ -0,0 +1,39 @@
+/**
+ * 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.
+ */
+package org.apache.fineract.infrastructure.springbatch.messagehandler.conditions.kafka;
+
+import org.springframework.boot.autoconfigure.condition.AllNestedConditions;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
+
+public class KafkaRemoteJobTopicAutoCreateCondition extends AllNestedConditions {
+
+    public KafkaRemoteJobTopicAutoCreateCondition() {
+        super(ConfigurationPhase.PARSE_CONFIGURATION);
+    }
+
+    @ConditionalOnProperty(value = "fineract.mode.batch-manager-enabled", havingValue = "true")
+    static class ManagerCondition {}
+
+    @ConditionalOnProperty(value = "fineract.remote-job-message-handler.kafka.enabled", havingValue = "true")
+    static class KafkaRemoteJobMessageHandlerEnabled {}
+
+    @ConditionalOnProperty(value = "fineract.remote-job-message-handler.kafka.topic.auto-create", havingValue = "true")
+    static class TopicAutoCreateCondition {}
+
+}

fineract-provider/src/main/java/org/apache/fineract/infrastructure/springbatch/messagehandler/conditions/kafka/KafkaWorkerCondition.java

@@ -0,0 +1,35 @@
+/**
+ * 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.
+ */
+package org.apache.fineract.infrastructure.springbatch.messagehandler.conditions.kafka;
+
+import org.springframework.boot.autoconfigure.condition.AllNestedConditions;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
+
+public class KafkaWorkerCondition extends AllNestedConditions {
+
+    public KafkaWorkerCondition() {
+        super(ConfigurationPhase.PARSE_CONFIGURATION);
+    }
+
+    @ConditionalOnProperty(value = "fineract.mode.batch-worker-enabled", havingValue = "true")
+    static class WorkerCondition {}
+
+    @ConditionalOnProperty(value = "fineract.remote-job-message-handler.kafka.enabled", havingValue = "true")
+    static class KafkaCondition {}
+}

fineract-provider/src/main/java/org/apache/fineract/infrastructure/springbatch/messagehandler/conditions/spring/SpringEventManagerCondition.java

@@ -16,7 +16,7 @@
  * specific language governing permissions and limitations
  * under the License.
  */
-package org.apache.fineract.infrastructure.springbatch.messagehandler.conditions;
+package org.apache.fineract.infrastructure.springbatch.messagehandler.conditions.spring;
 
 import org.springframework.boot.autoconfigure.condition.AllNestedConditions;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;

fineract-provider/src/main/java/org/apache/fineract/infrastructure/springbatch/messagehandler/conditions/spring/SpringEventWorkerCondition.java

@@ -16,7 +16,7 @@
  * specific language governing permissions and limitations
  * under the License.
  */
-package org.apache.fineract.infrastructure.springbatch.messagehandler.conditions;
+package org.apache.fineract.infrastructure.springbatch.messagehandler.conditions.spring;
 
 import org.springframework.boot.autoconfigure.condition.AllNestedConditions;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;

fineract-provider/src/main/java/org/apache/fineract/infrastructure/springbatch/messagehandler/jms/JmsBatchWorkerMessageListener.java

@@ -16,7 +16,7 @@
  * specific language governing permissions and limitations
  * under the License.
  */
-package org.apache.fineract.infrastructure.springbatch.messagehandler;
+package org.apache.fineract.infrastructure.springbatch.messagehandler.jms;
 
 import jakarta.jms.JMSException;
 import jakarta.jms.MessageListener;
@@ -24,7 +24,8 @@ import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.fineract.infrastructure.springbatch.ContextualMessage;
 import org.apache.fineract.infrastructure.springbatch.InputChannelInterceptor;
-import org.apache.fineract.infrastructure.springbatch.messagehandler.conditions.JmsWorkerCondition;
+import org.apache.fineract.infrastructure.springbatch.messagehandler.StepExecutionRequestHandler;
+import org.apache.fineract.infrastructure.springbatch.messagehandler.conditions.jms.JmsWorkerCondition;
 import org.springframework.batch.integration.partition.StepExecutionRequest;
 import org.springframework.beans.factory.InitializingBean;
 import org.springframework.context.annotation.Conditional;

fineract-provider/src/main/java/org/apache/fineract/infrastructure/springbatch/messagehandler/jms/JmsBrokerConfiguration.java

@@ -16,7 +16,7 @@
  * specific language governing permissions and limitations
  * under the License.
  */
-package org.apache.fineract.infrastructure.springbatch.messagehandler;
+package org.apache.fineract.infrastructure.springbatch.messagehandler.jms;
 
 import org.apache.activemq.ActiveMQConnectionFactory;
 import org.apache.activemq.ActiveMQPrefetchPolicy;

fineract-provider/src/main/java/org/apache/fineract/infrastructure/springbatch/messagehandler/jms/JmsManagerConfig.java

@@ -16,12 +16,12 @@
  * specific language governing permissions and limitations
  * under the License.
  */
-package org.apache.fineract.infrastructure.springbatch.messagehandler;
+package org.apache.fineract.infrastructure.springbatch.messagehandler.jms;
 
 import jakarta.jms.ConnectionFactory;
 import org.apache.fineract.infrastructure.core.config.FineractProperties;
 import org.apache.fineract.infrastructure.springbatch.OutputChannelInterceptor;
-import org.apache.fineract.infrastructure.springbatch.messagehandler.conditions.JmsManagerCondition;
+import org.apache.fineract.infrastructure.springbatch.messagehandler.conditions.jms.JmsManagerCondition;
 import org.springframework.batch.integration.config.annotation.EnableBatchIntegration;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.context.annotation.Bean;
@@ -30,7 +30,6 @@ import org.springframework.context.annotation.Configuration;
 import org.springframework.context.annotation.Import;
 import org.springframework.integration.channel.DirectChannel;
 import org.springframework.integration.dsl.IntegrationFlow;
-import org.springframework.integration.dsl.IntegrationFlows;
 import org.springframework.integration.handler.LoggingHandler;
 import org.springframework.integration.jms.dsl.Jms;
 
@@ -49,7 +48,7 @@ public class JmsManagerConfig {
 
     @Bean
     public IntegrationFlow outboundFlow(ConnectionFactory connectionFactory) {
-        return IntegrationFlows.from(outboundRequests) //
+        return IntegrationFlow.from(outboundRequests) //
                 .intercept(outputInterceptor) //
                 .log(LoggingHandler.Level.DEBUG) //
                 .handle(Jms.outboundAdapter(connectionFactory)

fineract-provider/src/main/java/org/apache/fineract/infrastructure/springbatch/messagehandler/jms/JmsWorkerConfig.java

@@ -16,14 +16,12 @@
  * specific language governing permissions and limitations
  * under the License.
  */
-package org.apache.fineract.infrastructure.springbatch.messagehandler;
+package org.apache.fineract.infrastructure.springbatch.messagehandler.jms;
 
 import jakarta.jms.ConnectionFactory;
 import jakarta.jms.Session;
 import org.apache.fineract.infrastructure.core.config.FineractProperties;
-import org.apache.fineract.infrastructure.springbatch.messagehandler.conditions.JmsWorkerCondition;
-import org.springframework.batch.core.step.StepLocator;
-import org.springframework.batch.integration.partition.BeanFactoryStepLocator;
+import org.apache.fineract.infrastructure.springbatch.messagehandler.conditions.jms.JmsWorkerCondition;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.beans.factory.annotation.Qualifier;
 import org.springframework.context.annotation.Bean;
@@ -63,8 +61,4 @@ public class JmsWorkerConfig {
         return factory.createListenerContainer(endpoint);
     }
 
-    @Bean
-    public StepLocator stepLocator() {
-        return new BeanFactoryStepLocator();
-    }
 }

fineract-provider/src/main/java/org/apache/fineract/infrastructure/springbatch/messagehandler/kafka/KafkaJobTopicConfig.java

@@ -0,0 +1,57 @@
+/**
+ * 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.
+ */
+package org.apache.fineract.infrastructure.springbatch.messagehandler.kafka;
+
+import static org.apache.kafka.clients.producer.ProducerConfig.BOOTSTRAP_SERVERS_CONFIG;
+
+import java.util.HashMap;
+import java.util.Map;
+import org.apache.fineract.infrastructure.core.config.FineractProperties;
+import org.apache.fineract.infrastructure.springbatch.messagehandler.conditions.kafka.KafkaRemoteJobTopicAutoCreateCondition;
+import org.apache.kafka.clients.admin.NewTopic;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Conditional;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.kafka.config.TopicBuilder;
+import org.springframework.kafka.core.KafkaAdmin;
+
+@Configuration
+@Conditional(KafkaRemoteJobTopicAutoCreateCondition.class)
+public class KafkaJobTopicConfig {
+
+    @Autowired
+    private FineractProperties fineractProperties;
+
+    @Bean
+    public KafkaAdmin admin() {
+        Map<String, Object> props = new HashMap<>(
+                fineractProperties.getRemoteJobMessageHandler().getKafka().getAdmin().getExtraPropertiesMap());
+        props.put(BOOTSTRAP_SERVERS_CONFIG, fineractProperties.getRemoteJobMessageHandler().getKafka().getBootstrapServers());
+        return new KafkaAdmin(props);
+    }
+
+    @Bean
+    public NewTopic jobTopic() {
+        FineractProperties.KafkaTopicProperties topicProperties = fineractProperties.getRemoteJobMessageHandler().getKafka().getTopic();
+        return TopicBuilder.name(topicProperties.getName()).partitions(topicProperties.getPartitions())
+                .replicas(topicProperties.getReplicas()).build();
+    }
+
+}

fineract-provider/src/main/java/org/apache/fineract/infrastructure/springbatch/messagehandler/kafka/KafkaManagerConfig.java

@@ -0,0 +1,91 @@
+/**
+ * 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.
+ */
+package org.apache.fineract.infrastructure.springbatch.messagehandler.kafka;
+
+import static org.apache.kafka.clients.producer.ProducerConfig.BOOTSTRAP_SERVERS_CONFIG;
+import static org.apache.kafka.clients.producer.ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG;
+import static org.apache.kafka.clients.producer.ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG;
+
+import java.util.HashMap;
+import java.util.Map;
+import org.apache.fineract.infrastructure.core.config.FineractProperties;
+import org.apache.fineract.infrastructure.springbatch.ContextualMessage;
+import org.apache.fineract.infrastructure.springbatch.OutputChannelInterceptor;
+import org.apache.fineract.infrastructure.springbatch.messagehandler.conditions.kafka.KafkaManagerCondition;
+import org.apache.kafka.common.serialization.IntegerSerializer;
+import org.springframework.batch.integration.config.annotation.EnableBatchIntegration;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Conditional;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.expression.common.LiteralExpression;
+import org.springframework.integration.channel.DirectChannel;
+import org.springframework.integration.dsl.IntegrationFlow;
+import org.springframework.integration.expression.FunctionExpression;
+import org.springframework.integration.handler.LoggingHandler;
+import org.springframework.integration.kafka.outbound.KafkaProducerMessageHandler;
+import org.springframework.kafka.core.DefaultKafkaProducerFactory;
+import org.springframework.kafka.core.KafkaTemplate;
+import org.springframework.kafka.core.ProducerFactory;
+import org.springframework.kafka.support.serializer.JsonSerializer;
+import org.springframework.messaging.support.GenericMessage;
+
+@Configuration
+@EnableBatchIntegration
+@Conditional(KafkaManagerCondition.class)
+public class KafkaManagerConfig {
+
+    @Autowired
+    private DirectChannel outboundRequests;
+    @Autowired
+    private OutputChannelInterceptor outputInterceptor;
+    @Autowired
+    private FineractProperties fineractProperties;
+
+    @Bean
+    public ProducerFactory<Object, Object> producerFactory() {
+        Map<String, Object> props = new HashMap<>(
+                fineractProperties.getRemoteJobMessageHandler().getKafka().getProducer().getExtraPropertiesMap());
+        props.put(BOOTSTRAP_SERVERS_CONFIG, fineractProperties.getRemoteJobMessageHandler().getKafka().getBootstrapServers());
+        props.put(KEY_SERIALIZER_CLASS_CONFIG, IntegerSerializer.class);
+        props.put(VALUE_SERIALIZER_CLASS_CONFIG, JsonSerializer.class);
+        return new DefaultKafkaProducerFactory<>(props);
+    }
+
+    @Bean
+    public KafkaTemplate<?, ?> kafkaTemplate(ProducerFactory<Object, Object> producerFactory) {
+        return new KafkaTemplate<>(producerFactory);
+    }
+
+    @Bean
+    public IntegrationFlow outboundFlow(KafkaTemplate<?, ?> kafkaTemplate) {
+        FineractProperties.KafkaTopicProperties topic = fineractProperties.getRemoteJobMessageHandler().getKafka().getTopic();
+
+        KafkaProducerMessageHandler<?, ?> messageHandler = new KafkaProducerMessageHandler<>(kafkaTemplate);
+        messageHandler.setTopicExpression(new LiteralExpression(topic.getName()));
+        messageHandler.setPartitionIdExpression(new FunctionExpression<GenericMessage<ContextualMessage>>(
+                message -> message.getPayload().getStepExecutionRequest().getStepExecutionId() % topic.getPartitions()));
+
+        return IntegrationFlow.from(outboundRequests) //
+                .intercept(outputInterceptor) //
+                .log(LoggingHandler.Level.DEBUG) //
+                .handle(messageHandler).get();
+    }
+
+}

fineract-provider/src/main/java/org/apache/fineract/infrastructure/springbatch/messagehandler/kafka/KafkaRemoteMessageListener.java

@@ -0,0 +1,57 @@
+/**
+ * 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.
+ */
+package org.apache.fineract.infrastructure.springbatch.messagehandler.kafka;
+
+import lombok.extern.slf4j.Slf4j;
+import org.apache.fineract.infrastructure.springbatch.ContextualMessage;
+import org.apache.fineract.infrastructure.springbatch.InputChannelInterceptor;
+import org.apache.fineract.infrastructure.springbatch.messagehandler.StepExecutionRequestHandler;
+import org.apache.fineract.infrastructure.springbatch.messagehandler.conditions.kafka.KafkaWorkerCondition;
+import org.springframework.batch.integration.partition.StepExecutionRequest;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.context.annotation.Conditional;
+import org.springframework.kafka.annotation.KafkaListener;
+import org.springframework.kafka.support.Acknowledgment;
+import org.springframework.messaging.handler.annotation.Payload;
+import org.springframework.stereotype.Component;
+
+@Component
+@Slf4j
+@Conditional(KafkaWorkerCondition.class)
+public class KafkaRemoteMessageListener {
+
+    @Autowired
+    private StepExecutionRequestHandler stepExecutionRequestHandler;
+
+    @Autowired
+    private InputChannelInterceptor inputInterceptor;
+
+    @KafkaListener(topics = "${fineract.remote-job-message-handler.kafka.topic.name}")
+    public void onMessage(@Payload ContextualMessage contextualMessage, Acknowledgment acknowledgment) {
+        try {
+            log.debug("Received Kafka partition message {}", contextualMessage);
+            StepExecutionRequest stepExecutionRequest = inputInterceptor.beforeHandleMessage(contextualMessage);
+            stepExecutionRequestHandler.handle(stepExecutionRequest);
+        } catch (Exception e) {
+            log.error("Exception while processing Kafka message", e);
+        }
+        acknowledgment.acknowledge();
+        log.debug("Message was acknowledged {}", acknowledgment);
+    }
+}

fineract-provider/src/main/java/org/apache/fineract/infrastructure/springbatch/messagehandler/kafka/KafkaWorkerConfig.java

@@ -0,0 +1,79 @@
+/**
+ * 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.
+ */
+package org.apache.fineract.infrastructure.springbatch.messagehandler.kafka;
+
+import static org.apache.kafka.clients.consumer.ConsumerConfig.AUTO_OFFSET_RESET_CONFIG;
+import static org.apache.kafka.clients.consumer.ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG;
+import static org.apache.kafka.clients.consumer.ConsumerConfig.GROUP_ID_CONFIG;
+import static org.apache.kafka.clients.consumer.ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG;
+import static org.apache.kafka.clients.consumer.ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG;
+import static org.springframework.kafka.support.serializer.JsonDeserializer.TRUSTED_PACKAGES;
+
+import java.util.HashMap;
+import java.util.Map;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.fineract.infrastructure.core.config.FineractProperties;
+import org.apache.fineract.infrastructure.core.config.FineractProperties.FineractRemoteJobMessageHandlerKafkaProperties;
+import org.apache.fineract.infrastructure.springbatch.messagehandler.conditions.kafka.KafkaWorkerCondition;
+import org.apache.kafka.common.serialization.IntegerDeserializer;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Conditional;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.integration.channel.QueueChannel;
+import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
+import org.springframework.kafka.core.ConsumerFactory;
+import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
+import org.springframework.kafka.listener.ContainerProperties;
+import org.springframework.kafka.support.serializer.JsonDeserializer;
+
+@Configuration
+@Conditional(KafkaWorkerCondition.class)
+@Slf4j
+public class KafkaWorkerConfig {
+
+    @Autowired
+    private QueueChannel inboundRequests;
+
+    @Autowired
+    private FineractProperties fineractProperties;
+
+    @Bean
+    public ConsumerFactory<Integer, Object> consumerFactory() {
+        FineractRemoteJobMessageHandlerKafkaProperties kafkaProperties = fineractProperties.getRemoteJobMessageHandler().getKafka();
+        Map<String, Object> props = new HashMap<>(kafkaProperties.getConsumer().getExtraPropertiesMap());
+        props.put(BOOTSTRAP_SERVERS_CONFIG, kafkaProperties.getBootstrapServers());
+        props.put(GROUP_ID_CONFIG, kafkaProperties.getConsumer().getGroupId());
+        props.put(KEY_DESERIALIZER_CLASS_CONFIG, IntegerDeserializer.class);
+        props.put(VALUE_DESERIALIZER_CLASS_CONFIG, JsonDeserializer.class);
+        props.put(TRUSTED_PACKAGES, "*");
+        props.put(AUTO_OFFSET_RESET_CONFIG, "earliest");
+        return new DefaultKafkaConsumerFactory<>(props);
+    }
+
+    @Bean
+    public ConcurrentKafkaListenerContainerFactory<Integer, Object> kafkaListenerContainerFactory(
+            ConsumerFactory<Integer, Object> consumerFactory) {
+        ConcurrentKafkaListenerContainerFactory<Integer, Object> factory = new ConcurrentKafkaListenerContainerFactory<>();
+        factory.setConsumerFactory(consumerFactory);
+        factory.getContainerProperties().setAckMode(ContainerProperties.AckMode.MANUAL);
+        return factory;
+    }
+
+}

fineract-provider/src/main/java/org/apache/fineract/infrastructure/springbatch/messagehandler/spring/SpringEventManagerConfig.java

@@ -16,10 +16,10 @@
  * specific language governing permissions and limitations
  * under the License.
  */
-package org.apache.fineract.infrastructure.springbatch.messagehandler;
+package org.apache.fineract.infrastructure.springbatch.messagehandler.spring;
 
 import org.apache.fineract.infrastructure.springbatch.OutputChannelInterceptor;
-import org.apache.fineract.infrastructure.springbatch.messagehandler.conditions.SpringEventManagerCondition;
+import org.apache.fineract.infrastructure.springbatch.messagehandler.conditions.spring.SpringEventManagerCondition;
 import org.springframework.batch.integration.config.annotation.EnableBatchIntegration;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.context.annotation.Bean;
@@ -27,7 +27,6 @@ import org.springframework.context.annotation.Conditional;
 import org.springframework.context.annotation.Configuration;
 import org.springframework.integration.channel.DirectChannel;
 import org.springframework.integration.dsl.IntegrationFlow;
-import org.springframework.integration.dsl.IntegrationFlows;
 import org.springframework.integration.event.outbound.ApplicationEventPublishingMessageHandler;
 import org.springframework.integration.handler.LoggingHandler;
 
@@ -44,7 +43,7 @@ public class SpringEventManagerConfig {
     @Bean
     public IntegrationFlow outboundFlow() {
         ApplicationEventPublishingMessageHandler handler = new ApplicationEventPublishingMessageHandler();
-        return IntegrationFlows.from(outboundRequests) //
+        return IntegrationFlow.from(outboundRequests) //
                 .intercept(outputInterceptor) //
                 .log(LoggingHandler.Level.DEBUG) //
                 .handle(handler) //

fineract-provider/src/main/java/org/apache/fineract/infrastructure/springbatch/messagehandler/spring/SpringEventWorkerConfig.java

@@ -16,10 +16,10 @@
  * specific language governing permissions and limitations
  * under the License.
  */
-package org.apache.fineract.infrastructure.springbatch.messagehandler;
+package org.apache.fineract.infrastructure.springbatch.messagehandler.spring;
 
 import org.apache.fineract.infrastructure.springbatch.InputChannelInterceptor;
-import org.apache.fineract.infrastructure.springbatch.messagehandler.conditions.SpringEventWorkerCondition;
+import org.apache.fineract.infrastructure.springbatch.messagehandler.conditions.spring.SpringEventWorkerCondition;
 import org.springframework.batch.integration.config.annotation.EnableBatchIntegration;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.context.annotation.Bean;
@@ -27,7 +27,6 @@ import org.springframework.context.annotation.Conditional;
 import org.springframework.context.annotation.Configuration;
 import org.springframework.integration.channel.QueueChannel;
 import org.springframework.integration.dsl.IntegrationFlow;
-import org.springframework.integration.dsl.IntegrationFlows;
 import org.springframework.integration.event.core.MessagingEvent;
 import org.springframework.integration.event.inbound.ApplicationEventListeningMessageProducer;
 import org.springframework.integration.handler.LoggingHandler;
@@ -44,7 +43,7 @@ public class SpringEventWorkerConfig {
 
     @Bean
     public IntegrationFlow inboundFlow() {
-        return IntegrationFlows.from(eventListener()) //
+        return IntegrationFlow.from(eventListener()) //
                 .log(LoggingHandler.Level.DEBUG) //
                 .channel(inboundRequests) //
                 .intercept(inputInterceptor) //

fineract-provider/src/main/resources/application.properties

@@ -74,6 +74,26 @@ fineract.remote-job-message-handler.jms.broker-url=${FINERACT_REMOTE_JOB_MESSAGE
 fineract.remote-job-message-handler.jms.broker-username=${FINERACT_REMOTE_JOB_MESSAGE_HANDLER_JMS_BROKER_USERNAME:}
 fineract.remote-job-message-handler.jms.broker-password=${FINERACT_REMOTE_JOB_MESSAGE_HANDLER_JMS_BROKER_PASSWORD:}
 
+fineract.remote-job-message-handler.kafka.enabled=${FINERACT_REMOTE_JOB_MESSAGE_HANDLER_KAFKA_ENABLED:false}
+fineract.remote-job-message-handler.kafka.topic.auto-create=${FINERACT_REMOTE_JOB_MESSAGE_HANDLER_KAFKA_TOPIC_AUTO_CREATE:true}
+fineract.remote-job-message-handler.kafka.topic.name=${FINERACT_REMOTE_JOB_MESSAGE_HANDLER_KAFKA_TOPIC_NAME:job-topic}
+fineract.remote-job-message-handler.kafka.topic.replicas=${FINERACT_REMOTE_JOB_MESSAGE_HANDLER_KAFKA_TOPIC_REPLICAS:1}
+fineract.remote-job-message-handler.kafka.topic.partitions=${FINERACT_REMOTE_JOB_MESSAGE_HANDLER_KAFKA_TOPIC_PARTITIONS:10}
+fineract.remote-job-message-handler.kafka.bootstrap-servers=${FINERACT_REMOTE_JOB_MESSAGE_HANDLER_KAFKA_BOOTSTRAP_SERVERS:localhost:9092}
+fineract.remote-job-message-handler.kafka.consumer.group-id=${FINERACT_REMOTE_JOB_MESSAGE_HANDLER_KAFKA_CONSUMER_GROUPID:fineract-consumer-group-id}
+#defines the key and value separator, e.g.: key=value
+fineract.remote-job-message-handler.kafka.consumer.extra-properties-key-value-separator=${FINERACT_REMOTE_JOB_MESSAGE_HANDLER_KAFKA_CONSUMER_EXTRA_PROPERTIES_SEPARATOR:=}
+#defines the item separator, e.g.: key1=value1|key2=value2
+fineract.remote-job-message-handler.kafka.consumer.extra-properties-separator=${FINERACT_REMOTE_JOB_MESSAGE_HANDLER_KAFKA_CONSUMER_EXTRA_PROPERTIES_SEPARATOR:|}
+#holds list of key value pairs using the above defined separators:  key1=value1|key2=value2|...|keyn=valuen
+fineract.remote-job-message-handler.kafka.consumer.extra-properties=${FINERACT_REMOTE_JOB_MESSAGE_HANDLER_KAFKA_CONSUMER_EXTRA_PROPERTIES:}
+fineract.remote-job-message-handler.kafka.producer.extra-properties-separator=${FINERACT_REMOTE_JOB_MESSAGE_HANDLER_KAFKA_PRODUCER_EXTRA_PROPERTIES_SEPARATOR:|}
+fineract.remote-job-message-handler.kafka.producer.extra-properties-key-value-separator=${FINERACT_REMOTE_JOB_MESSAGE_HANDLER_KAFKA_PRODUCER_EXTRA_PROPERTIES_SEPARATOR:=}
+fineract.remote-job-message-handler.kafka.producer.extra-properties=${FINERACT_REMOTE_JOB_MESSAGE_HANDLER_KAFKA_PRODUCER_EXTRA_PROPERTIES:}
+fineract.remote-job-message-handler.kafka.admin.extra-properties-separator=${FINERACT_REMOTE_JOB_MESSAGE_HANDLER_KAFKA_ADMIN_EXTRA_PROPERTIES_SEPARATOR:|}
+fineract.remote-job-message-handler.kafka.admin.extra-properties-key-value-separator=${FINERACT_REMOTE_JOB_MESSAGE_HANDLER_KAFKA_ADMIN_EXTRA_PROPERTIES_KEY_VALUE_SEPARATOR:=}
+fineract.remote-job-message-handler.kafka.admin.extra-properties=${FINERACT_REMOTE_JOB_MESSAGE_HANDLER_KAFKA_ADMIN_EXTRA_PROPERTIES:}
+
 fineract.events.external.enabled=${FINERACT_EXTERNAL_EVENTS_ENABLED:false}
 fineract.events.external.producer.jms.enabled=${FINERACT_EXTERNAL_EVENTS_PRODUCER_JMS_ENABLED:false}
 fineract.events.external.producer.jms.async-send-enabled=${FINERACT_EXTERNAL_EVENTS_PRODUCER_JMS_ASYNC_SEND_ENABLED:false}
@@ -86,6 +106,21 @@ fineract.events.external.producer.jms.producer-count=${FINERACT_EXTERNAL_EVENTS_
 fineract.events.external.producer.jms.thread-pool-task-executor-core-pool-size=${FINERACT_EVENT_TASK_EXECUTOR_CORE_POOL_SIZE:10}
 fineract.events.external.producer.jms.thread-pool-task-executor-max-pool-size=${FINERACT_EVENT_TASK_EXECUTOR_MAX_POOL_SIZE:100}
 
+fineract.events.external.producer.kafka.enabled=${FINERACT_EXTERNAL_EVENTS_KAFKA_ENABLED:false}
+fineract.events.external.producer.kafka.timeout-in-seconds=${FINERACT_EXTERNAL_EVENTS_KAFKA_TIMEOUT_IN_SECONDS:10}
+fineract.events.external.producer.kafka.topic.auto-create=${FINERACT_EXTERNAL_EVENTS_KAFKA_TOPIC_AUTO_CREATE:true}
+fineract.events.external.producer.kafka.topic.name=${FINERACT_EXTERNAL_EVENTS_KAFKA_TOPIC_NAME:external-events}
+fineract.events.external.producer.kafka.topic.replicas=${FINERACT_EXTERNAL_EVENTS_KAFKA_TOPIC_REPLICAS:1}
+fineract.events.external.producer.kafka.topic.partitions=${FINERACT_EXTERNAL_EVENTS_KAFKA_TOPIC_PARTITIONS:10}
+fineract.events.external.producer.kafka.bootstrap-servers=${FINERACT_EXTERNAL_EVENTS_KAFKA_BOOTSTRAP_SERVERS:localhost:9092}
+fineract.events.external.producer.kafka.producer.extra-properties-separator=${FINERACT_EXTERNAL_EVENTS_KAFKA_PRODUCER_EXTRA_PROPERTIES_SEPARATOR:|}
+fineract.events.external.producer.kafka.producer.extra-properties-key-value-separator=${FINERACT_EXTERNAL_EVENTS_KAFKA_PRODUCER_EXTRA_PROPERTIES_SEPARATOR:=}
+fineract.events.external.producer.kafka.producer.extra-properties=${FINERACT_EXTERNAL_EVENTS_KAFKA_PRODUCER_EXTRA_PROPERTIES:linger.ms=10|batch.size=16384}
+fineract.events.external.producer.kafka.admin.extra-properties-separator=${FINERACT_EXTERNAL_EVENTS_KAFKA_ADMIN_EXTRA_PROPERTIES_SEPARATOR:|}
+fineract.events.external.producer.kafka.admin.extra-properties-key-value-separator=${FINERACT_EXTERNAL_EVENTS_KAFKA_ADMIN_EXTRA_PROPERTIES_KEY_VALUE_SEPARATOR:=}
+fineract.events.external.producer.kafka.admin.extra-properties=${FINERACT_EXTERNAL_EVENTS_KAFKA_ADMIN_EXTRA_PROPERTIES:}
+
+
 fineract.task-executor.default-task-executor-core-pool-size=${FINERACT_DEFAULT_TASK_EXECUTOR_CORE_POOL_SIZE:10}
 fineract.task-executor.default-task-executor-max-pool-size=${FINERACT_DEFAULT_TASK_EXECUTOR_MAX_POOL_SIZE:100}
 

fineract-provider/src/test/java/org/apache/fineract/infrastructure/event/external/producer/kafka/KafkaExternalEventProducerTest.java

@@ -0,0 +1,137 @@
+/**
+ * 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.
+ */
+package org.apache.fineract.infrastructure.event.external.producer.kafka;
+
+import static org.mockito.Mockito.times;
+
+import java.nio.charset.Charset;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.CompletableFuture;
+import org.apache.fineract.infrastructure.core.config.FineractProperties;
+import org.jetbrains.annotations.NotNull;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.mockito.Mock;
+import org.mockito.Mockito;
+import org.mockito.junit.jupiter.MockitoExtension;
+import org.springframework.kafka.core.KafkaTemplate;
+import org.springframework.kafka.support.SendResult;
+
+@ExtendWith(MockitoExtension.class)
+class KafkaExternalEventProducerTest {
+
+    public static final String TOPIC_NAME = "unit-test";
+    @Mock
+    private KafkaTemplate<Long, byte[]> kafkaTemplate;
+
+    @Mock
+    private SendResult<Long, byte[]> sendResult1;
+
+    @Mock
+    private SendResult<Long, byte[]> sendResult2;
+
+    @Mock
+    private SendResult<Long, byte[]> sendResult3;
+
+    private static final byte[] FIRST = "first".getBytes(Charset.defaultCharset());
+    private static final byte[] SECOND = "second".getBytes(Charset.defaultCharset());
+    private static final byte[] THIRD = "third".getBytes(Charset.defaultCharset());
+
+    @Test
+    public void testSendOK() {
+        // given
+        KafkaExternalEventProducer underTest = new KafkaExternalEventProducer(kafkaTemplate, createProperties());
+        Mockito.when(kafkaTemplate.send(TOPIC_NAME, 1L, FIRST)).thenReturn(CompletableFuture.completedFuture(sendResult1));
+        Mockito.when(kafkaTemplate.send(TOPIC_NAME, 1L, SECOND)).thenReturn(CompletableFuture.completedFuture(sendResult2));
+        Mockito.when(kafkaTemplate.send(TOPIC_NAME, 2L, THIRD)).thenReturn(CompletableFuture.completedFuture(sendResult2));
+
+        // when
+        underTest.sendEvents(Map.of(1L, List.of(FIRST, SECOND), 2L, List.of(THIRD)));
+
+        // then
+        Mockito.verify(kafkaTemplate, times(1)).send(TOPIC_NAME, 1L, FIRST);
+        Mockito.verify(kafkaTemplate, times(1)).send(TOPIC_NAME, 1L, SECOND);
+        Mockito.verify(kafkaTemplate, times(1)).send(TOPIC_NAME, 2L, THIRD);
+        Mockito.verifyNoMoreInteractions(kafkaTemplate);
+    }
+
+    @Test
+    public void testSendOneFails() {
+        // given
+        KafkaExternalEventProducer underTest = new KafkaExternalEventProducer(kafkaTemplate, createProperties());
+        Mockito.when(kafkaTemplate.send(TOPIC_NAME, 1L, FIRST)).thenReturn(CompletableFuture.completedFuture(sendResult1));
+        Mockito.when(kafkaTemplate.send(TOPIC_NAME, 1L, SECOND)).thenReturn(CompletableFuture.completedFuture(sendResult2));
+        Mockito.when(kafkaTemplate.send(TOPIC_NAME, 2L, THIRD))
+                .thenReturn(CompletableFuture.failedFuture(new RuntimeException("Kafka error")));
+
+        // when
+        Assertions.assertThrows(RuntimeException.class, () -> underTest.sendEvents(Map.of(1L, List.of(FIRST, SECOND), 2L, List.of(THIRD))));
+
+        // then
+        Mockito.verify(kafkaTemplate, times(1)).send(TOPIC_NAME, 1L, FIRST);
+        Mockito.verify(kafkaTemplate, times(1)).send(TOPIC_NAME, 1L, SECOND);
+        Mockito.verify(kafkaTemplate, times(1)).send(TOPIC_NAME, 2L, THIRD);
+        Mockito.verifyNoMoreInteractions(kafkaTemplate);
+    }
+
+    @Test
+    public void testTimeOut() {
+        // given
+        KafkaExternalEventProducer underTest = new KafkaExternalEventProducer(kafkaTemplate, createProperties());
+        Mockito.when(kafkaTemplate.send(TOPIC_NAME, 1L, FIRST)).thenReturn(CompletableFuture.completedFuture(sendResult1));
+        Mockito.when(kafkaTemplate.send(TOPIC_NAME, 1L, SECOND)).thenReturn(CompletableFuture.completedFuture(sendResult2));
+        Mockito.when(kafkaTemplate.send(TOPIC_NAME, 2L, THIRD)).thenReturn(new CompletableFuture<>());
+
+        // when
+        Assertions.assertThrows(RuntimeException.class, () -> underTest.sendEvents(Map.of(1L, List.of(FIRST, SECOND), 2L, List.of(THIRD))));
+
+        // then
+        Mockito.verify(kafkaTemplate, times(1)).send(TOPIC_NAME, 1L, FIRST);
+        Mockito.verify(kafkaTemplate, times(1)).send(TOPIC_NAME, 1L, SECOND);
+        Mockito.verify(kafkaTemplate, times(1)).send(TOPIC_NAME, 2L, THIRD);
+        Mockito.verifyNoMoreInteractions(kafkaTemplate);
+    }
+
+    @NotNull
+    private static FineractProperties createProperties() {
+        FineractProperties props = new FineractProperties();
+
+        FineractProperties.FineractEventsProperties fineractEventsProperties = new FineractProperties.FineractEventsProperties();
+        props.setEvents(fineractEventsProperties);
+
+        FineractProperties.FineractExternalEventsProperties externalEventsProperties = new FineractProperties.FineractExternalEventsProperties();
+        fineractEventsProperties.setExternal(externalEventsProperties);
+
+        FineractProperties.FineractExternalEventsProducerProperties producer = new FineractProperties.FineractExternalEventsProducerProperties();
+        externalEventsProperties.setProducer(producer);
+
+        FineractProperties.FineractExternalEventsProducerKafkaProperties kafkaProperties = new FineractProperties.FineractExternalEventsProducerKafkaProperties();
+        producer.setKafka(kafkaProperties);
+
+        FineractProperties.KafkaTopicProperties kafkaTopicProperties = new FineractProperties.KafkaTopicProperties();
+        kafkaProperties.setTopic(kafkaTopicProperties);
+        kafkaProperties.setTimeoutInSeconds(1);
+
+        kafkaTopicProperties.setName(TOPIC_NAME);
+        return props;
+    }
+
+}