FINERACT-2066: Remove parallelization of external event status update
This commit is contained in:
+1
-7
@@ -34,9 +34,7 @@ import lombok.extern.slf4j.Slf4j;
|
|||||||
import org.apache.fineract.avro.MessageV1;
|
import org.apache.fineract.avro.MessageV1;
|
||||||
import org.apache.fineract.infrastructure.configuration.domain.ConfigurationDomainService;
|
import org.apache.fineract.infrastructure.configuration.domain.ConfigurationDomainService;
|
||||||
import org.apache.fineract.infrastructure.core.config.FineractProperties;
|
import org.apache.fineract.infrastructure.core.config.FineractProperties;
|
||||||
import org.apache.fineract.infrastructure.core.domain.FineractContext;
|
|
||||||
import org.apache.fineract.infrastructure.core.service.DateUtils;
|
import org.apache.fineract.infrastructure.core.service.DateUtils;
|
||||||
import org.apache.fineract.infrastructure.core.service.ThreadLocalContextUtil;
|
|
||||||
import org.apache.fineract.infrastructure.event.external.producer.ExternalEventProducer;
|
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.ExternalEventRepository;
|
||||||
import org.apache.fineract.infrastructure.event.external.repository.domain.ExternalEventStatus;
|
import org.apache.fineract.infrastructure.event.external.repository.domain.ExternalEventStatus;
|
||||||
@@ -106,14 +104,10 @@ public class SendAsynchronousEventsTasklet implements Tasklet {
|
|||||||
// Partitioning dataset to avoid exception: PreparedStatement can have at most 65,535 parameters
|
// Partitioning dataset to avoid exception: PreparedStatement can have at most 65,535 parameters
|
||||||
final int partitionSize = fineractProperties.getEvents().getExternal().getPartitionSize();
|
final int partitionSize = fineractProperties.getEvents().getExternal().getPartitionSize();
|
||||||
List<List<Long>> partitions = Lists.partition(eventIds, partitionSize);
|
List<List<Long>> partitions = Lists.partition(eventIds, partitionSize);
|
||||||
FineractContext context = ThreadLocalContextUtil.getContext();
|
partitions //
|
||||||
partitions.stream() //
|
|
||||||
.parallel() //
|
|
||||||
.forEach(partitionedEventIds -> {
|
.forEach(partitionedEventIds -> {
|
||||||
measure(() -> {
|
measure(() -> {
|
||||||
ThreadLocalContextUtil.init(context);
|
|
||||||
repository.markEventsSent(partitionedEventIds, sentAt);
|
repository.markEventsSent(partitionedEventIds, sentAt);
|
||||||
ThreadLocalContextUtil.reset();
|
|
||||||
}, timeTaken -> {
|
}, timeTaken -> {
|
||||||
log.debug("Took {}ms to update {} events", timeTaken.toMillis(), partitionedEventIds.size());
|
log.debug("Took {}ms to update {} events", timeTaken.toMillis(), partitionedEventIds.size());
|
||||||
});
|
});
|
||||||
|
|||||||
-2
@@ -18,7 +18,6 @@
|
|||||||
*/
|
*/
|
||||||
package org.apache.fineract.infrastructure.event.external.repository;
|
package org.apache.fineract.infrastructure.event.external.repository;
|
||||||
|
|
||||||
import jakarta.transaction.Transactional;
|
|
||||||
import java.time.LocalDate;
|
import java.time.LocalDate;
|
||||||
import java.time.OffsetDateTime;
|
import java.time.OffsetDateTime;
|
||||||
import java.util.List;
|
import java.util.List;
|
||||||
@@ -41,7 +40,6 @@ public interface ExternalEventRepository extends JpaRepository<ExternalEvent, Lo
|
|||||||
void deleteOlderEventsWithSentStatus(@Param("status") ExternalEventStatus status,
|
void deleteOlderEventsWithSentStatus(@Param("status") ExternalEventStatus status,
|
||||||
@Param("dateForPurgeCriteria") LocalDate dateForPurgeCriteria);
|
@Param("dateForPurgeCriteria") LocalDate dateForPurgeCriteria);
|
||||||
|
|
||||||
@Transactional
|
|
||||||
@Modifying
|
@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")
|
@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(@Param("ids") List<Long> ids, @Param("sentAt") OffsetDateTime sentAt);
|
void markEventsSent(@Param("ids") List<Long> ids, @Param("sentAt") OffsetDateTime sentAt);
|
||||||
|
|||||||
Reference in New Issue
Block a user