You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@fineract.apache.org by "magyari-adam (via GitHub)" <gi...@apache.org> on 2024/11/13 14:59:14 UTC
[PR] FINERACT-2081: External event sending make mark as sent async [fineract]
magyari-adam opened a new pull request, #4164:
URL: https://github.com/apache/fineract/pull/4164
## Description
Enchancement for external event sending job, making the marking of events as sent async.
## Checklist
Please make sure these boxes are checked before submitting your pull request - thanks!
- [x] Write the commit message as per https://github.com/apache/fineract/#pull-requests
- [x] Acknowledge that we will not review PRs that are not passing the build _("green")_ - it is your responsibility to get a proposed PR to pass the build, not primarily the project's maintainers.
- [x] Create/update unit or integration tests for verifying the changes made.
- [x] Follow coding conventions at https://cwiki.apache.org/confluence/display/FINERACT/Coding+Conventions.
- [x] Add required Swagger annotation and update API documentation at fineract-provider/src/main/resources/static/legacy-docs/apiLive.htm with details of any API changes
- [x] Submission is not a "code dump". (Large changes can be made "in repository" via a branch. Ask on the developer mailing list for guidance, if required.)
FYI our guidelines for code reviews are at https://cwiki.apache.org/confluence/display/FINERACT/Code+Review+Guide.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@fineract.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] FINERACT-2081: External event sending make mark as sent async [fineract]
Posted by "KnSup (via GitHub)" <gi...@apache.org>.
KnSup commented on PR #4164:
URL: https://github.com/apache/fineract/pull/4164#issuecomment-2542690600
Hello, I am currently learning this part of the code, and I would like to ask: if multiple instances of microservice web services are started, will the SendAsynchronousEventsTasklet also start multiple times and run continuously, ultimately leading to multiple sends?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@fineract.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] FINERACT-2081: External event sending make mark as sent async [fineract]
Posted by "adamsaghy (via GitHub)" <gi...@apache.org>.
adamsaghy commented on code in PR #4164:
URL: https://github.com/apache/fineract/pull/4164#discussion_r1840575907
##########
fineract-core/src/main/java/org/apache/fineract/infrastructure/event/external/jobs/SendAsynchronousEventsTasklet.java:
##########
@@ -104,14 +111,40 @@ private void markEventsAsSent(List<Long> eventIds) {
// Partitioning dataset to avoid exception: PreparedStatement can have at most 65,535 parameters
final int partitionSize = fineractProperties.getEvents().getExternal().getPartitionSize();
List<List<Long>> partitions = Lists.partition(eventIds, partitionSize);
+ final ThreadPoolTaskExecutor executor = getThreadPoolTaskExecutor();
+ final FineractContext context = ThreadLocalContextUtil.getContext();
+ List<Future<?>> tasks = new ArrayList<>();
partitions //
.forEach(partitionedEventIds -> {
- measure(() -> {
- repository.markEventsSent(partitionedEventIds, sentAt);
- }, timeTaken -> {
- log.debug("Took {}ms to update {} events", timeTaken.toMillis(), partitionedEventIds.size());
- });
+ tasks.add(executor.submit(() -> {
+ measure(() -> {
+ ThreadLocalContextUtil.init(context);
+ repository.markEventsSent(partitionedEventIds, sentAt);
+ }, timeTaken -> {
+ log.debug("Took {}ms to update {} events", timeTaken.toMillis(), partitionedEventIds.size());
+ });
+ }));
});
+ tasks.forEach(task -> {
+ try {
+ task.get();
+ } catch (InterruptedException | ExecutionException e) {
+ throw new RuntimeException(e);
+ }
+ });
+ }
+
+ @NotNull
+ private ThreadPoolTaskExecutor getThreadPoolTaskExecutor() {
Review Comment:
Please move configuration into application.properties!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@fineract.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] FINERACT-2081: External event sending make mark as sent async [fineract]
Posted by "adamsaghy (via GitHub)" <gi...@apache.org>.
adamsaghy commented on code in PR #4164:
URL: https://github.com/apache/fineract/pull/4164#discussion_r1842425932
##########
fineract-provider/src/main/resources/application.properties:
##########
@@ -99,7 +99,10 @@ fineract.remote-job-message-handler.kafka.admin.extra-properties-key-value-separ
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.partition-size=${FINERACT_EXTERNAL_EVENTS_PARTITION_SIZE:5000}
+fineract.events.external.partition-size=${FINERACT_EXTERNAL_EVENTS_PARTITION_SIZE:5}
Review Comment:
5 is a little too low ;)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@fineract.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] FINERACT-2081: External event sending make mark as sent async [fineract]
Posted by "adamsaghy (via GitHub)" <gi...@apache.org>.
adamsaghy commented on code in PR #4164:
URL: https://github.com/apache/fineract/pull/4164#discussion_r1840713075
##########
fineract-core/src/main/java/org/apache/fineract/infrastructure/event/external/jobs/SendAsynchronousEventsTasklet.java:
##########
@@ -104,14 +111,40 @@ private void markEventsAsSent(List<Long> eventIds) {
// Partitioning dataset to avoid exception: PreparedStatement can have at most 65,535 parameters
final int partitionSize = fineractProperties.getEvents().getExternal().getPartitionSize();
List<List<Long>> partitions = Lists.partition(eventIds, partitionSize);
+ final ThreadPoolTaskExecutor executor = getThreadPoolTaskExecutor();
+ final FineractContext context = ThreadLocalContextUtil.getContext();
+ List<Future<?>> tasks = new ArrayList<>();
partitions //
.forEach(partitionedEventIds -> {
Review Comment:
Have you considered parallel stream instead of foreach?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@fineract.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] FINERACT-2081: External event sending make mark as sent async [fineract]
Posted by "galovics (via GitHub)" <gi...@apache.org>.
galovics commented on PR #4164:
URL: https://github.com/apache/fineract/pull/4164#issuecomment-2543034183
@KnSup unlikely but possible since there's a tiny window where multiple instances could see (tx isolation) events to be sent. In case you're running in a multi-instance mode, you should have a separate instance to run the batch jobs.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@fineract.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] FINERACT-2081: External event sending make mark as sent async [fineract]
Posted by "magyari-adam (via GitHub)" <gi...@apache.org>.
magyari-adam commented on code in PR #4164:
URL: https://github.com/apache/fineract/pull/4164#discussion_r1850196979
##########
fineract-core/src/main/java/org/apache/fineract/infrastructure/event/external/jobs/SendAsynchronousEventsTasklet.java:
##########
@@ -104,16 +109,34 @@ private void markEventsAsSent(List<Long> eventIds) {
// Partitioning dataset to avoid exception: PreparedStatement can have at most 65,535 parameters
final int partitionSize = fineractProperties.getEvents().getExternal().getPartitionSize();
List<List<Long>> partitions = Lists.partition(eventIds, partitionSize);
+ final ThreadPoolTaskExecutor executor = getThreadPoolTaskExecutor();
+ final FineractContext context = ThreadLocalContextUtil.getContext();
partitions //
.forEach(partitionedEventIds -> {
- measure(() -> {
- repository.markEventsSent(partitionedEventIds, sentAt);
- }, timeTaken -> {
- log.debug("Took {}ms to update {} events", timeTaken.toMillis(), partitionedEventIds.size());
+ executor.execute(() -> {
+ measure(() -> {
+ ThreadLocalContextUtil.init(context);
+ repository.markEventsSent(partitionedEventIds, sentAt);
+ }, timeTaken -> {
+ log.debug("Took {}ms to update {} events", timeTaken.toMillis(), partitionedEventIds.size());
+ });
});
});
}
+ @Nonnull
+ private ThreadPoolTaskExecutor getThreadPoolTaskExecutor() {
+ if (threadPoolTaskExecutor == null) {
+ threadPoolTaskExecutor = new ThreadPoolTaskExecutor();
+ threadPoolTaskExecutor.setCorePoolSize(fineractProperties.getEvents().getExternal().getThreadPoolCorePoolSize());
+ threadPoolTaskExecutor.setMaxPoolSize(fineractProperties.getEvents().getExternal().getThreadPoolMaxPoolSize());
+ threadPoolTaskExecutor.setQueueCapacity(fineractProperties.getEvents().getExternal().getThreadPoolQueueCapacity());
+ threadPoolTaskExecutor.setThreadNamePrefix("Fineract-external-event-");
Review Comment:
sure
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@fineract.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] FINERACT-2081: External event sending make mark as sent async [fineract]
Posted by "magyari-adam (via GitHub)" <gi...@apache.org>.
magyari-adam commented on code in PR #4164:
URL: https://github.com/apache/fineract/pull/4164#discussion_r1850198947
##########
fineract-core/src/main/java/org/apache/fineract/infrastructure/event/external/jobs/SendAsynchronousEventsTasklet.java:
##########
@@ -104,16 +109,34 @@ private void markEventsAsSent(List<Long> eventIds) {
// Partitioning dataset to avoid exception: PreparedStatement can have at most 65,535 parameters
final int partitionSize = fineractProperties.getEvents().getExternal().getPartitionSize();
List<List<Long>> partitions = Lists.partition(eventIds, partitionSize);
+ final ThreadPoolTaskExecutor executor = getThreadPoolTaskExecutor();
+ final FineractContext context = ThreadLocalContextUtil.getContext();
partitions //
.forEach(partitionedEventIds -> {
- measure(() -> {
- repository.markEventsSent(partitionedEventIds, sentAt);
- }, timeTaken -> {
- log.debug("Took {}ms to update {} events", timeTaken.toMillis(), partitionedEventIds.size());
+ executor.execute(() -> {
+ measure(() -> {
+ ThreadLocalContextUtil.init(context);
+ repository.markEventsSent(partitionedEventIds, sentAt);
+ }, timeTaken -> {
+ log.debug("Took {}ms to update {} events", timeTaken.toMillis(), partitionedEventIds.size());
+ });
});
});
}
+ @Nonnull
+ private ThreadPoolTaskExecutor getThreadPoolTaskExecutor() {
Review Comment:
Couldn't move it to the tasklet config as it would create a circular dependency between them, moved it to it's own config.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@fineract.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] FINERACT-2081: External event sending make mark as sent async [fineract]
Posted by "adamsaghy (via GitHub)" <gi...@apache.org>.
adamsaghy commented on PR #4164:
URL: https://github.com/apache/fineract/pull/4164#issuecomment-2483136288
@magyari-adam please squash your commits!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@fineract.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] FINERACT-2081: External event sending make mark as sent async [fineract]
Posted by "magyari-adam (via GitHub)" <gi...@apache.org>.
magyari-adam commented on code in PR #4164:
URL: https://github.com/apache/fineract/pull/4164#discussion_r1840862445
##########
fineract-core/src/main/java/org/apache/fineract/infrastructure/event/external/jobs/SendAsynchronousEventsTasklet.java:
##########
@@ -104,14 +111,40 @@ private void markEventsAsSent(List<Long> eventIds) {
// Partitioning dataset to avoid exception: PreparedStatement can have at most 65,535 parameters
final int partitionSize = fineractProperties.getEvents().getExternal().getPartitionSize();
List<List<Long>> partitions = Lists.partition(eventIds, partitionSize);
+ final ThreadPoolTaskExecutor executor = getThreadPoolTaskExecutor();
+ final FineractContext context = ThreadLocalContextUtil.getContext();
+ List<Future<?>> tasks = new ArrayList<>();
partitions //
.forEach(partitionedEventIds -> {
- measure(() -> {
- repository.markEventsSent(partitionedEventIds, sentAt);
- }, timeTaken -> {
- log.debug("Took {}ms to update {} events", timeTaken.toMillis(), partitionedEventIds.size());
- });
+ tasks.add(executor.submit(() -> {
+ measure(() -> {
+ ThreadLocalContextUtil.init(context);
+ repository.markEventsSent(partitionedEventIds, sentAt);
+ }, timeTaken -> {
+ log.debug("Took {}ms to update {} events", timeTaken.toMillis(), partitionedEventIds.size());
+ });
+ }));
});
+ tasks.forEach(task -> {
+ try {
+ task.get();
+ } catch (InterruptedException | ExecutionException e) {
+ throw new RuntimeException(e);
+ }
+ });
+ }
+
+ @NotNull
+ private ThreadPoolTaskExecutor getThreadPoolTaskExecutor() {
Review Comment:
You're right, moved it to config.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@fineract.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] FINERACT-2081: External event sending make mark as sent async [fineract]
Posted by "adamsaghy (via GitHub)" <gi...@apache.org>.
adamsaghy commented on code in PR #4164:
URL: https://github.com/apache/fineract/pull/4164#discussion_r1840715180
##########
fineract-core/src/main/java/org/apache/fineract/infrastructure/event/external/jobs/SendAsynchronousEventsTasklet.java:
##########
@@ -104,14 +111,40 @@ private void markEventsAsSent(List<Long> eventIds) {
// Partitioning dataset to avoid exception: PreparedStatement can have at most 65,535 parameters
final int partitionSize = fineractProperties.getEvents().getExternal().getPartitionSize();
List<List<Long>> partitions = Lists.partition(eventIds, partitionSize);
+ final ThreadPoolTaskExecutor executor = getThreadPoolTaskExecutor();
+ final FineractContext context = ThreadLocalContextUtil.getContext();
+ List<Future<?>> tasks = new ArrayList<>();
partitions //
.forEach(partitionedEventIds -> {
- measure(() -> {
- repository.markEventsSent(partitionedEventIds, sentAt);
- }, timeTaken -> {
- log.debug("Took {}ms to update {} events", timeTaken.toMillis(), partitionedEventIds.size());
- });
+ tasks.add(executor.submit(() -> {
+ measure(() -> {
+ ThreadLocalContextUtil.init(context);
+ repository.markEventsSent(partitionedEventIds, sentAt);
+ }, timeTaken -> {
+ log.debug("Took {}ms to update {} events", timeTaken.toMillis(), partitionedEventIds.size());
+ });
+ }));
});
+ tasks.forEach(task -> {
Review Comment:
The problem is the `task.get` is blocking call and we dont really need the result...
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@fineract.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] FINERACT-2081: External event sending make mark as sent async [fineract]
Posted by "magyari-adam (via GitHub)" <gi...@apache.org>.
magyari-adam commented on code in PR #4164:
URL: https://github.com/apache/fineract/pull/4164#discussion_r1850201063
##########
fineract-core/src/main/java/org/apache/fineract/infrastructure/event/external/repository/ExternalEventRepository.java:
##########
@@ -40,6 +41,7 @@ public interface ExternalEventRepository extends JpaRepository<ExternalEvent, Lo
void deleteOlderEventsWithSentStatus(@Param("status") ExternalEventStatus status,
@Param("dateForPurgeCriteria") LocalDate dateForPurgeCriteria);
+ @Transactional
Review Comment:
hmm, yea you're right, thanks for the tip with TrancactionTemplate
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@fineract.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] FINERACT-2081: External event sending make mark as sent async [fineract]
Posted by "galovics (via GitHub)" <gi...@apache.org>.
galovics merged PR #4164:
URL: https://github.com/apache/fineract/pull/4164
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@fineract.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] FINERACT-2081: External event sending make mark as sent async [fineract]
Posted by "janez89 (via GitHub)" <gi...@apache.org>.
janez89 commented on code in PR #4164:
URL: https://github.com/apache/fineract/pull/4164#discussion_r1843439507
##########
fineract-core/src/main/java/org/apache/fineract/infrastructure/event/external/jobs/SendAsynchronousEventsTasklet.java:
##########
@@ -34,19 +34,23 @@
import org.apache.fineract.avro.MessageV1;
import org.apache.fineract.infrastructure.configuration.domain.ConfigurationDomainService;
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.ThreadLocalContextUtil;
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.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.jetbrains.annotations.NotNull;
Review Comment:
Use Jakarta EE annotations, pls
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@fineract.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] FINERACT-2081: External event sending make mark as sent async [fineract]
Posted by "magyari-adam (via GitHub)" <gi...@apache.org>.
magyari-adam commented on code in PR #4164:
URL: https://github.com/apache/fineract/pull/4164#discussion_r1840883948
##########
fineract-core/src/main/java/org/apache/fineract/infrastructure/event/external/jobs/SendAsynchronousEventsTasklet.java:
##########
@@ -104,14 +111,40 @@ private void markEventsAsSent(List<Long> eventIds) {
// Partitioning dataset to avoid exception: PreparedStatement can have at most 65,535 parameters
final int partitionSize = fineractProperties.getEvents().getExternal().getPartitionSize();
List<List<Long>> partitions = Lists.partition(eventIds, partitionSize);
+ final ThreadPoolTaskExecutor executor = getThreadPoolTaskExecutor();
+ final FineractContext context = ThreadLocalContextUtil.getContext();
+ List<Future<?>> tasks = new ArrayList<>();
Review Comment:
Gotcha, fixed.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@fineract.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] FINERACT-2081: External event sending make mark as sent async [fineract]
Posted by "galovics (via GitHub)" <gi...@apache.org>.
galovics commented on code in PR #4164:
URL: https://github.com/apache/fineract/pull/4164#discussion_r1846827337
##########
fineract-core/src/main/java/org/apache/fineract/infrastructure/event/external/jobs/SendAsynchronousEventsTasklet.java:
##########
@@ -104,16 +109,34 @@ private void markEventsAsSent(List<Long> eventIds) {
// Partitioning dataset to avoid exception: PreparedStatement can have at most 65,535 parameters
final int partitionSize = fineractProperties.getEvents().getExternal().getPartitionSize();
List<List<Long>> partitions = Lists.partition(eventIds, partitionSize);
+ final ThreadPoolTaskExecutor executor = getThreadPoolTaskExecutor();
+ final FineractContext context = ThreadLocalContextUtil.getContext();
partitions //
.forEach(partitionedEventIds -> {
- measure(() -> {
- repository.markEventsSent(partitionedEventIds, sentAt);
- }, timeTaken -> {
- log.debug("Took {}ms to update {} events", timeTaken.toMillis(), partitionedEventIds.size());
+ executor.execute(() -> {
+ measure(() -> {
+ ThreadLocalContextUtil.init(context);
+ repository.markEventsSent(partitionedEventIds, sentAt);
+ }, timeTaken -> {
+ log.debug("Took {}ms to update {} events", timeTaken.toMillis(), partitionedEventIds.size());
+ });
});
});
}
+ @Nonnull
+ private ThreadPoolTaskExecutor getThreadPoolTaskExecutor() {
+ if (threadPoolTaskExecutor == null) {
+ threadPoolTaskExecutor = new ThreadPoolTaskExecutor();
+ threadPoolTaskExecutor.setCorePoolSize(fineractProperties.getEvents().getExternal().getThreadPoolCorePoolSize());
+ threadPoolTaskExecutor.setMaxPoolSize(fineractProperties.getEvents().getExternal().getThreadPoolMaxPoolSize());
+ threadPoolTaskExecutor.setQueueCapacity(fineractProperties.getEvents().getExternal().getThreadPoolQueueCapacity());
+ threadPoolTaskExecutor.setThreadNamePrefix("Fineract-external-event-");
Review Comment:
let's just name it external-events-
##########
fineract-core/src/main/java/org/apache/fineract/infrastructure/event/external/jobs/SendAsynchronousEventsTasklet.java:
##########
@@ -104,16 +109,34 @@ private void markEventsAsSent(List<Long> eventIds) {
// Partitioning dataset to avoid exception: PreparedStatement can have at most 65,535 parameters
final int partitionSize = fineractProperties.getEvents().getExternal().getPartitionSize();
List<List<Long>> partitions = Lists.partition(eventIds, partitionSize);
+ final ThreadPoolTaskExecutor executor = getThreadPoolTaskExecutor();
+ final FineractContext context = ThreadLocalContextUtil.getContext();
partitions //
.forEach(partitionedEventIds -> {
- measure(() -> {
- repository.markEventsSent(partitionedEventIds, sentAt);
- }, timeTaken -> {
- log.debug("Took {}ms to update {} events", timeTaken.toMillis(), partitionedEventIds.size());
+ executor.execute(() -> {
+ measure(() -> {
+ ThreadLocalContextUtil.init(context);
+ repository.markEventsSent(partitionedEventIds, sentAt);
+ }, timeTaken -> {
+ log.debug("Took {}ms to update {} events", timeTaken.toMillis(), partitionedEventIds.size());
+ });
});
});
}
+ @Nonnull
+ private ThreadPoolTaskExecutor getThreadPoolTaskExecutor() {
Review Comment:
Don't create this inside the Tasklet. Extract it to the tasklet's config (that's why it's there) and have the threadpool configured there as a bean.
##########
fineract-core/src/main/java/org/apache/fineract/infrastructure/event/external/repository/ExternalEventRepository.java:
##########
@@ -40,6 +41,7 @@ public interface ExternalEventRepository extends JpaRepository<ExternalEvent, Lo
void deleteOlderEventsWithSentStatus(@Param("status") ExternalEventStatus status,
@Param("dateForPurgeCriteria") LocalDate dateForPurgeCriteria);
+ @Transactional
Review Comment:
frankly speaking, defining the transactional boundary on the repository level is often a code-smell. With transactionality you want to have atomicity across an operation, not a single repository call. I'd move this one level up (and since you're using a lambda there, I'd use TransactionTemplate for it)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@fineract.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] FINERACT-2081: External event sending make mark as sent async [fineract]
Posted by "adamsaghy (via GitHub)" <gi...@apache.org>.
adamsaghy commented on code in PR #4164:
URL: https://github.com/apache/fineract/pull/4164#discussion_r1840575907
##########
fineract-core/src/main/java/org/apache/fineract/infrastructure/event/external/jobs/SendAsynchronousEventsTasklet.java:
##########
@@ -104,14 +111,40 @@ private void markEventsAsSent(List<Long> eventIds) {
// Partitioning dataset to avoid exception: PreparedStatement can have at most 65,535 parameters
final int partitionSize = fineractProperties.getEvents().getExternal().getPartitionSize();
List<List<Long>> partitions = Lists.partition(eventIds, partitionSize);
+ final ThreadPoolTaskExecutor executor = getThreadPoolTaskExecutor();
+ final FineractContext context = ThreadLocalContextUtil.getContext();
+ List<Future<?>> tasks = new ArrayList<>();
partitions //
.forEach(partitionedEventIds -> {
- measure(() -> {
- repository.markEventsSent(partitionedEventIds, sentAt);
- }, timeTaken -> {
- log.debug("Took {}ms to update {} events", timeTaken.toMillis(), partitionedEventIds.size());
- });
+ tasks.add(executor.submit(() -> {
+ measure(() -> {
+ ThreadLocalContextUtil.init(context);
+ repository.markEventsSent(partitionedEventIds, sentAt);
+ }, timeTaken -> {
+ log.debug("Took {}ms to update {} events", timeTaken.toMillis(), partitionedEventIds.size());
+ });
+ }));
});
+ tasks.forEach(task -> {
+ try {
+ task.get();
+ } catch (InterruptedException | ExecutionException e) {
+ throw new RuntimeException(e);
+ }
+ });
+ }
+
+ @NotNull
+ private ThreadPoolTaskExecutor getThreadPoolTaskExecutor() {
Review Comment:
Not nice. Please move configuration into application.properties!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@fineract.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] FINERACT-2081: External event sending make mark as sent async [fineract]
Posted by "magyari-adam (via GitHub)" <gi...@apache.org>.
magyari-adam commented on code in PR #4164:
URL: https://github.com/apache/fineract/pull/4164#discussion_r1840662152
##########
fineract-core/src/main/java/org/apache/fineract/infrastructure/event/external/repository/ExternalEventRepository.java:
##########
@@ -40,6 +41,7 @@ public interface ExternalEventRepository extends JpaRepository<ExternalEvent, Lo
void deleteOlderEventsWithSentStatus(@Param("status") ExternalEventStatus status,
@Param("dateForPurgeCriteria") LocalDate dateForPurgeCriteria);
+ @Transactional
Review Comment:
In the new thread it needs a transaction.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@fineract.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] FINERACT-2081: External event sending make mark as sent async [fineract]
Posted by "adamsaghy (via GitHub)" <gi...@apache.org>.
adamsaghy commented on code in PR #4164:
URL: https://github.com/apache/fineract/pull/4164#discussion_r1840709493
##########
fineract-core/src/main/java/org/apache/fineract/infrastructure/event/external/jobs/SendAsynchronousEventsTasklet.java:
##########
@@ -104,14 +111,40 @@ private void markEventsAsSent(List<Long> eventIds) {
// Partitioning dataset to avoid exception: PreparedStatement can have at most 65,535 parameters
final int partitionSize = fineractProperties.getEvents().getExternal().getPartitionSize();
List<List<Long>> partitions = Lists.partition(eventIds, partitionSize);
+ final ThreadPoolTaskExecutor executor = getThreadPoolTaskExecutor();
+ final FineractContext context = ThreadLocalContextUtil.getContext();
+ List<Future<?>> tasks = new ArrayList<>();
Review Comment:
We dont need to wait on the result. Please use `execute` instead of `submit` + get()
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@fineract.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] FINERACT-2081: External event sending make mark as sent async [fineract]
Posted by "magyari-adam (via GitHub)" <gi...@apache.org>.
magyari-adam commented on code in PR #4164:
URL: https://github.com/apache/fineract/pull/4164#discussion_r1843845978
##########
fineract-core/src/main/java/org/apache/fineract/infrastructure/event/external/jobs/SendAsynchronousEventsTasklet.java:
##########
@@ -34,19 +34,23 @@
import org.apache.fineract.avro.MessageV1;
import org.apache.fineract.infrastructure.configuration.domain.ConfigurationDomainService;
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.ThreadLocalContextUtil;
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.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.jetbrains.annotations.NotNull;
Review Comment:
Idea's method extract auto generated it, thanks for catching it, fixed it.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@fineract.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] FINERACT-2081: External event sending make mark as sent async [fineract]
Posted by "adamsaghy (via GitHub)" <gi...@apache.org>.
adamsaghy commented on code in PR #4164:
URL: https://github.com/apache/fineract/pull/4164#discussion_r1840577738
##########
fineract-core/src/main/java/org/apache/fineract/infrastructure/event/external/repository/ExternalEventRepository.java:
##########
@@ -40,6 +41,7 @@ public interface ExternalEventRepository extends JpaRepository<ExternalEvent, Lo
void deleteOlderEventsWithSentStatus(@Param("status") ExternalEventStatus status,
@Param("dateForPurgeCriteria") LocalDate dateForPurgeCriteria);
+ @Transactional
Review Comment:
Kind of useless, no?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@fineract.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] FINERACT-2081: External event sending make mark as sent async [fineract]
Posted by "magyari-adam (via GitHub)" <gi...@apache.org>.
magyari-adam commented on code in PR #4164:
URL: https://github.com/apache/fineract/pull/4164#discussion_r1842456856
##########
fineract-provider/src/main/resources/application.properties:
##########
@@ -99,7 +99,10 @@ fineract.remote-job-message-handler.kafka.admin.extra-properties-key-value-separ
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.partition-size=${FINERACT_EXTERNAL_EVENTS_PARTITION_SIZE:5000}
+fineract.events.external.partition-size=${FINERACT_EXTERNAL_EVENTS_PARTITION_SIZE:5}
Review Comment:
oops, left that in from testing, fixed
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@fineract.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] FINERACT-2081: External event sending make mark as sent async [fineract]
Posted by "adamsaghy (via GitHub)" <gi...@apache.org>.
adamsaghy commented on code in PR #4164:
URL: https://github.com/apache/fineract/pull/4164#discussion_r1840721498
##########
fineract-core/src/main/java/org/apache/fineract/infrastructure/event/external/jobs/SendAsynchronousEventsTasklet.java:
##########
@@ -104,14 +111,40 @@ private void markEventsAsSent(List<Long> eventIds) {
// Partitioning dataset to avoid exception: PreparedStatement can have at most 65,535 parameters
final int partitionSize = fineractProperties.getEvents().getExternal().getPartitionSize();
List<List<Long>> partitions = Lists.partition(eventIds, partitionSize);
+ final ThreadPoolTaskExecutor executor = getThreadPoolTaskExecutor();
+ final FineractContext context = ThreadLocalContextUtil.getContext();
+ List<Future<?>> tasks = new ArrayList<>();
partitions //
.forEach(partitionedEventIds -> {
- measure(() -> {
- repository.markEventsSent(partitionedEventIds, sentAt);
- }, timeTaken -> {
- log.debug("Took {}ms to update {} events", timeTaken.toMillis(), partitionedEventIds.size());
- });
+ tasks.add(executor.submit(() -> {
+ measure(() -> {
+ ThreadLocalContextUtil.init(context);
+ repository.markEventsSent(partitionedEventIds, sentAt);
+ }, timeTaken -> {
+ log.debug("Took {}ms to update {} events", timeTaken.toMillis(), partitionedEventIds.size());
+ });
+ }));
});
+ tasks.forEach(task -> {
+ try {
+ task.get();
+ } catch (InterruptedException | ExecutionException e) {
+ throw new RuntimeException(e);
+ }
+ });
+ }
+
+ @NotNull
+ private ThreadPoolTaskExecutor getThreadPoolTaskExecutor() {
Review Comment:
We can consider to introduce a producer which fetch the details from `application.properties` and create an executor for us if needed.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@fineract.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] FINERACT-2081: External event sending make mark as sent async [fineract]
Posted by "magyari-adam (via GitHub)" <gi...@apache.org>.
magyari-adam commented on code in PR #4164:
URL: https://github.com/apache/fineract/pull/4164#discussion_r1850201063
##########
fineract-core/src/main/java/org/apache/fineract/infrastructure/event/external/repository/ExternalEventRepository.java:
##########
@@ -40,6 +41,7 @@ public interface ExternalEventRepository extends JpaRepository<ExternalEvent, Lo
void deleteOlderEventsWithSentStatus(@Param("status") ExternalEventStatus status,
@Param("dateForPurgeCriteria") LocalDate dateForPurgeCriteria);
+ @Transactional
Review Comment:
hmm, yea you're right, thanks for the tip with TransactionTemplate
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@fineract.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org