FINERACT-1724: Implemented support for multiple parallel JMS producers for external events

fineract-provider/src/main/java/org/apache/fineract/cob/loan/LoanAccountsStayedLockedBusinessEvent.java

@@ -39,4 +39,9 @@ public class LoanAccountsStayedLockedBusinessEvent extends AbstractBusinessEvent
     public String getCategory() {
         return CATEGORY;
     }
+
+    @Override
+    public Long getAggregateRootId() {
+        return null;
+    }
 }

fineract-provider/src/main/java/org/apache/fineract/infrastructure/core/config/FineractProperties.java

@@ -165,6 +165,7 @@ public class FineractProperties {
         private String brokerUrl;
         private String brokerUsername;
         private String brokerPassword;
+        private int producerCount;
 
         public boolean isBrokerPasswordProtected() {
             return StringUtils.isNotBlank(brokerUsername) || StringUtils.isNotBlank(brokerPassword);

fineract-provider/src/main/java/org/apache/fineract/infrastructure/core/messaging/jms/ActiveMQMessageFactory.java

@@ -16,21 +16,20 @@
  * specific language governing permissions and limitations
  * under the License.
  */
-package org.apache.fineract.infrastructure.event.external.config;
+package org.apache.fineract.infrastructure.core.messaging.jms;
 
-import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
+import javax.jms.BytesMessage;
-import org.springframework.context.annotation.Bean;
+import javax.jms.JMSException;
-import org.springframework.context.annotation.Configuration;
+import org.apache.activemq.command.ActiveMQBytesMessage;
-import org.springframework.integration.channel.DirectChannel;
+import org.springframework.stereotype.Component;
-import org.springframework.integration.config.EnableIntegration;
 
-@Configuration
+@Component
-@EnableIntegration
+public class ActiveMQMessageFactory implements MessageFactory {
-@ConditionalOnProperty(value = "fineract.events.external.enabled", havingValue = "true")
-public class ExternalEventProducerConfiguration {
 
-    @Bean
+    @Override
-    public DirectChannel outboundRequestsEvents() {
+    public BytesMessage createByteMessage(byte[] msg) throws JMSException {
-        return new DirectChannel();
+        ActiveMQBytesMessage result = new ActiveMQBytesMessage();
+        result.writeBytes(msg);
+        return result;
     }
 }

fineract-provider/src/main/java/org/apache/fineract/infrastructure/core/messaging/jms/MessageFactory.java

@@ -0,0 +1,27 @@
+/**
+ * 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.core.messaging.jms;
+
+import javax.jms.BytesMessage;
+import javax.jms.JMSException;
+
+public interface MessageFactory {
+
+    BytesMessage createByteMessage(byte[] msg) throws JMSException;
+}

fineract-provider/src/main/java/org/apache/fineract/infrastructure/core/service/HashingService.java

@@ -0,0 +1,30 @@
+/**
+ * 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.core.service;
+
+import com.google.common.hash.Hashing;
+import org.springframework.stereotype.Component;
+
+@Component
+public class HashingService {
+
+    public int consistentHash(long input, int buckets) {
+        return Hashing.consistentHash(input, buckets);
+    }
+}

fineract-provider/src/main/java/org/apache/fineract/infrastructure/core/service/MeasuringUtil.java

@@ -0,0 +1,50 @@
+/**
+ * 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.core.service;
+
+import java.time.Duration;
+import java.util.function.BiConsumer;
+import java.util.function.Consumer;
+import java.util.function.Supplier;
+import org.springframework.util.StopWatch;
+
+public final class MeasuringUtil {
+
+    private MeasuringUtil() {}
+
+    public static void measure(Runnable r, Consumer<Duration> c) {
+        measure(() -> {
+            r.run();
+            return null;
+        }, c);
+    }
+
+    public static <T> T measure(Supplier<T> s, Consumer<Duration> c) {
+        return measure(s, (result, timeTaken) -> c.accept(timeTaken));
+    }
+
+    public static <T> T measure(Supplier<T> s, BiConsumer<T, Duration> c) {
+        StopWatch sw = new StopWatch();
+        sw.start();
+        T result = s.get();
+        sw.stop();
+        c.accept(result, Duration.ofMillis(sw.getTotalTimeMillis()));
+        return result;
+    }
+}

fineract-provider/src/main/java/org/apache/fineract/infrastructure/event/business/domain/BulkBusinessEvent.java

@@ -18,7 +18,11 @@
  */
 package org.apache.fineract.infrastructure.event.business.domain;
 
+import static java.util.stream.Collectors.toSet;
+
 import java.util.List;
+import java.util.Objects;
+import java.util.Set;
 
 public class BulkBusinessEvent extends AbstractBusinessEvent<List<BusinessEvent<?>>> {
 
@@ -27,6 +31,14 @@ public class BulkBusinessEvent extends AbstractBusinessEvent<List<BusinessEvent<
 
     public BulkBusinessEvent(List<BusinessEvent<?>> value) {
         super(value);
+        verifySameAggregate(value);
+    }
+
+    private void verifySameAggregate(List<BusinessEvent<?>> events) {
+        Set<Long> aggregateRootIds = events.stream().map(BusinessEvent::getAggregateRootId).filter(Objects::nonNull).collect(toSet());
+        if (aggregateRootIds.size() > 1) {
+            throw new IllegalArgumentException("The business events are related to multiple aggregate roots which is not allowed");
+        }
     }
 
     @Override
@@ -38,4 +50,9 @@ public class BulkBusinessEvent extends AbstractBusinessEvent<List<BusinessEvent<
     public String getCategory() {
         return CATEGORY;
     }
+
+    @Override
+    public Long getAggregateRootId() {
+        return get().iterator().next().getAggregateRootId();
+    }
 }

fineract-provider/src/main/java/org/apache/fineract/infrastructure/event/business/domain/BusinessEvent.java

@@ -25,4 +25,6 @@ public interface BusinessEvent<T> {
     String getType();
 
     String getCategory();
+
+    Long getAggregateRootId();
 }

fineract-provider/src/main/java/org/apache/fineract/infrastructure/event/business/domain/client/ClientBusinessEvent.java

@@ -33,4 +33,9 @@ public abstract class ClientBusinessEvent extends AbstractBusinessEvent<Client>
     public String getCategory() {
         return CATEGORY;
     }
+
+    @Override
+    public Long getAggregateRootId() {
+        return get().getId();
+    }
 }

fineract-provider/src/main/java/org/apache/fineract/infrastructure/event/business/domain/deposit/FixedDepositAccountBusinessEvent.java

@@ -33,4 +33,9 @@ public abstract class FixedDepositAccountBusinessEvent extends AbstractBusinessE
     public String getCategory() {
         return CATEGORY;
     }
+
+    @Override
+    public Long getAggregateRootId() {
+        return get().getId();
+    }
 }

fineract-provider/src/main/java/org/apache/fineract/infrastructure/event/business/domain/deposit/RecurringDepositAccountBusinessEvent.java

@@ -33,4 +33,9 @@ public abstract class RecurringDepositAccountBusinessEvent extends AbstractBusin
     public String getCategory() {
         return CATEGORY;
     }
+
+    @Override
+    public Long getAggregateRootId() {
+        return get().getId();
+    }
 }

fineract-provider/src/main/java/org/apache/fineract/infrastructure/event/business/domain/group/GroupsBusinessEvent.java

@@ -33,4 +33,9 @@ public abstract class GroupsBusinessEvent extends AbstractBusinessEvent<CommandP
     public String getCategory() {
         return CATEGORY;
     }
+
+    @Override
+    public Long getAggregateRootId() {
+        return get().getResourceId();
+    }
 }

fineract-provider/src/main/java/org/apache/fineract/infrastructure/event/business/domain/loan/LoanAdjustTransactionBusinessEvent.java

@@ -43,6 +43,11 @@ public class LoanAdjustTransactionBusinessEvent extends AbstractBusinessEvent<Lo
         return CATEGORY;
     }
 
+    @Override
+    public Long getAggregateRootId() {
+        return get().getTransactionToAdjust().getLoan().getId();
+    }
+
     @RequiredArgsConstructor
     @Getter
     public static class Data {

fineract-provider/src/main/java/org/apache/fineract/infrastructure/event/business/domain/loan/LoanBusinessEvent.java

@@ -33,4 +33,9 @@ public abstract class LoanBusinessEvent extends AbstractBusinessEvent<Loan> {
     public String getCategory() {
         return CATEGORY;
     }
+
+    @Override
+    public Long getAggregateRootId() {
+        return get().getId();
+    }
 }

fineract-provider/src/main/java/org/apache/fineract/infrastructure/event/business/domain/loan/charge/LoanChargeBusinessEvent.java

@@ -33,4 +33,9 @@ public abstract class LoanChargeBusinessEvent extends AbstractBusinessEvent<Loan
     public String getCategory() {
         return CATEGORY;
     }
+
+    @Override
+    public Long getAggregateRootId() {
+        return get().getLoan().getId();
+    }
 }

fineract-provider/src/main/java/org/apache/fineract/infrastructure/event/business/domain/loan/product/LoanProductBusinessEvent.java

@@ -33,4 +33,9 @@ public abstract class LoanProductBusinessEvent extends AbstractBusinessEvent<Loa
     public String getCategory() {
         return CATEGORY;
     }
+
+    @Override
+    public Long getAggregateRootId() {
+        return get().getId();
+    }
 }

fineract-provider/src/main/java/org/apache/fineract/infrastructure/event/business/domain/loan/repayment/LoanRepaymentBusinessEvent.java

@@ -33,4 +33,9 @@ public abstract class LoanRepaymentBusinessEvent extends AbstractBusinessEvent<L
     public String getCategory() {
         return CATEGORY;
     }
+
+    @Override
+    public Long getAggregateRootId() {
+        return get().getLoan().getId();
+    }
 }

fineract-provider/src/main/java/org/apache/fineract/infrastructure/event/business/domain/loan/transaction/LoanTransactionBusinessEvent.java

@@ -33,4 +33,9 @@ public abstract class LoanTransactionBusinessEvent extends AbstractBusinessEvent
     public String getCategory() {
         return CATEGORY;
     }
+
+    @Override
+    public Long getAggregateRootId() {
+        return get().getLoan().getId();
+    }
 }

fineract-provider/src/main/java/org/apache/fineract/infrastructure/event/business/domain/savings/SavingsAccountBusinessEvent.java

@@ -33,4 +33,9 @@ public abstract class SavingsAccountBusinessEvent extends AbstractBusinessEvent<
     public String getCategory() {
         return CATEGORY;
     }
+
+    @Override
+    public Long getAggregateRootId() {
+        return get().getId();
+    }
 }

fineract-provider/src/main/java/org/apache/fineract/infrastructure/event/business/domain/savings/transaction/SavingsAccountTransactionBusinessEvent.java

@@ -33,4 +33,9 @@ public abstract class SavingsAccountTransactionBusinessEvent extends AbstractBus
     public String getCategory() {
         return CATEGORY;
     }
+
+    @Override
+    public Long getAggregateRootId() {
+        return get().getSavingsAccount().getId();
+    }
 }

fineract-provider/src/main/java/org/apache/fineract/infrastructure/event/business/domain/share/ShareAccountBusinessEvent.java

@@ -33,4 +33,9 @@ public abstract class ShareAccountBusinessEvent extends AbstractBusinessEvent<Sh
     public String getCategory() {
         return CATEGORY;
     }
+
+    @Override
+    public Long getAggregateRootId() {
+        return get().getId();
+    }
 }

fineract-provider/src/main/java/org/apache/fineract/infrastructure/event/business/domain/share/ShareProductDividentsCreateBusinessEvent.java

@@ -38,4 +38,9 @@ public class ShareProductDividentsCreateBusinessEvent extends AbstractBusinessEv
     public String getCategory() {
         return CATEGORY;
     }
+
+    @Override
+    public Long getAggregateRootId() {
+        return get();
+    }
 }

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

@@ -28,10 +28,11 @@ import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Conditional;
 import org.springframework.context.annotation.Configuration;
+import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
 
 @Configuration
 @ConditionalOnProperty(value = "fineract.events.external.producer.jms.enabled", havingValue = "true")
-public class ExternalEventJMSBrokerConfiguration {
+public class ExternalEventJMSConfiguration {
 
     @Autowired
     private FineractProperties fineractProperties;
@@ -60,4 +61,13 @@ public class ExternalEventJMSBrokerConfiguration {
     public ActiveMQQueue activeMqQueue() {
         return new ActiveMQQueue(fineractProperties.getEvents().getExternal().getProducer().getJms().getEventQueueName());
     }
+
+    @Bean("externalEventJmsProducerExecutor")
+    public ThreadPoolTaskExecutor externalEventJmsProducerExecutor() {
+        ThreadPoolTaskExecutor threadPoolTaskExecutor = new ThreadPoolTaskExecutor();
+        threadPoolTaskExecutor.setCorePoolSize(10);
+        threadPoolTaskExecutor.setMaxPoolSize(100);
+        threadPoolTaskExecutor.setThreadNamePrefix("externalEventJms");
+        return threadPoolTaskExecutor;
+    }
 }

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

@@ -1,55 +0,0 @@
-/**
- * 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 javax.jms.ConnectionFactory;
-import javax.jms.Destination;
-import org.apache.fineract.infrastructure.core.config.FineractProperties;
-import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.beans.factory.annotation.Qualifier;
-import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
-import org.springframework.context.annotation.Bean;
-import org.springframework.context.annotation.Configuration;
-import org.springframework.context.annotation.Import;
-import org.springframework.integration.channel.DirectChannel;
-import org.springframework.integration.config.EnableIntegration;
-import org.springframework.integration.dsl.IntegrationFlow;
-import org.springframework.integration.dsl.IntegrationFlows;
-import org.springframework.integration.jms.dsl.Jms;
-
-@Configuration
-@EnableIntegration
-@ConditionalOnProperty(value = "fineract.events.external.producer.jms.enabled", havingValue = "true")
-@Import(value = { ExternalEventJMSBrokerConfiguration.class })
-public class ExternalEventJMSProducerConfiguration {
-
-    @Autowired
-    private DirectChannel outboundRequestsEvents;
-
-    @Autowired
-    private FineractProperties fineractProperties;
-
-    @Bean
-    public IntegrationFlow outboundFlowEvents(ConnectionFactory connectionFactory,
-            @Qualifier("eventDestination") Destination eventDestination) {
-        return IntegrationFlows.from(outboundRequestsEvents) //
-                .handle(Jms.outboundAdapter(connectionFactory).destination(eventDestination)).get();
-    }
-
-}

fineract-provider/src/main/java/org/apache/fineract/infrastructure/event/external/jobs/SendAsynchronousEventsConfig.java

@@ -19,6 +19,7 @@
 package org.apache.fineract.infrastructure.event.external.jobs;
 
 import org.apache.fineract.infrastructure.jobs.service.JobName;
+import org.apache.fineract.infrastructure.jobs.service.StepName;
 import org.springframework.batch.core.Job;
 import org.springframework.batch.core.Step;
 import org.springframework.batch.core.configuration.annotation.JobBuilderFactory;
@@ -40,7 +41,7 @@ public class SendAsynchronousEventsConfig {
 
     @Bean
     protected Step sendAsynchronousEventsStep() {
-        return steps.get(JobName.SEND_ASYNCHRONOUS_EVENTS.name()).tasklet(tasklet).build();
+        return steps.get(StepName.SEND_ASYNCHRONOUS_EVENTS_STEP.name()).tasklet(tasklet).build();
     }
 
     @Bean

fineract-provider/src/main/java/org/apache/fineract/infrastructure/event/external/jobs/SendAsynchronousEventsTasklet.java

@@ -18,8 +18,17 @@
  */
 package org.apache.fineract.infrastructure.event.external.jobs;
 
+import static java.util.stream.Collectors.groupingBy;
+import static java.util.stream.Collectors.toMap;
+import static org.apache.fineract.infrastructure.core.service.MeasuringUtil.measure;
+
+import com.google.common.collect.Lists;
 import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.time.OffsetDateTime;
+import java.util.ArrayList;
 import java.util.List;
+import java.util.Map;
 import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.fineract.avro.MessageV1;
@@ -28,8 +37,8 @@ import org.apache.fineract.infrastructure.core.config.FineractProperties;
 import org.apache.fineract.infrastructure.core.service.DateUtils;
 import org.apache.fineract.infrastructure.event.external.producer.ExternalEventProducer;
 import org.apache.fineract.infrastructure.event.external.repository.ExternalEventRepository;
-import org.apache.fineract.infrastructure.event.external.repository.domain.ExternalEvent;
 import org.apache.fineract.infrastructure.event.external.repository.domain.ExternalEventStatus;
+import org.apache.fineract.infrastructure.event.external.repository.domain.ExternalEventView;
 import org.apache.fineract.infrastructure.event.external.service.message.MessageFactory;
 import org.apache.fineract.infrastructure.event.external.service.support.ByteBufferConverter;
 import org.springframework.batch.core.StepContribution;
@@ -56,8 +65,9 @@ public class SendAsynchronousEventsTasklet implements Tasklet {
     public RepeatStatus execute(StepContribution contribution, ChunkContext chunkContext) {
         try {
             if (isDownstreamChannelEnabled()) {
-                List<ExternalEvent> events = getQueuedEventsBatch();
+                List<ExternalEventView> events = getQueuedEventsBatch();
-                processEvents(events);
+                log.debug("Queued events size: {}", events.size());
+                sendEvents(events);
             }
         } catch (Exception e) {
             log.error("Error occurred while processing events: ", e);
@@ -69,20 +79,66 @@ public class SendAsynchronousEventsTasklet implements Tasklet {
         return fineractProperties.getEvents().getExternal().getProducer().getJms().isEnabled();
     }
 
-    private List<ExternalEvent> getQueuedEventsBatch() {
+    private List<ExternalEventView> getQueuedEventsBatch() {
         int readBatchSize = getBatchSize();
         Pageable batchSize = PageRequest.ofSize(readBatchSize);
-        return repository.findByStatusOrderById(ExternalEventStatus.TO_BE_SENT, batchSize);
+        return measure(() -> repository.findByStatusOrderById(ExternalEventStatus.TO_BE_SENT, batchSize),
+                (events, timeTaken) -> log.debug("Loaded {} events in {}ms", events.size(), timeTaken.toMillis()));
     }
 
-    private void processEvents(List<ExternalEvent> queuedEvents) throws IOException {
+    private void sendEvents(List<ExternalEventView> queuedEvents) {
-        for (ExternalEvent event : queuedEvents) {
+        Map<Long, List<byte[]>> partitions = generatePartitions(queuedEvents);
+        List<Long> eventIds = queuedEvents.stream().map(ExternalEventView::getId).toList();
+        sendEventsToProducer(partitions);
+        markEventsAsSent(eventIds);
+    }
+
+    private void sendEventsToProducer(Map<Long, List<byte[]>> partitions) {
+        eventProducer.sendEvents(partitions);
+    }
+
+    private void markEventsAsSent(List<Long> eventIds) {
+        OffsetDateTime sentAt = DateUtils.getOffsetDateTimeOfTenant();
+
+        // Partitioning dataset to avoid exception: PreparedStatement can have at most 65,535 parameters
+        List<List<Long>> partitions = Lists.partition(eventIds, 5_000);
+        partitions.forEach(partitionedEventIds -> {
+            measure(() -> {
+                repository.markEventsSent(partitionedEventIds, sentAt);
+            }, timeTaken -> {
+                log.debug("Took {}ms to update {} events", timeTaken.toMillis(), partitionedEventIds.size());
+            });
+        });
+    }
+
+    private Map<Long, List<byte[]>> generatePartitions(List<ExternalEventView> queuedEvents) {
+        Map<Long, List<ExternalEventView>> initialPartitions = queuedEvents.stream().collect(groupingBy(externalEvent -> {
+            Long aggregateRootId = externalEvent.getAggregateRootId();
+            if (aggregateRootId == null) {
+                aggregateRootId = -1L;
+            }
+            return aggregateRootId;
+        }));
+        Map<Long, List<byte[]>> partitions = measure(
+                () -> initialPartitions.entrySet().stream().collect(toMap(Map.Entry::getKey, e -> createMessages(e.getValue()))),
+                timeTaken -> {
+                    log.debug("Took {}ms to create message partitions", timeTaken.toMillis());
+                });
+        return partitions;
+    }
+
+    private List<byte[]> createMessages(List<ExternalEventView> events) {
+        try {
+            List<byte[]> messages = new ArrayList<>();
+            for (ExternalEventView event : events) {
                 MessageV1 message = messageFactory.createMessage(event);
-            byte[] byteMessage = byteBufferConverter.convert(message.toByteBuffer());
+                ByteBuffer toByteBuffer = message.toByteBuffer();
-            eventProducer.sendEvent(byteMessage);
+                byte[] convert = byteBufferConverter.convert(toByteBuffer);
-            event.setStatus(ExternalEventStatus.SENT);
+                messages.add(convert);
-            event.setSentAt(DateUtils.getOffsetDateTimeOfTenant());
+            }
-            repository.save(event);
+            return messages;
+        } catch (IOException e) {
+            throw new RuntimeException("Error while serializing the message", e);
         }
     }
 

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

@@ -18,15 +18,11 @@
  */
 package org.apache.fineract.infrastructure.event.external.producer;
 
+import java.util.List;
+import java.util.Map;
 import org.apache.fineract.infrastructure.event.external.exception.AcknowledgementTimeoutException;
-import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
-import org.springframework.integration.annotation.Gateway;
-import org.springframework.integration.annotation.MessagingGateway;
 
-@MessagingGateway(name = "externalEventGateway")
-@ConditionalOnProperty(value = "fineract.events.external.enabled", havingValue = "true")
 public interface ExternalEventProducer {
 
-    @Gateway(requestChannel = "outboundRequestsEvents", replyTimeout = 2, requestTimeout = 200)
+    void sendEvents(Map<Long, List<byte[]>> partitions) throws AcknowledgementTimeoutException;
-    void sendEvent(byte[] message) throws AcknowledgementTimeoutException;
 }

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

@@ -18,16 +18,18 @@
  */
 package org.apache.fineract.infrastructure.event.external.producer;
 
+import java.util.List;
+import java.util.Map;
+import lombok.extern.slf4j.Slf4j;
 import org.apache.fineract.infrastructure.event.external.exception.AcknowledgementTimeoutException;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
-import org.springframework.stereotype.Service;
+import org.springframework.stereotype.Component;
 
-@Service
+@Component
-@ConditionalOnProperty(value = "fineract.events.external.enabled", havingValue = "false")
+@ConditionalOnProperty(value = "fineract.events.external.producer.jms.enabled", havingValue = "false")
-public class DummyExternalEventProducerImpl implements ExternalEventProducer {
+@Slf4j
+public class NoopExternalEventProducer implements ExternalEventProducer {
 
     @Override
-    public void sendEvent(byte[] message) throws AcknowledgementTimeoutException {
+    public void sendEvents(Map<Long, List<byte[]>> messages) throws AcknowledgementTimeoutException {}
-        return;
-    }
 }

fineract-provider/src/main/java/org/apache/fineract/infrastructure/event/external/producer/jms/JMSMultiExternalEventProducer.java

@@ -0,0 +1,153 @@
+/**
+ * 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.jms;
+
+import static org.apache.fineract.infrastructure.core.service.MeasuringUtil.measure;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.Future;
+import javax.jms.Connection;
+import javax.jms.ConnectionFactory;
+import javax.jms.Destination;
+import javax.jms.JMSException;
+import javax.jms.MessageProducer;
+import javax.jms.Session;
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.fineract.infrastructure.core.config.FineractProperties;
+import org.apache.fineract.infrastructure.core.messaging.jms.MessageFactory;
+import org.apache.fineract.infrastructure.core.service.HashingService;
+import org.apache.fineract.infrastructure.event.external.exception.AcknowledgementTimeoutException;
+import org.apache.fineract.infrastructure.event.external.producer.ExternalEventProducer;
+import org.springframework.beans.factory.InitializingBean;
+import org.springframework.beans.factory.annotation.Qualifier;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
+import org.springframework.core.task.AsyncTaskExecutor;
+import org.springframework.stereotype.Component;
+
+@Component
+@Slf4j
+@RequiredArgsConstructor
+@ConditionalOnProperty(value = "fineract.events.external.producer.jms.enabled", havingValue = "true")
+public class JMSMultiExternalEventProducer implements ExternalEventProducer, InitializingBean {
+
+    @Qualifier("eventDestination")
+    private final Destination destination;
+
+    private final ConnectionFactory connectionFactory;
+
+    private final MessageFactory messageFactory;
+
+    @Qualifier("externalEventJmsProducerExecutor")
+    private final AsyncTaskExecutor taskExecutor;
+
+    private final HashingService hashingService;
+
+    private final FineractProperties fineractProperties;
+
+    private final List<MessageProducer> producers = new ArrayList<>();
+
+    @Override
+    public void afterPropertiesSet() throws Exception {
+        int producerCount = getProducerCount();
+        try (Connection connection = connectionFactory.createConnection()) {
+            for (int i = 0; i < producerCount; i++) {
+                // It's crucial to create the session within the loop, otherwise the producers won't be handled as
+                // parallel
+                // producers
+                Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
+                MessageProducer producer = session.createProducer(destination);
+                producers.add(producer);
+            }
+        }
+        log.info("Initialized JMS multi producer for external events with {} parallel producers", producerCount);
+    }
+
+    private int getProducerCount() {
+        return fineractProperties.getEvents().getExternal().getProducer().getJms().getProducerCount();
+    }
+
+    @Override
+    public void sendEvents(Map<Long, List<byte[]>> partitions) throws AcknowledgementTimeoutException {
+        Map<Integer, List<byte[]>> indexedPartitions = mapPartitionsToProducers(partitions);
+        measure(() -> {
+            List<Future<?>> tasks = sendPartitions(indexedPartitions);
+            waitForSendingCompletion(tasks);
+        }, timeTaken -> {
+            if (log.isDebugEnabled()) {
+                // in case execution is faster than 1sec
+                long seconds = Math.max(timeTaken.toSeconds(), 1L);
+                Integer eventCount = partitions.values().stream().map(Collection::size).reduce(0, Integer::sum);
+                log.debug("Sent messages with {} msg/s", (eventCount / seconds));
+            }
+        });
+    }
+
+    private List<Future<?>> sendPartitions(Map<Integer, List<byte[]>> indexedPartitions) {
+        List<Future<?>> tasks = new ArrayList<>();
+        for (Map.Entry<Integer, List<byte[]>> entry : indexedPartitions.entrySet()) {
+            Integer producerIndex = entry.getKey();
+            List<byte[]> messages = entry.getValue();
+            Future<?> future = createSendingTask(producerIndex, messages);
+            tasks.add(future);
+        }
+        return tasks;
+    }
+
+    private Future<?> createSendingTask(Integer producerIndex, List<byte[]> messages) {
+        return taskExecutor.submit(() -> {
+            MessageProducer messageProducer = producers.get(producerIndex);
+
+            for (byte[] message : messages) {
+                try {
+                    messageProducer.send(destination, messageFactory.createByteMessage(message));
+                } catch (JMSException e) {
+                    throw new RuntimeException("Error while sending the message", e);
+                }
+            }
+        });
+    }
+
+    private Map<Integer, List<byte[]>> mapPartitionsToProducers(Map<Long, List<byte[]>> partitions) {
+        Map<Integer, List<byte[]>> indexedPartitions = new HashMap<>();
+        for (Map.Entry<Long, List<byte[]>> partition : partitions.entrySet()) {
+            Long key = partition.getKey();
+            List<byte[]> messages = partition.getValue();
+
+            int producerIndex = hashingService.consistentHash(key, getProducerCount());
+            indexedPartitions.putIfAbsent(producerIndex, new ArrayList<>());
+            indexedPartitions.get(producerIndex).addAll(messages);
+        }
+        return indexedPartitions;
+    }
+
+    private void waitForSendingCompletion(List<Future<?>> tasks) {
+        try {
+            for (Future<?> task : tasks) {
+                task.get();
+            }
+        } catch (Exception e) {
+            throw new RuntimeException(e);
+        }
+    }
+}

fineract-provider/src/main/java/org/apache/fineract/infrastructure/event/external/repository/ExternalEventRepository.java

@@ -19,9 +19,11 @@
 package org.apache.fineract.infrastructure.event.external.repository;
 
 import java.time.LocalDate;
+import java.time.OffsetDateTime;
 import java.util.List;
 import org.apache.fineract.infrastructure.event.external.repository.domain.ExternalEvent;
 import org.apache.fineract.infrastructure.event.external.repository.domain.ExternalEventStatus;
+import org.apache.fineract.infrastructure.event.external.repository.domain.ExternalEventView;
 import org.springframework.data.domain.Pageable;
 import org.springframework.data.jpa.repository.JpaRepository;
 import org.springframework.data.jpa.repository.Modifying;
@@ -29,9 +31,13 @@ import org.springframework.data.jpa.repository.Query;
 
 public interface ExternalEventRepository extends JpaRepository<ExternalEvent, Long> {
 
-    List<ExternalEvent> findByStatusOrderById(ExternalEventStatus status, Pageable batchSize);
+    List<ExternalEventView> findByStatusOrderById(ExternalEventStatus status, Pageable batchSize);
 
     @Modifying(flushAutomatically = true)
     @Query("delete from ExternalEvent e where e.status = :status and e.businessDate <= :dateForPurgeCriteria")
     void deleteOlderEventsWithSentStatus(ExternalEventStatus status, LocalDate dateForPurgeCriteria);
+
+    @Modifying
+    @Query("UPDATE ExternalEvent e SET e.status = org.apache.fineract.infrastructure.event.external.repository.domain.ExternalEventStatus.SENT, e.sentAt = :sentAt WHERE e.id IN :ids")
+    void markEventsSent(List<Long> ids, OffsetDateTime sentAt);
 }

fineract-provider/src/main/java/org/apache/fineract/infrastructure/event/external/repository/domain/ExternalEvent.java

@@ -70,7 +70,10 @@ public class ExternalEvent extends AbstractPersistableCustom {
     @Column(name = "business_date", nullable = false)
     private LocalDate businessDate;
 
-    public ExternalEvent(String type, String category, String schema, byte[] data, String idempotencyKey) {
+    @Column(name = "aggregate_root_id", nullable = true)
+    private Long aggregateRootId;
+
+    public ExternalEvent(String type, String category, String schema, byte[] data, String idempotencyKey, Long aggregateRootId) {
         this.type = type;
         this.category = category;
         this.schema = schema;

fineract-provider/src/main/java/org/apache/fineract/infrastructure/event/external/repository/domain/ExternalEventView.java

@@ -0,0 +1,47 @@
+/**
+ * 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.repository.domain;
+
+import java.time.LocalDate;
+import java.time.OffsetDateTime;
+
+public interface ExternalEventView {
+
+    Long getId();
+
+    String getType();
+
+    String getCategory();
+
+    String getSchema();
+
+    byte[] getData();
+
+    OffsetDateTime getCreatedAt();
+
+    ExternalEventStatus getStatus();
+
+    OffsetDateTime getSentAt();
+
+    String getIdempotencyKey();
+
+    LocalDate getBusinessDate();
+
+    Long getAggregateRootId();
+}

fineract-provider/src/main/java/org/apache/fineract/infrastructure/event/external/service/ExternalEventService.java

@@ -85,7 +85,7 @@ public class ExternalEventService {
         byte[] data = byteBufferConverter.convert(avroDto.toByteBuffer());
 
         return new ExternalEvent(bulkBusinessEvent.getType(), bulkBusinessEvent.getCategory(), BulkMessagePayloadV1.class.getName(), data,
-                idempotencyKey);
+                idempotencyKey, bulkBusinessEvent.getAggregateRootId());
     }
 
     private <T> ExternalEvent handleRegularBusinessEvent(BusinessEvent<T> event) throws IOException {
@@ -95,8 +95,9 @@ public class ExternalEventService {
         BusinessEventSerializer serializer = serializerFactory.create(event);
         String schema = serializer.getSupportedSchema().getName();
         byte[] data = serializer.serialize(event);
+        Long aggregateRootId = event.getAggregateRootId();
 
-        return new ExternalEvent(eventType, eventCategory, schema, data, idempotencyKey);
+        return new ExternalEvent(eventType, eventCategory, schema, data, idempotencyKey, aggregateRootId);
     }
 
     private void flushChangesBeforeSerialization() {

fineract-provider/src/main/java/org/apache/fineract/infrastructure/event/external/service/message/MessageFactory.java

@@ -27,7 +27,7 @@ import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.fineract.avro.MessageV1;
 import org.apache.fineract.infrastructure.core.service.ThreadLocalContextUtil;
-import org.apache.fineract.infrastructure.event.external.repository.domain.ExternalEvent;
+import org.apache.fineract.infrastructure.event.external.repository.domain.ExternalEventView;
 import org.apache.fineract.infrastructure.event.external.service.message.domain.MessageBusinessDate;
 import org.apache.fineract.infrastructure.event.external.service.message.domain.MessageCategory;
 import org.apache.fineract.infrastructure.event.external.service.message.domain.MessageCreatedAt;
@@ -66,7 +66,7 @@ public class MessageFactory implements InitializingBean {
         return result;
     }
 
-    public MessageV1 createMessage(ExternalEvent event) {
+    public MessageV1 createMessage(ExternalEventView event) {
         MessageId id = new MessageId(event.getId().intValue());
         MessageSource source = new MessageSource(SOURCE_UUID);
         MessageType type = new MessageType(event.getType());

fineract-provider/src/main/java/org/apache/fineract/infrastructure/jobs/service/StepName.java

@@ -19,5 +19,5 @@
 package org.apache.fineract.infrastructure.jobs.service;
 
 public enum StepName {
-    PURGE_PROCESSED_COMMANDS_STEP;
+    PURGE_PROCESSED_COMMANDS_STEP, SEND_ASYNCHRONOUS_EVENTS_STEP
 }

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

@@ -62,6 +62,7 @@ fineract.events.external.producer.jms.event-topic-name=${FINERACT_EXTERNAL_EVENT
 fineract.events.external.producer.jms.broker-url=${FINERACT_EXTERNAL_EVENTS_PRODUCER_JMS_BROKER_URL:tcp://127.0.0.1:61616}
 fineract.events.external.producer.jms.broker-username=${FINERACT_EXTERNAL_EVENTS_PRODUCER_JMS_BROKER_USERNAME:}
 fineract.events.external.producer.jms.broker-password=${FINERACT_EXTERNAL_EVENTS_PRODUCER_JMS_BROKER_PASSWORD:}
+fineract.events.external.producer.jms.producer-count=${FINERACT_EXTERNAL_EVENTS_PRODUCER_JMS_PRODUCER_COUNT:1}
 
 fineract.idempotency-key-header-name=${FINERACT_IDEMPOTENCY_KEY_HEADER_NAME:Idempotency-Key}
 

fineract-provider/src/main/resources/db/changelog/tenant/changelog-tenant.xml

@@ -104,4 +104,5 @@
     <include file="parts/0082_add_external_event_default_configuration.xml" relativeToChangelogFile="true" />
     <include file="parts/0083_add_loan_transaction_enum_values.xml" relativeToChangelogFile="true" />
     <include file="parts/0084_add_general_accounting_table_reports.xml" relativeToChangelogFile="true" />
+    <include file="parts/0085_add_aggregate_root_id_external_events.xml" relativeToChangelogFile="true" />
 </databaseChangeLog>

fineract-provider/src/main/resources/db/changelog/tenant/parts/0085_add_aggregate_root_id_external_events.xml

@@ -0,0 +1,32 @@
+<?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.3.xsd">
+    <changeSet author="fineract" id="1">
+        <addColumn tableName="m_external_event">
+            <column name="aggregate_root_id" type="BIGINT">
+                <constraints nullable="true"/>
+            </column>
+        </addColumn>
+    </changeSet>
+</databaseChangeLog>

fineract-provider/src/test/java/org/apache/fineract/infrastructure/core/service/HashingServiceTest.java

@@ -0,0 +1,62 @@
+/**
+ * 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.core.service;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+import java.util.Random;
+import org.junit.jupiter.api.Test;
+
+class HashingServiceTest {
+
+    private final Random rnd = new Random();
+
+    private HashingService underTest = new HashingService();
+
+    @Test
+    public void testConsistentHashGeneratesHashesConsistentlyForTheSameValue() {
+        // given
+        int initialResult = underTest.consistentHash(1L, 20);
+        // when & then
+        for (int i = 0; i < 20_000; i++) {
+            int result = underTest.consistentHash(1L, 20);
+            assertThat(result).isEqualTo(initialResult);
+        }
+    }
+
+    @Test
+    public void testConsistentHashWorksForNegativeValues() {
+        // given
+        // when
+        int result = underTest.consistentHash(-1L, 20);
+        // then
+        assertThat(result).isBetween(0, 19);
+    }
+
+    @Test
+    public void testConsistentHashGeneratesHashesWithinBuckets() {
+        // given
+        int buckets = 10;
+        // when & then
+        for (int i = 0; i < 20_000; i++) {
+            int result = underTest.consistentHash(rnd.nextLong(), buckets);
+            assertThat(result).isLessThan(buckets);
+        }
+    }
+}

fineract-provider/src/test/java/org/apache/fineract/infrastructure/event/business/domain/BulkBusinessEventTest.java

@@ -0,0 +1,64 @@
+/**
+ * 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.business.domain;
+
+import static org.mockito.BDDMockito.given;
+import static org.mockito.Mockito.mock;
+
+import java.util.List;
+import org.apache.fineract.infrastructure.event.business.domain.client.ClientActivateBusinessEvent;
+import org.apache.fineract.infrastructure.event.business.domain.client.ClientCreateBusinessEvent;
+import org.apache.fineract.portfolio.client.domain.Client;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+class BulkBusinessEventTest {
+
+    @Test
+    public void testConstructorWorksForSameAggregateId() {
+        // given
+        // when
+        new BulkBusinessEvent(List.of(new ClientCreateBusinessEvent(client(1L)), new ClientActivateBusinessEvent(client(1L))));
+        // then no exception thrown
+    }
+
+    @Test
+    public void testConstructorThrowsExceptionForDifferentAggregateId() {
+        // given
+        // when
+        Assertions.assertThrows(IllegalArgumentException.class, () -> new BulkBusinessEvent(
+                List.of(new ClientCreateBusinessEvent(client(1L)), new ClientActivateBusinessEvent(client(2L)))));
+        // then no exception thrown
+    }
+
+    @Test
+    public void testConstructorWorksForNullAggregateId() {
+        // given
+        // when
+        new BulkBusinessEvent(List.of(new ClientCreateBusinessEvent(client(1L)), new ClientActivateBusinessEvent(client(null))));
+        // then no exception thrown
+    }
+
+    private Client client(Long id) {
+        Client client = mock(Client.class);
+        given(client.getId()).willReturn(id);
+        return client;
+    }
+
+}

fineract-provider/src/test/java/org/apache/fineract/infrastructure/event/business/service/BusinessEventNotifierServiceImplTest.java

@@ -226,6 +226,11 @@ class BusinessEventNotifierServiceImplTest {
         public String getCategory() {
             return null;
         }
+
+        @Override
+        public Long getAggregateRootId() {
+            return null;
+        }
     }
 
 }

fineract-provider/src/test/java/org/apache/fineract/infrastructure/event/external/jobs/SendAsynchronousEventsTaskletTest.java

@@ -20,7 +20,6 @@ package org.apache.fineract.infrastructure.event.external.jobs;
 
 import static com.google.common.truth.Truth.assertThat;
 import static org.junit.jupiter.api.Assertions.assertEquals;
-import static org.mockito.Mockito.doNothing;
 import static org.mockito.Mockito.doThrow;
 import static org.mockito.Mockito.times;
 import static org.mockito.Mockito.verify;
@@ -34,6 +33,7 @@ import java.util.Arrays;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.Random;
 import org.apache.fineract.avro.MessageV1;
 import org.apache.fineract.infrastructure.businessdate.domain.BusinessDateType;
 import org.apache.fineract.infrastructure.configuration.domain.ConfigurationDomainService;
@@ -43,8 +43,7 @@ import org.apache.fineract.infrastructure.core.service.ThreadLocalContextUtil;
 import org.apache.fineract.infrastructure.event.external.exception.AcknowledgementTimeoutException;
 import org.apache.fineract.infrastructure.event.external.producer.ExternalEventProducer;
 import org.apache.fineract.infrastructure.event.external.repository.ExternalEventRepository;
-import org.apache.fineract.infrastructure.event.external.repository.domain.ExternalEvent;
+import org.apache.fineract.infrastructure.event.external.repository.domain.ExternalEventView;
-import org.apache.fineract.infrastructure.event.external.repository.domain.ExternalEventStatus;
 import org.apache.fineract.infrastructure.event.external.service.message.MessageFactory;
 import org.apache.fineract.infrastructure.event.external.service.support.ByteBufferConverter;
 import org.junit.jupiter.api.BeforeEach;
@@ -54,12 +53,15 @@ import org.mockito.ArgumentCaptor;
 import org.mockito.Mock;
 import org.mockito.Mockito;
 import org.mockito.junit.jupiter.MockitoExtension;
+import org.mockito.junit.jupiter.MockitoSettings;
+import org.mockito.quality.Strictness;
 import org.springframework.batch.core.StepContribution;
 import org.springframework.batch.core.scope.context.ChunkContext;
 import org.springframework.batch.repeat.RepeatStatus;
 import org.springframework.data.domain.Pageable;
 
 @ExtendWith(MockitoExtension.class)
+@MockitoSettings(strictness = Strictness.LENIENT)
 class SendAsynchronousEventsTaskletTest {
 
     @Mock
@@ -81,6 +83,8 @@ class SendAsynchronousEventsTaskletTest {
     private SendAsynchronousEventsTasklet underTest;
     private RepeatStatus resultStatus;
 
+    private Random rnd = new Random();
+
     @BeforeEach
     public void setUp() {
         ThreadLocalContextUtil.setTenant(new FineractPlatformTenant(1L, "default", "Default", "Asia/Kolkata", null));
@@ -108,8 +112,9 @@ class SendAsynchronousEventsTaskletTest {
     @Test
     public void givenBatchSize2WhenTaskExecutionThenSend2Events() throws Exception {
         // given
-        List<ExternalEvent> events = Arrays.asList(new ExternalEvent("aType", "aCategory", "aSchema", new byte[0], "aIdempotencyKey"),
+        List<ExternalEventView> events = Arrays.asList(
-                new ExternalEvent("aType", "aCategory", "aSchema", new byte[0], "aIdempotencyKey"));
+                createExternalEventView("aType", "aCategory", "aSchema", new byte[0], "aIdempotencyKey", 1L),
+                createExternalEventView("aType", "aCategory", "aSchema", new byte[0], "aIdempotencyKey", 1L));
         // Dummy Message
         MessageV1 dummyMessage = new MessageV1(1, "aSource", "aType", "nocategory", "aCreateDate", "aBusinessDate", "aTenantId",
                 "anidempotencyKey", "aSchema", Mockito.mock(ByteBuffer.class));
@@ -117,64 +122,95 @@ class SendAsynchronousEventsTaskletTest {
         when(repository.findByStatusOrderById(Mockito.any(), Mockito.any())).thenReturn(events);
         when(messageFactory.createMessage(Mockito.any())).thenReturn(dummyMessage);
         when(byteBufferConverter.convert(Mockito.any(ByteBuffer.class))).thenReturn(new byte[0]);
-        doNothing().when(eventProducer).sendEvent(Mockito.any());
         // when
-        resultStatus = this.underTest.execute(stepContribution, chunkContext);
+        resultStatus = underTest.execute(stepContribution, chunkContext);
         // then
-        verify(eventProducer, times(2)).sendEvent(new byte[0]);
+        verify(eventProducer).sendEvents(Mockito.any());
-        verify(repository, times(2)).save(Mockito.any(ExternalEvent.class));
+        verify(repository).markEventsSent(Mockito.eq(events.stream().map(ExternalEventView::getId).toList()), Mockito.any());
         assertEquals(RepeatStatus.FINISHED, resultStatus);
     }
 
     @Test
     public void givenBatchSize2WhenEventSendFailsThenExecutionStops() throws Exception {
         // given
-        List<ExternalEvent> events = Arrays.asList(new ExternalEvent("aType", "aCategory", "aSchema", new byte[0], "aIdempotencyKey"),
+        List<ExternalEventView> events = Arrays.asList(
-                new ExternalEvent("aType", "aCategory", "aSchema", new byte[0], "aIdempotencyKey"));
+                createExternalEventView("aType", "aCategory", "aSchema", new byte[0], "aIdempotencyKey", 1L),
+                createExternalEventView("aType", "aCategory", "aSchema", new byte[0], "aIdempotencyKey", 1L));
         MessageV1 dummyMessage = new MessageV1(1, "aSource", "aType", "nocategory", "aCreateDate", "aBusinessDate", "aTenantId",
                 "anidempotencyKey", "aSchema", Mockito.mock(ByteBuffer.class));
         when(repository.findByStatusOrderById(Mockito.any(), Mockito.any())).thenReturn(events);
         when(messageFactory.createMessage(Mockito.any())).thenReturn(dummyMessage);
         when(byteBufferConverter.convert(Mockito.any(ByteBuffer.class))).thenReturn(new byte[0]);
         doThrow(new AcknowledgementTimeoutException("Event Send Exception", new RuntimeException())).when(eventProducer)
-                .sendEvent(Mockito.any());
+                .sendEvents(Mockito.any());
         // when
-        resultStatus = this.underTest.execute(stepContribution, chunkContext);
+        resultStatus = underTest.execute(stepContribution, chunkContext);
         // then
-        verify(repository, times(0)).save(Mockito.any(ExternalEvent.class));
+        verify(repository, times(0)).markEventsSent(Mockito.any(), Mockito.any());
         assertEquals(RepeatStatus.FINISHED, resultStatus);
     }
 
     @Test
     public void givenOneEventWhenEventSentThenEventStatusUpdates() throws Exception {
         // given
-        ArgumentCaptor<ExternalEvent> externalEventArgumentCaptor = ArgumentCaptor.forClass(ExternalEvent.class);
+        List<ExternalEventView> events = Arrays
-        List<ExternalEvent> events = Arrays.asList(new ExternalEvent("aType", "aCategory", "aSchema", new byte[0], "aIdempotencyKey"));
+                .asList(createExternalEventView("aType", "aCategory", "aSchema", new byte[0], "aIdempotencyKey", 1L));
         MessageV1 dummyMessage = new MessageV1(1, "aSource", "aType", "nocategory", "aCreateDate", "aBusinessDate", "aTenantId",
                 "anidempotencyKey", "aSchema", Mockito.mock(ByteBuffer.class));
         when(repository.findByStatusOrderById(Mockito.any(), Mockito.any())).thenReturn(events);
         when(messageFactory.createMessage(Mockito.any())).thenReturn(dummyMessage);
         when(byteBufferConverter.convert(Mockito.any(ByteBuffer.class))).thenReturn(new byte[0]);
-        doNothing().when(eventProducer).sendEvent(Mockito.any());
         // when
-        resultStatus = this.underTest.execute(stepContribution, chunkContext);
+        resultStatus = underTest.execute(stepContribution, chunkContext);
         // then
         verify(messageFactory).createMessage(Mockito.any());
-        verify(repository).save(externalEventArgumentCaptor.capture());
+        verify(eventProducer).sendEvents(Mockito.any());
-        ExternalEvent externalEvent = externalEventArgumentCaptor.getValue();
+        verify(repository).markEventsSent(Mockito.eq(events.stream().map(ExternalEventView::getId).toList()), Mockito.any());
-        assertThat(externalEvent.getStatus()).isEqualTo(ExternalEventStatus.SENT);
+        assertEquals(RepeatStatus.FINISHED, resultStatus);
+    }
+
+    @Test
+    public void testExecuteShouldHandleNullAggregateId() throws Exception {
+        // given
+        List<ExternalEventView> events = Arrays
+                .asList(createExternalEventView("aType", "aCategory", "aSchema", new byte[0], "aIdempotencyKey", null));
+        MessageV1 dummyMessage = new MessageV1(1, "aSource", "aType", "nocategory", "aCreateDate", "aBusinessDate", "aTenantId",
+                "anidempotencyKey", "aSchema", Mockito.mock(ByteBuffer.class));
+        when(repository.findByStatusOrderById(Mockito.any(), Mockito.any())).thenReturn(events);
+        when(messageFactory.createMessage(Mockito.any())).thenReturn(dummyMessage);
+        byte[] byteMsg = new byte[0];
+        when(byteBufferConverter.convert(Mockito.any(ByteBuffer.class))).thenReturn(byteMsg);
+        // when
+        resultStatus = underTest.execute(stepContribution, chunkContext);
+        // then
+        verify(messageFactory).createMessage(Mockito.any());
+        verify(eventProducer).sendEvents(Map.of(-1L, List.of(byteMsg)));
+        verify(repository).markEventsSent(Mockito.eq(events.stream().map(ExternalEventView::getId).toList()), Mockito.any());
         assertEquals(RepeatStatus.FINISHED, resultStatus);
     }
 
     @Test
     public void givenEventBatchSizeIsConfiguredAs10WhenTaskExecutionThenEventReadPageSizeIsCorrect() {
         ArgumentCaptor<Pageable> externalEventPageSizeArgumentCaptor = ArgumentCaptor.forClass(Pageable.class);
-        List<ExternalEvent> events = new ArrayList<>();
+        List<ExternalEventView> events = new ArrayList<>();
         when(repository.findByStatusOrderById(Mockito.any(), Mockito.any())).thenReturn(events);
         // when
-        resultStatus = this.underTest.execute(stepContribution, chunkContext);
+        resultStatus = underTest.execute(stepContribution, chunkContext);
         // then
         verify(repository).findByStatusOrderById(Mockito.any(), externalEventPageSizeArgumentCaptor.capture());
         assertThat(externalEventPageSizeArgumentCaptor.getValue().getPageSize()).isEqualTo(10);
     }
+
+    private ExternalEventView createExternalEventView(String type, String category, String schema, byte[] data, String idempotencyKey,
+            Long aggregateRootId) {
+        ExternalEventView result = Mockito.mock(ExternalEventView.class);
+        Mockito.when(result.getId()).thenReturn(rnd.nextLong());
+        Mockito.when(result.getType()).thenReturn(type);
+        Mockito.when(result.getCategory()).thenReturn(category);
+        Mockito.when(result.getSchema()).thenReturn(schema);
+        Mockito.when(result.getData()).thenReturn(data);
+        Mockito.when(result.getIdempotencyKey()).thenReturn(idempotencyKey);
+        Mockito.when(result.getAggregateRootId()).thenReturn(aggregateRootId);
+        return result;
+    }
 }

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

@@ -1,120 +0,0 @@
-/**
- * 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;
-
-import static org.assertj.core.api.Assertions.assertThat;
-import static org.junit.jupiter.api.Assertions.assertTrue;
-
-import java.util.concurrent.atomic.AtomicInteger;
-import javax.jms.ConnectionFactory;
-import org.apache.activemq.ActiveMQConnectionFactory;
-import org.junit.jupiter.api.Test;
-import org.junit.jupiter.api.extension.ExtendWith;
-import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.beans.factory.annotation.Qualifier;
-import org.springframework.context.annotation.Bean;
-import org.springframework.context.annotation.Configuration;
-import org.springframework.integration.annotation.IntegrationComponentScan;
-import org.springframework.integration.channel.DirectChannel;
-import org.springframework.integration.config.EnableIntegration;
-import org.springframework.integration.config.GlobalChannelInterceptor;
-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;
-import org.springframework.messaging.Message;
-import org.springframework.messaging.MessageChannel;
-import org.springframework.messaging.support.ChannelInterceptor;
-import org.springframework.stereotype.Component;
-import org.springframework.test.annotation.DirtiesContext;
-import org.springframework.test.context.ContextConfiguration;
-import org.springframework.test.context.TestPropertySource;
-import org.springframework.test.context.junit.jupiter.SpringExtension;
-
-@ExtendWith(SpringExtension.class)
-@ContextConfiguration
-@DirtiesContext
-@TestPropertySource(properties = { "fineract.events.external.enabled=true" })
-public class EventsJMSIntegrationTest {
-
-    @Autowired
-    @Qualifier("outboundRequestsEvents")
-    private DirectChannel outboundRequestsEvents;
-
-    @Autowired
-    private ContextConfiguration.TestChannelInterceptor testChannelInterceptor;
-
-    @Autowired
-    private ExternalEventProducer underTest;
-
-    @Test
-    public void testJmsDownstreamChannelIntegration() {
-        assertThat(outboundRequestsEvents.getSubscriberCount()).isEqualTo(1);
-    }
-
-    @Test
-    void given2EventsThenOutBoundChannelIsInvokedTwice() {
-        // when
-        underTest.sendEvent(new byte[0]);
-        underTest.sendEvent(new byte[0]);
-        // then
-        assertTrue(outboundRequestsEvents.getInterceptors().contains(this.testChannelInterceptor));
-        assertThat(testChannelInterceptor.getInvoked()).isEqualTo(2);
-
-    }
-
-    @Configuration
-    @IntegrationComponentScan
-    @EnableIntegration
-    public static class ContextConfiguration {
-
-        private static ConnectionFactory connectionFactory = new ActiveMQConnectionFactory("vm://localhost?broker.persistent=false");
-
-        @Bean
-        public DirectChannel outboundRequestsEvents() {
-            return new DirectChannel();
-        }
-
-        @Bean
-        public IntegrationFlow outboundFlow() {
-            return IntegrationFlows.from("outboundRequestsEvents") //
-                    .log(LoggingHandler.Level.DEBUG) //
-                    .handle(Jms.outboundAdapter(connectionFactory).destination("destinationChannel")).get();
-        }
-
-        @Component
-        @GlobalChannelInterceptor(patterns = "outboundRequestsEvents")
-        public static class TestChannelInterceptor implements ChannelInterceptor {
-
-            private final AtomicInteger invoked = new AtomicInteger();
-
-            public int getInvoked() {
-                return invoked.get();
-            }
-
-            @Override
-            public Message<?> preSend(Message<?> message, MessageChannel channel) {
-                this.invoked.incrementAndGet();
-                return message;
-            }
-
-        }
-
-    }
-}

fineract-provider/src/test/java/org/apache/fineract/infrastructure/event/external/producer/jms/JMSMultiExternalEventProducerTest.java

@@ -0,0 +1,200 @@
+/**
+ * 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.jms;
+
+import static org.mockito.BDDMockito.given;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.Random;
+import javax.jms.BytesMessage;
+import javax.jms.Connection;
+import javax.jms.Destination;
+import javax.jms.JMSException;
+import javax.jms.MessageProducer;
+import javax.jms.Session;
+import org.apache.activemq.ActiveMQConnectionFactory;
+import org.apache.fineract.avro.MessageV1;
+import org.apache.fineract.infrastructure.core.config.FineractProperties;
+import org.apache.fineract.infrastructure.core.messaging.jms.MessageFactory;
+import org.apache.fineract.infrastructure.core.service.HashingService;
+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.mockito.junit.jupiter.MockitoSettings;
+import org.mockito.quality.Strictness;
+import org.springframework.core.task.AsyncTaskExecutor;
+import org.springframework.core.task.SimpleAsyncTaskExecutor;
+
+@ExtendWith(MockitoExtension.class)
+@MockitoSettings(strictness = Strictness.LENIENT)
+class JMSMultiExternalEventProducerTest {
+
+    private static final int PRODUCER_COUNT = 3;
+    private final Random rnd = new Random();
+
+    @Mock
+    private Destination destination;
+    @Mock
+    private ActiveMQConnectionFactory connectionFactory;
+    @Mock
+    private MessageFactory messageFactory;
+
+    @Mock
+    private HashingService hashingService;
+
+    private AsyncTaskExecutor taskExecutor = new SimpleAsyncTaskExecutor();
+
+    @Mock
+    private Connection connection;
+    @Mock
+    private Session session1;
+    @Mock
+    private Session session2;
+    @Mock
+    private Session session3;
+
+    @Mock
+    private MessageProducer producer1;
+    @Mock
+    private MessageProducer producer2;
+    @Mock
+    private MessageProducer producer3;
+
+    private JMSMultiExternalEventProducer underTest;
+
+    private FineractProperties fineractProperties;
+
+    private List<MessageProducer> producers = new ArrayList<>();
+
+    @Test
+    public void testAfterPropertiesShouldCreateMultipleSessions() throws Exception {
+        // given
+        initializeMocks();
+        // when
+        underTest.afterPropertiesSet();
+        // then
+        verify(connection, times(3)).createSession(false, Session.AUTO_ACKNOWLEDGE);
+        verify(session1).createProducer(destination);
+        verify(session2).createProducer(destination);
+        verify(session3).createProducer(destination);
+        verify(connection).close();
+    }
+
+    @Test
+    public void testSendEventsShouldWork() throws Exception {
+        // given
+        initializeMocks();
+        underTest.afterPropertiesSet();
+
+        byte[] msg1 = createMessage();
+        List<byte[]> messages = new ArrayList<>();
+        messages.add(msg1);
+        Map<Long, List<byte[]>> partitions = Map.of(1L, messages);
+
+        BytesMessage bytesMsg1 = Mockito.mock(BytesMessage.class);
+        given(messageFactory.createByteMessage(msg1)).willReturn(bytesMsg1);
+        given(hashingService.consistentHash(1L, producers.size())).willReturn(0);
+        // when
+        underTest.sendEvents(partitions);
+        // then
+        verify(producer1).send(destination, bytesMsg1);
+    }
+
+    @Test
+    public void testSendEventsBalancesBetweenProducers() throws Exception {
+        // given
+        initializeMocks();
+        underTest.afterPropertiesSet();
+
+        byte[] msg1 = createMessage();
+        byte[] msg2 = createMessage();
+        byte[] msg3 = createMessage();
+        List<byte[]> messages1 = new ArrayList<>();
+        messages1.add(msg1);
+        List<byte[]> messages2 = new ArrayList<>();
+        messages2.add(msg2);
+        List<byte[]> messages3 = new ArrayList<>();
+        messages3.add(msg3);
+        Map<Long, List<byte[]>> partitions = Map.of(1L, messages1, 2L, messages2, 3L, messages3);
+
+        BytesMessage bytesMsg1 = Mockito.mock(BytesMessage.class);
+        BytesMessage bytesMsg2 = Mockito.mock(BytesMessage.class);
+        BytesMessage bytesMsg3 = Mockito.mock(BytesMessage.class);
+        given(messageFactory.createByteMessage(msg1)).willReturn(bytesMsg1);
+        given(messageFactory.createByteMessage(msg2)).willReturn(bytesMsg2);
+        given(messageFactory.createByteMessage(msg3)).willReturn(bytesMsg3);
+        given(hashingService.consistentHash(1L, producers.size())).willReturn(0);
+        given(hashingService.consistentHash(2L, producers.size())).willReturn(1);
+        given(hashingService.consistentHash(3L, producers.size())).willReturn(2);
+        // when
+        underTest.sendEvents(partitions);
+        // then
+        verify(producer1).send(destination, bytesMsg1);
+        verify(producer2).send(destination, bytesMsg2);
+        verify(producer3).send(destination, bytesMsg3);
+    }
+
+    private byte[] createMessage() throws IOException {
+        MessageV1 messageV1 = new MessageV1();
+        messageV1.setId(rnd.nextInt());
+        messageV1.setSource("");
+        messageV1.setBusinessDate("");
+        messageV1.setCategory("");
+        messageV1.setCreatedAt("");
+        messageV1.setDataschema("");
+        messageV1.setIdempotencyKey("");
+        messageV1.setTenantId("");
+        messageV1.setType("");
+        messageV1.setData(ByteBuffer.wrap(new byte[0]));
+        return messageV1.toByteBuffer().array();
+    }
+
+    private void initializeMocks() throws JMSException {
+        FineractProperties.FineractExternalEventsProducerJmsProperties jms = new FineractProperties.FineractExternalEventsProducerJmsProperties();
+        jms.setProducerCount(PRODUCER_COUNT);
+        FineractProperties.FineractExternalEventsProducerProperties producer = new FineractProperties.FineractExternalEventsProducerProperties();
+        producer.setJms(jms);
+        FineractProperties.FineractExternalEventsProperties external = new FineractProperties.FineractExternalEventsProperties();
+        external.setProducer(producer);
+        FineractProperties.FineractEventsProperties events = new FineractProperties.FineractEventsProperties();
+        events.setExternal(external);
+        fineractProperties = new FineractProperties();
+        fineractProperties.setEvents(events);
+        underTest = new JMSMultiExternalEventProducer(destination, connectionFactory, messageFactory, taskExecutor, hashingService,
+                fineractProperties);
+
+        given(connectionFactory.createConnection()).willReturn(connection);
+        given(connection.createSession(false, Session.AUTO_ACKNOWLEDGE)).willReturn(session1, session2, session3);
+        given(session1.createProducer(destination)).willReturn(producer1);
+        given(session2.createProducer(destination)).willReturn(producer2);
+        given(session3.createProducer(destination)).willReturn(producer3);
+
+        producers.add(producer1);
+        producers.add(producer2);
+        producers.add(producer3);
+    }
+}

lombok.config

@@ -19,3 +19,5 @@
 config.stopBubbling = true
 
 lombok.addLombokGeneratedAnnotation = true
+
+lombok.copyableAnnotations += org.springframework.beans.factory.annotation.Qualifier