From 32945c657375a7494c80e4d0279e32c483efcfb3 Mon Sep 17 00:00:00 2001 From: Peter Bagrij Date: Thu, 20 Jul 2023 12:03:04 +0200 Subject: [PATCH] FINERACT-1824 - Kafka support for remote batch jobs: cob - Kafka support for external events --- .github/workflows/build-docker-postgresql.yml | 4 +- .github/workflows/smoke-activemq.yml | 37 ++++ .github/workflows/smoke-kafka.yml | 37 ++++ docker-compose-postgresql-activemq.yml | 8 +- docker-compose-postgresql-kafka.yml | 171 ++++++++++++++++++ .../core/config/FineractProperties.java | 85 +++++++++ .../core/domain/FineractContext.java | 4 + .../core/domain/FineractPlatformTenant.java | 4 + .../FineractPlatformTenantConnection.java | 6 + .../jobs/SendAsynchronousEventsTasklet.java | 3 +- .../producer/ExternalEventProducer.java | 8 + .../producer/NoopExternalEventEnabled.java | 49 +++++ .../producer/NoopExternalEventProducer.java | 4 +- .../core/config/FineractPropertiesTest.java | 52 ++++++ fineract-provider/dependencies.gradle | 6 + .../ExternalEventKafkaConfiguration.java | 61 +++++++ ...alEventsKafkaTopicAutoCreateCondition.java | 36 ++++ .../config/KafkaExternalEventTopicConfig.java | 57 ++++++ .../kafka/KafkaExternalEventProducer.java | 82 +++++++++ .../springbatch/InputChannelInterceptor.java | 9 +- .../messagehandler/MessageHandlerConfig.java | 36 ++++ .../StepExecutionRequestHandler.java | 5 +- .../{ => jms}/JmsManagerCondition.java | 2 +- .../{ => jms}/JmsWorkerCondition.java | 6 +- .../kafka/KafkaManagerCondition.java | 35 ++++ ...afkaRemoteJobTopicAutoCreateCondition.java | 39 ++++ .../kafka/KafkaWorkerCondition.java | 35 ++++ .../SpringEventManagerCondition.java | 2 +- .../SpringEventWorkerCondition.java | 2 +- .../JmsBatchWorkerMessageListener.java | 5 +- .../{ => jms}/JmsBrokerConfiguration.java | 2 +- .../{ => jms}/JmsManagerConfig.java | 7 +- .../{ => jms}/JmsWorkerConfig.java | 10 +- .../kafka/KafkaJobTopicConfig.java | 57 ++++++ .../kafka/KafkaManagerConfig.java | 91 ++++++++++ .../kafka/KafkaRemoteMessageListener.java | 57 ++++++ .../kafka/KafkaWorkerConfig.java | 79 ++++++++ .../SpringEventManagerConfig.java | 7 +- .../{ => spring}/SpringEventWorkerConfig.java | 7 +- .../src/main/resources/application.properties | 35 ++++ .../kafka/KafkaExternalEventProducerTest.java | 137 ++++++++++++++ integration-tests/build.gradle | 8 +- 42 files changed, 1342 insertions(+), 45 deletions(-) create mode 100644 .github/workflows/smoke-activemq.yml create mode 100644 .github/workflows/smoke-kafka.yml create mode 100644 docker-compose-postgresql-kafka.yml create mode 100644 fineract-core/src/main/java/org/apache/fineract/infrastructure/event/external/producer/NoopExternalEventEnabled.java create mode 100644 fineract-core/src/test/java/org/apache/fineract/infrastructure/core/config/FineractPropertiesTest.java create mode 100644 fineract-provider/src/main/java/org/apache/fineract/infrastructure/event/external/config/ExternalEventKafkaConfiguration.java create mode 100644 fineract-provider/src/main/java/org/apache/fineract/infrastructure/event/external/config/ExternalEventsKafkaTopicAutoCreateCondition.java create mode 100644 fineract-provider/src/main/java/org/apache/fineract/infrastructure/event/external/config/KafkaExternalEventTopicConfig.java create mode 100644 fineract-provider/src/main/java/org/apache/fineract/infrastructure/event/external/producer/kafka/KafkaExternalEventProducer.java create mode 100644 fineract-provider/src/main/java/org/apache/fineract/infrastructure/springbatch/messagehandler/MessageHandlerConfig.java rename fineract-provider/src/main/java/org/apache/fineract/infrastructure/springbatch/messagehandler/conditions/{ => jms}/JmsManagerCondition.java (98%) rename fineract-provider/src/main/java/org/apache/fineract/infrastructure/springbatch/messagehandler/conditions/{ => jms}/JmsWorkerCondition.java (92%) create mode 100644 fineract-provider/src/main/java/org/apache/fineract/infrastructure/springbatch/messagehandler/conditions/kafka/KafkaManagerCondition.java create mode 100644 fineract-provider/src/main/java/org/apache/fineract/infrastructure/springbatch/messagehandler/conditions/kafka/KafkaRemoteJobTopicAutoCreateCondition.java create mode 100644 fineract-provider/src/main/java/org/apache/fineract/infrastructure/springbatch/messagehandler/conditions/kafka/KafkaWorkerCondition.java rename fineract-provider/src/main/java/org/apache/fineract/infrastructure/springbatch/messagehandler/conditions/{ => spring}/SpringEventManagerCondition.java (98%) rename fineract-provider/src/main/java/org/apache/fineract/infrastructure/springbatch/messagehandler/conditions/{ => spring}/SpringEventWorkerCondition.java (98%) rename fineract-provider/src/main/java/org/apache/fineract/infrastructure/springbatch/messagehandler/{ => jms}/JmsBatchWorkerMessageListener.java (95%) rename fineract-provider/src/main/java/org/apache/fineract/infrastructure/springbatch/messagehandler/{ => jms}/JmsBrokerConfiguration.java (99%) rename fineract-provider/src/main/java/org/apache/fineract/infrastructure/springbatch/messagehandler/{ => jms}/JmsManagerConfig.java (93%) rename fineract-provider/src/main/java/org/apache/fineract/infrastructure/springbatch/messagehandler/{ => jms}/JmsWorkerConfig.java (91%) create mode 100644 fineract-provider/src/main/java/org/apache/fineract/infrastructure/springbatch/messagehandler/kafka/KafkaJobTopicConfig.java create mode 100644 fineract-provider/src/main/java/org/apache/fineract/infrastructure/springbatch/messagehandler/kafka/KafkaManagerConfig.java create mode 100644 fineract-provider/src/main/java/org/apache/fineract/infrastructure/springbatch/messagehandler/kafka/KafkaRemoteMessageListener.java create mode 100644 fineract-provider/src/main/java/org/apache/fineract/infrastructure/springbatch/messagehandler/kafka/KafkaWorkerConfig.java rename fineract-provider/src/main/java/org/apache/fineract/infrastructure/springbatch/messagehandler/{ => spring}/SpringEventManagerConfig.java (92%) rename fineract-provider/src/main/java/org/apache/fineract/infrastructure/springbatch/messagehandler/{ => spring}/SpringEventWorkerConfig.java (93%) create mode 100644 fineract-provider/src/test/java/org/apache/fineract/infrastructure/event/external/producer/kafka/KafkaExternalEventProducerTest.java diff --git a/.github/workflows/build-docker-postgresql.yml b/.github/workflows/build-docker-postgresql.yml index ee531b85a..dbbca5264 100644 --- a/.github/workflows/build-docker-postgresql.yml +++ b/.github/workflows/build-docker-postgresql.yml @@ -20,8 +20,8 @@ jobs: distribution: 'zulu' cache: gradle - name: Build the image - run: ./gradlew :fineract-provider:clean :fineract-provider:jibDockerBuild -x test - - name: Start the stack + run: ./gradlew --no-daemon --console=plain :fineract-provider:clean :fineract-provider:jibDockerBuild -x test + - name: Start the Standalone Stack run: docker-compose -f docker-compose-postgresql.yml up -d - name: Wait for stack to come up run: sleep 300 diff --git a/.github/workflows/smoke-activemq.yml b/.github/workflows/smoke-activemq.yml new file mode 100644 index 000000000..3255b48d2 --- /dev/null +++ b/.github/workflows/smoke-activemq.yml @@ -0,0 +1,37 @@ +name: Fineract ActiveMQ Smoke + +on: [push, pull_request] + +permissions: + contents: read + +jobs: + build: + runs-on: ubuntu-20.04 + env: + GRADLE_ENTERPRISE_ACCESS_KEY: ${{ secrets.GE_ACCESS_TOKEN }} + + steps: + - uses: actions/checkout@c85c95e3d7251135ab7dc9ce3241c5835cc595a9 # tag=v3 + - name: Set up JDK 17 + uses: actions/setup-java@5ffc13f4174014e2d4d4572b3d74c3fa61aeb2c2 # v3 + with: + java-version: '17' + distribution: 'zulu' + cache: gradle + - name: Build the image + run: ./gradlew --no-daemon --console=plain :fineract-provider:clean :fineract-provider:jibDockerBuild -x test + - name: Start the ActiveMQ Stack + run: docker-compose -f docker-compose-postgresql-activemq.yml up --scale fineract-worker=1 -d + - name: Wait for stack to come up + run: sleep 500 + - name: Check health Manager + run: curl -f -k --retry 10 --retry-connrefused --connect-timeout 30 --retry-delay 30 https://localhost:8443/fineract-provider/actuator/health + - name: Check health Worker1 + run: curl -f -k --retry 10 --retry-connrefused --connect-timeout 30 --retry-delay 30 https://localhost:8444/fineract-provider/actuator/health + - name: Check info Manager + run: (( $(curl -f -k --retry 5 --retry-connrefused --connect-timeout 30 --retry-delay 30 https://localhost:8443/fineract-provider/actuator/info | wc --chars) > 100 )) + - name: Check info Worker1 + run: (( $(curl -f -k --retry 5 --retry-connrefused --connect-timeout 30 --retry-delay 30 https://localhost:8444/fineract-provider/actuator/info | wc --chars) > 100 )) + - name: Run Smoke Test with Remote COB + run: ./gradlew --no-daemon --console=plain :integration-tests:cleanTest :integration-tests:test --tests "org.apache.fineract.integrationtests.investor.externalassetowner.InitiateExternalAssetOwnerTransferTest" -PcargoDisabled diff --git a/.github/workflows/smoke-kafka.yml b/.github/workflows/smoke-kafka.yml new file mode 100644 index 000000000..bc434963a --- /dev/null +++ b/.github/workflows/smoke-kafka.yml @@ -0,0 +1,37 @@ +name: Fineract Kafka Smoke + +on: [push, pull_request] + +permissions: + contents: read + +jobs: + build: + runs-on: ubuntu-20.04 + env: + GRADLE_ENTERPRISE_ACCESS_KEY: ${{ secrets.GE_ACCESS_TOKEN }} + + steps: + - uses: actions/checkout@c85c95e3d7251135ab7dc9ce3241c5835cc595a9 # tag=v3 + - name: Set up JDK 17 + uses: actions/setup-java@5ffc13f4174014e2d4d4572b3d74c3fa61aeb2c2 # v3 + with: + java-version: '17' + distribution: 'zulu' + cache: gradle + - name: Build the image + run: ./gradlew --no-daemon --console=plain :fineract-provider:clean :fineract-provider:jibDockerBuild -x test + - name: Start the Kafka Stack + run: docker-compose -f docker-compose-postgresql-kafka.yml up --scale fineract-worker=1 -d + - name: Wait for stack to come up + run: sleep 500 + - name: Check health Manager + run: curl -f -k --retry 10 --retry-connrefused --connect-timeout 30 --retry-delay 30 https://localhost:8443/fineract-provider/actuator/health + - name: Check health Worker1 + run: curl -f -k --retry 10 --retry-connrefused --connect-timeout 30 --retry-delay 30 https://localhost:8444/fineract-provider/actuator/health + - name: Check info Manager + run: (( $(curl -f -k --retry 5 --retry-connrefused --connect-timeout 30 --retry-delay 30 https://localhost:8443/fineract-provider/actuator/info | wc --chars) > 100 )) + - name: Check info Worker1 + run: (( $(curl -f -k --retry 5 --retry-connrefused --connect-timeout 30 --retry-delay 30 https://localhost:8444/fineract-provider/actuator/info | wc --chars) > 100 )) + - name: Run Smoke Test with Remote COB + run: ./gradlew --no-daemon --console=plain :integration-tests:cleanTest :integration-tests:test --tests "org.apache.fineract.integrationtests.investor.externalassetowner.InitiateExternalAssetOwnerTransferTest" -PcargoDisabled diff --git a/docker-compose-postgresql-activemq.yml b/docker-compose-postgresql-activemq.yml index 013bfeb54..f57c830de 100644 --- a/docker-compose-postgresql-activemq.yml +++ b/docker-compose-postgresql-activemq.yml @@ -43,7 +43,7 @@ services: retries: 10 ports: - "5432:5432" - fineract-server: + fineract-manager: image: fineract:latest volumes: - ./fineract-provider/build/data:/data @@ -68,6 +68,7 @@ services: - FINERACT_HIKARI_CONNECTION_TIMEOUT=20000 - FINERACT_HIKARI_TEST_QUERY=SELECT 1 - FINERACT_HIKARI_AUTO_COMMIT=true + - FINERACT_SERVER_SSL_ENABLED=true - FINERACT_HIKARI_DS_PROPERTIES_CACHE_PREP_STMTS=true - FINERACT_HIKARI_DS_PROPERTIES_PREP_STMT_CACHE_SIZE=250 - FINERACT_HIKARI_DS_PROPERTIES_PREP_STMT_CACHE_SQL_LIMIT=2048 @@ -109,11 +110,13 @@ services: test: ["CMD", 'sh', '-c', 'echo -e "Checking for the availability of Fineract server deployment"; while ! nc -z "fineract-server" 8443; do sleep 1; printf "-"; done; echo -e " >> Fineract server has started";' ] timeout: 10s retries: 10 + ports: + - "8444-8445:8443" depends_on: fineractpostgresql: condition: service_healthy environment: - - FINERACT_NODE_ID=1 + - FINERACT_NODE_ID=2 - FINERACT_HIKARI_DRIVER_SOURCE_CLASS_NAME=org.postgresql.Driver - FINERACT_HIKARI_JDBC_URL=jdbc:postgresql://fineractpostgresql:5432/fineract_tenants - FINERACT_HIKARI_USERNAME=postgres @@ -124,6 +127,7 @@ services: - FINERACT_HIKARI_CONNECTION_TIMEOUT=20000 - FINERACT_HIKARI_TEST_QUERY=SELECT 1 - FINERACT_HIKARI_AUTO_COMMIT=true + - FINERACT_SERVER_SSL_ENABLED=true - FINERACT_HIKARI_DS_PROPERTIES_CACHE_PREP_STMTS=true - FINERACT_HIKARI_DS_PROPERTIES_PREP_STMT_CACHE_SIZE=250 - FINERACT_HIKARI_DS_PROPERTIES_PREP_STMT_CACHE_SQL_LIMIT=2048 diff --git a/docker-compose-postgresql-kafka.yml b/docker-compose-postgresql-kafka.yml new file mode 100644 index 000000000..bffa10e26 --- /dev/null +++ b/docker-compose-postgresql-kafka.yml @@ -0,0 +1,171 @@ +# 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. +# + +# You can replace and test a more recent version of docker compose +version: '3.7' +services: + kafka: + image: 'bitnami/kafka:latest' + ports: + - '9092:9092' + environment: + - ALLOW_PLAINTEXT_LISTENER=yes + - KAFKA_ADVERTISED_LISTENERS=PLAINTEXT://kafka:9092 + # Backend service + fineractpostgresql: + image: postgres:15.3 + volumes: + - ./fineract-db/docker/postgresql:/docker-entrypoint-initdb.d/:Z,ro + environment: + POSTGRES_USER: root + POSTGRES_PASSWORD: skdcnwauicn2ucnaecasdsajdnizucawencascdca + FINERACT_DB_USER: postgres + FINERACT_DB_PASS: skdcnwauicn2ucnaecasdsajdnizucawencascdca + FINERACT_TENANTS_DB_NAME: fineract_tenants + FINERACT_TENANT_DEFAULT_DB_NAME: fineract_default + healthcheck: + test: [ "CMD", "pg_isready", "-q", "-d", "postgres", "-U", "root" ] + timeout: 10s + retries: 10 + ports: + - "5432:5432" + fineract-manager: + image: fineract:latest + volumes: + - ./fineract-provider/build/data:/data + healthcheck: + test: ["CMD", 'sh', '-c', 'echo -e "Checking for the availability of Fineract server deployment"; while ! nc -z "fineract-server" 8443; do sleep 1; printf "-"; done; echo -e " >> Fineract server has started";' ] + timeout: 10s + retries: 10 + ports: + - 8443:8443 + depends_on: + fineractpostgresql: + condition: service_healthy + environment: + - FINERACT_NODE_ID=1 + - FINERACT_HIKARI_DRIVER_SOURCE_CLASS_NAME=org.postgresql.Driver + - FINERACT_HIKARI_JDBC_URL=jdbc:postgresql://fineractpostgresql:5432/fineract_tenants + - FINERACT_HIKARI_USERNAME=postgres + - FINERACT_HIKARI_PASSWORD=skdcnwauicn2ucnaecasdsajdnizucawencascdca + - FINERACT_HIKARI_MINIMUM_IDLE=3 + - FINERACT_HIKARI_MAXIMUM_POOL_SIZE=10 + - FINERACT_HIKARI_IDLE_TIMEOUT=60000 + - FINERACT_HIKARI_CONNECTION_TIMEOUT=20000 + - FINERACT_HIKARI_TEST_QUERY=SELECT 1 + - FINERACT_HIKARI_AUTO_COMMIT=true + - FINERACT_SERVER_SSL_ENABLED=true + - FINERACT_HIKARI_DS_PROPERTIES_CACHE_PREP_STMTS=true + - FINERACT_HIKARI_DS_PROPERTIES_PREP_STMT_CACHE_SIZE=250 + - FINERACT_HIKARI_DS_PROPERTIES_PREP_STMT_CACHE_SQL_LIMIT=2048 + - FINERACT_HIKARI_DS_PROPERTIES_USE_SERVER_PREP_STMTS=true + - FINERACT_HIKARI_DS_PROPERTIES_USE_LOCAL_SESSION_STATE=true + - FINERACT_HIKARI_DS_PROPERTIES_REWRITE_BATCHED_STATEMENTS=true + - FINERACT_HIKARI_DS_PROPERTIES_CACHE_RESULT_SET_METADATA=true + - FINERACT_HIKARI_DS_PROPERTIES_CACHE_SERVER_CONFIGURATION=true + - FINERACT_HIKARI_DS_PROPERTIES_ELIDE_SET_AUTO_COMMITS=true + - FINERACT_HIKARI_DS_PROPERTIES_MAINTAIN_TIME_STATS=false + - FINERACT_HIKARI_DS_PROPERTIES_LOG_SLOW_QUERIES=true + - FINERACT_HIKARI_DS_PROPERTIES_DUMP_QUERIES_IN_EXCEPTION=true + - FINERACT_DEFAULT_TENANTDB_HOSTNAME=fineractpostgresql + - FINERACT_DEFAULT_TENANTDB_PORT=5432 + - FINERACT_DEFAULT_TENANTDB_UID=postgres + - FINERACT_DEFAULT_TENANTDB_PWD=skdcnwauicn2ucnaecasdsajdnizucawencascdca + - FINERACT_DEFAULT_TENANTDB_CONN_PARAMS= + - FINERACT_DEFAULT_TENANTDB_TIMEZONE=Asia/Kolkata + - FINERACT_DEFAULT_TENANTDB_IDENTIFIER=default + - FINERACT_DEFAULT_TENANTDB_NAME=fineract_default + - FINERACT_DEFAULT_TENANTDB_DESCRIPTION=Default Demo Tenant + - JAVA_TOOL_OPTIONS="-Xmx1G" + - FINERACT_REMOTE_JOB_MESSAGE_HANDLER_JMS_ENABLED=false + - FINERACT_REMOTE_JOB_MESSAGE_HANDLER_SPRING_EVENTS_ENABLED=false + - FINERACT_REMOTE_JOB_MESSAGE_HANDLER_KAFKA_ENABLED=true + - FINERACT_REMOTE_JOB_MESSAGE_HANDLER_KAFKA_BOOTSTRAP_SERVERS=kafka:9092 + - FINERACT_REMOTE_JOB_MESSAGE_HANDLER_KAFKA_TOPIC_NAME=job-topic + - FINERACT_MODE_BATCH_MANAGER_ENABLED=true + - FINERACT_MODE_BATCH_WORKER_ENABLED=false + - LOAN_COB_CHUNK_SIZE=10 + - LOAN_COB_PARTITION_SIZE=10 + - FINERACT_EXTERNAL_EVENTS_ENABLED=true + - FINERACT_EXTERNAL_EVENTS_KAFKA_ENABLED=true + - FINERACT_EXTERNAL_EVENTS_KAFKA_BOOTSTRAP_SERVERS=kafka:9092 + - FINERACT_EXTERNAL_EVENTS_KAFKA_TOPIC_NAME=external-events + - SPRING_PROFILES_ACTIVE=test,diagnostics + fineract-worker: + image: fineract:latest + volumes: + - ./fineract-provider/build/data:/data + deploy: + mode: replicated + replicas: 2 + healthcheck: + test: ["CMD", 'sh', '-c', 'echo -e "Checking for the availability of Fineract server deployment"; while ! nc -z "fineract-server" 8443; do sleep 1; printf "-"; done; echo -e " >> Fineract server has started";' ] + timeout: 10s + retries: 10 + ports: + - "8444-8445:8443" + depends_on: + fineractpostgresql: + condition: service_healthy + environment: + - FINERACT_NODE_ID=1 + - FINERACT_HIKARI_DRIVER_SOURCE_CLASS_NAME=org.postgresql.Driver + - FINERACT_HIKARI_JDBC_URL=jdbc:postgresql://fineractpostgresql:5432/fineract_tenants + - FINERACT_HIKARI_USERNAME=postgres + - FINERACT_HIKARI_PASSWORD=skdcnwauicn2ucnaecasdsajdnizucawencascdca + - FINERACT_HIKARI_MINIMUM_IDLE=3 + - FINERACT_HIKARI_MAXIMUM_POOL_SIZE=10 + - FINERACT_HIKARI_IDLE_TIMEOUT=60000 + - FINERACT_HIKARI_CONNECTION_TIMEOUT=20000 + - FINERACT_HIKARI_TEST_QUERY=SELECT 1 + - FINERACT_HIKARI_AUTO_COMMIT=true + - FINERACT_SERVER_SSL_ENABLED=true + - FINERACT_HIKARI_DS_PROPERTIES_CACHE_PREP_STMTS=true + - FINERACT_HIKARI_DS_PROPERTIES_PREP_STMT_CACHE_SIZE=250 + - FINERACT_HIKARI_DS_PROPERTIES_PREP_STMT_CACHE_SQL_LIMIT=2048 + - FINERACT_HIKARI_DS_PROPERTIES_USE_SERVER_PREP_STMTS=true + - FINERACT_HIKARI_DS_PROPERTIES_USE_LOCAL_SESSION_STATE=true + - FINERACT_HIKARI_DS_PROPERTIES_REWRITE_BATCHED_STATEMENTS=true + - FINERACT_HIKARI_DS_PROPERTIES_CACHE_RESULT_SET_METADATA=true + - FINERACT_HIKARI_DS_PROPERTIES_CACHE_SERVER_CONFIGURATION=true + - FINERACT_HIKARI_DS_PROPERTIES_ELIDE_SET_AUTO_COMMITS=true + - FINERACT_HIKARI_DS_PROPERTIES_MAINTAIN_TIME_STATS=false + - FINERACT_HIKARI_DS_PROPERTIES_LOG_SLOW_QUERIES=true + - FINERACT_HIKARI_DS_PROPERTIES_DUMP_QUERIES_IN_EXCEPTION=true + - FINERACT_DEFAULT_TENANTDB_HOSTNAME=fineractpostgresql + - FINERACT_DEFAULT_TENANTDB_PORT=5432 + - FINERACT_DEFAULT_TENANTDB_UID=postgres + - FINERACT_DEFAULT_TENANTDB_PWD=skdcnwauicn2ucnaecasdsajdnizucawencascdca + - FINERACT_DEFAULT_TENANTDB_CONN_PARAMS= + - FINERACT_DEFAULT_TENANTDB_TIMEZONE=Asia/Kolkata + - FINERACT_DEFAULT_TENANTDB_IDENTIFIER=default + - FINERACT_DEFAULT_TENANTDB_NAME=fineract_default + - FINERACT_DEFAULT_TENANTDB_DESCRIPTION=Default Demo Tenant + - JAVA_TOOL_OPTIONS="-Xmx1G" + - FINERACT_MODE_BATCH_MANAGER_ENABLED=false + - FINERACT_MODE_BATCH_WORKER_ENABLED=true + - FINERACT_REMOTE_JOB_MESSAGE_HANDLER_JMS_ENABLED=false + - FINERACT_REMOTE_JOB_MESSAGE_HANDLER_SPRING_EVENTS_ENABLED=false + - FINERACT_REMOTE_JOB_MESSAGE_HANDLER_KAFKA_ENABLED=true + - FINERACT_REMOTE_JOB_MESSAGE_HANDLER_KAFKA_BOOTSTRAP_SERVERS=kafka:9092 + - FINERACT_REMOTE_JOB_MESSAGE_HANDLER_KAFKA_TOPIC_NAME=job-topic + - FINERACT_EXTERNAL_EVENTS_ENABLED=true + - FINERACT_EXTERNAL_EVENTS_KAFKA_ENABLED=true + - FINERACT_EXTERNAL_EVENTS_KAFKA_BOOTSTRAP_SERVERS=kafka:9092 + - FINERACT_EXTERNAL_EVENTS_KAFKA_TOPIC_NAME=external-events + - SPRING_PROFILES_ACTIVE=test,diagnostics diff --git a/fineract-core/src/main/java/org/apache/fineract/infrastructure/core/config/FineractProperties.java b/fineract-core/src/main/java/org/apache/fineract/infrastructure/core/config/FineractProperties.java index cbfae0298..bd149d4eb 100644 --- a/fineract-core/src/main/java/org/apache/fineract/infrastructure/core/config/FineractProperties.java +++ b/fineract-core/src/main/java/org/apache/fineract/infrastructure/core/config/FineractProperties.java @@ -19,9 +19,13 @@ package org.apache.fineract.infrastructure.core.config; +import java.util.Arrays; +import java.util.HashMap; import java.util.List; +import java.util.Map; import lombok.Getter; import lombok.Setter; +import lombok.extern.slf4j.Slf4j; import org.apache.commons.lang3.StringUtils; import org.springframework.boot.context.properties.ConfigurationProperties; @@ -161,6 +165,7 @@ public class FineractProperties { private FineractRemoteJobMessageHandlerSpringEventsProperties springEvents; private FineractRemoteJobMessageHandlerJmsProperties jms; + private FineractRemoteJobMessageHandlerKafkaProperties kafka; } @Getter @@ -185,6 +190,73 @@ public class FineractProperties { } } + @Getter + @Setter + public static class FineractRemoteJobMessageHandlerKafkaProperties { + + private boolean enabled; + private String bootstrapServers; + private KafkaTopicProperties topic; + private KafkaConsumerProperties consumer; + private KafkaProperties producer; + private KafkaProperties admin; + } + + @Getter + @Setter + public static class KafkaTopicProperties { + + private boolean autoCreate; + private String name; + private int replicas; + private int partitions; + } + + @Getter + @Setter + public static class KafkaConsumerProperties extends KafkaProperties { + + private String groupId; + } + + @Getter + @Setter + @Slf4j + public static class KafkaProperties { + + private String extraPropertiesKeyValueSeparator; + private String extraPropertiesSeparator; + private String extraProperties; + + public Map getExtraPropertiesMap() { + Map map = new HashMap<>(); + if (StringUtils.isNotEmpty(getExtraProperties()) && validateSeparators()) { + String[] lines = StringUtils.split(getExtraProperties(), extraPropertiesSeparator); + Arrays.stream(lines).forEach(line -> { + String[] keyAndValue = StringUtils.split(line, extraPropertiesKeyValueSeparator); + if (keyAndValue.length == 2) { + map.put(keyAndValue[0], keyAndValue[1]); + } else { + log.warn("Invalid property: {}", line); + } + + }); + } + return map; + } + + private boolean validateSeparators() { + boolean valid = (StringUtils.isNotEmpty(extraPropertiesSeparator) && extraPropertiesSeparator.length() == 1 + && StringUtils.isNotEmpty(extraPropertiesKeyValueSeparator) && extraPropertiesKeyValueSeparator.length() == 1 + && !extraPropertiesSeparator.equals(extraPropertiesKeyValueSeparator)); + if (!valid) { + log.warn("Invalid KafkaProperties configuration, lineSeparator '{}' and keyValueSeparator '{}'", extraPropertiesSeparator, + extraPropertiesKeyValueSeparator); + } + return valid; + } + } + @Getter @Setter public static class FineractEventsProperties { @@ -213,6 +285,7 @@ public class FineractProperties { public static class FineractExternalEventsProducerProperties { private FineractExternalEventsProducerJmsProperties jms; + private FineractExternalEventsProducerKafkaProperties kafka; } @Getter @@ -235,6 +308,18 @@ public class FineractProperties { } } + @Getter + @Setter + public static class FineractExternalEventsProducerKafkaProperties { + + private boolean enabled; + private String bootstrapServers; + private KafkaTopicProperties topic; + private KafkaProperties producer; + private KafkaProperties admin; + private int timeoutInSeconds; + } + @Getter @Setter public static class FineractContentProperties { diff --git a/fineract-core/src/main/java/org/apache/fineract/infrastructure/core/domain/FineractContext.java b/fineract-core/src/main/java/org/apache/fineract/infrastructure/core/domain/FineractContext.java index 76736fbfe..b5e77a276 100644 --- a/fineract-core/src/main/java/org/apache/fineract/infrastructure/core/domain/FineractContext.java +++ b/fineract-core/src/main/java/org/apache/fineract/infrastructure/core/domain/FineractContext.java @@ -22,10 +22,14 @@ import java.io.Serializable; import java.time.LocalDate; import java.util.HashMap; import lombok.AllArgsConstructor; +import lombok.Builder; import lombok.Getter; +import lombok.extern.jackson.Jacksonized; import org.apache.fineract.infrastructure.businessdate.domain.BusinessDateType; @AllArgsConstructor +@Jacksonized +@Builder @Getter public class FineractContext implements Serializable { diff --git a/fineract-core/src/main/java/org/apache/fineract/infrastructure/core/domain/FineractPlatformTenant.java b/fineract-core/src/main/java/org/apache/fineract/infrastructure/core/domain/FineractPlatformTenant.java index 06f3fbc7a..bddce4c68 100644 --- a/fineract-core/src/main/java/org/apache/fineract/infrastructure/core/domain/FineractPlatformTenant.java +++ b/fineract-core/src/main/java/org/apache/fineract/infrastructure/core/domain/FineractPlatformTenant.java @@ -19,7 +19,11 @@ package org.apache.fineract.infrastructure.core.domain; import java.io.Serializable; +import lombok.Builder; +import lombok.extern.jackson.Jacksonized; +@Jacksonized +@Builder public class FineractPlatformTenant implements Serializable { private final Long id; diff --git a/fineract-core/src/main/java/org/apache/fineract/infrastructure/core/domain/FineractPlatformTenantConnection.java b/fineract-core/src/main/java/org/apache/fineract/infrastructure/core/domain/FineractPlatformTenantConnection.java index 02181a759..37f169911 100644 --- a/fineract-core/src/main/java/org/apache/fineract/infrastructure/core/domain/FineractPlatformTenantConnection.java +++ b/fineract-core/src/main/java/org/apache/fineract/infrastructure/core/domain/FineractPlatformTenantConnection.java @@ -21,13 +21,19 @@ package org.apache.fineract.infrastructure.core.domain; import java.io.Serializable; import java.sql.Connection; import javax.sql.DataSource; +import lombok.AllArgsConstructor; +import lombok.Builder; import lombok.Getter; +import lombok.extern.jackson.Jacksonized; import org.apache.commons.lang3.StringUtils; /** * Holds Tenant's DB server connection connection details. */ @Getter +@Builder +@AllArgsConstructor +@Jacksonized public class FineractPlatformTenantConnection implements Serializable { private final Long connectionId; diff --git a/fineract-core/src/main/java/org/apache/fineract/infrastructure/event/external/jobs/SendAsynchronousEventsTasklet.java b/fineract-core/src/main/java/org/apache/fineract/infrastructure/event/external/jobs/SendAsynchronousEventsTasklet.java index d89866517..4217620ac 100644 --- a/fineract-core/src/main/java/org/apache/fineract/infrastructure/event/external/jobs/SendAsynchronousEventsTasklet.java +++ b/fineract-core/src/main/java/org/apache/fineract/infrastructure/event/external/jobs/SendAsynchronousEventsTasklet.java @@ -76,7 +76,8 @@ public class SendAsynchronousEventsTasklet implements Tasklet { } private boolean isDownstreamChannelEnabled() { - return fineractProperties.getEvents().getExternal().getProducer().getJms().isEnabled(); + return fineractProperties.getEvents().getExternal().getProducer().getJms().isEnabled() + || fineractProperties.getEvents().getExternal().getProducer().getKafka().isEnabled(); } private List getQueuedEventsBatch() { diff --git a/fineract-core/src/main/java/org/apache/fineract/infrastructure/event/external/producer/ExternalEventProducer.java b/fineract-core/src/main/java/org/apache/fineract/infrastructure/event/external/producer/ExternalEventProducer.java index 14dd60ce5..4f4143a4e 100644 --- a/fineract-core/src/main/java/org/apache/fineract/infrastructure/event/external/producer/ExternalEventProducer.java +++ b/fineract-core/src/main/java/org/apache/fineract/infrastructure/event/external/producer/ExternalEventProducer.java @@ -24,5 +24,13 @@ import org.apache.fineract.infrastructure.event.external.exception.Acknowledgeme public interface ExternalEventProducer { + /** + * Sends the created ExternalEvents + * + * @param partitions + * is a Map> partitions, the key here the id of the aggregated root. The value is list + * of external events belong to the same key, serialized into byte array + * @throws AcknowledgementTimeoutException + */ void sendEvents(Map> partitions) throws AcknowledgementTimeoutException; } diff --git a/fineract-core/src/main/java/org/apache/fineract/infrastructure/event/external/producer/NoopExternalEventEnabled.java b/fineract-core/src/main/java/org/apache/fineract/infrastructure/event/external/producer/NoopExternalEventEnabled.java new file mode 100644 index 000000000..cc015c530 --- /dev/null +++ b/fineract-core/src/main/java/org/apache/fineract/infrastructure/event/external/producer/NoopExternalEventEnabled.java @@ -0,0 +1,49 @@ +/** + * 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 org.apache.fineract.infrastructure.core.condition.PropertiesCondition; +import org.apache.fineract.infrastructure.core.config.FineractProperties; + +public class NoopExternalEventEnabled extends PropertiesCondition { + + @Override + protected boolean matches(FineractProperties properties) { + return (!getEffectiveJMSProperty(properties) && !getEffectiveKafkaProperty(properties)); + } + + private boolean getEffectiveJMSProperty(FineractProperties properties) { + if (properties.getEvents() == null || properties.getEvents().getExternal() == null + || properties.getEvents().getExternal().getProducer() == null + || properties.getEvents().getExternal().getProducer().getJms() == null) { + return false; + } + return properties.getEvents().getExternal().getProducer().getJms().isEnabled(); + } + + private boolean getEffectiveKafkaProperty(FineractProperties properties) { + if (properties.getEvents() == null || properties.getEvents().getExternal() == null + || properties.getEvents().getExternal().getProducer() == null + || properties.getEvents().getExternal().getProducer().getKafka() == null) { + return false; + } + return properties.getEvents().getExternal().getProducer().getKafka().isEnabled(); + } + +} diff --git a/fineract-core/src/main/java/org/apache/fineract/infrastructure/event/external/producer/NoopExternalEventProducer.java b/fineract-core/src/main/java/org/apache/fineract/infrastructure/event/external/producer/NoopExternalEventProducer.java index 8e6227eb1..6fd13a8d8 100644 --- a/fineract-core/src/main/java/org/apache/fineract/infrastructure/event/external/producer/NoopExternalEventProducer.java +++ b/fineract-core/src/main/java/org/apache/fineract/infrastructure/event/external/producer/NoopExternalEventProducer.java @@ -22,11 +22,11 @@ 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.context.annotation.Conditional; import org.springframework.stereotype.Component; @Component -@ConditionalOnProperty(value = "fineract.events.external.producer.jms.enabled", havingValue = "false") +@Conditional(NoopExternalEventEnabled.class) @Slf4j public class NoopExternalEventProducer implements ExternalEventProducer { diff --git a/fineract-core/src/test/java/org/apache/fineract/infrastructure/core/config/FineractPropertiesTest.java b/fineract-core/src/test/java/org/apache/fineract/infrastructure/core/config/FineractPropertiesTest.java new file mode 100644 index 000000000..e6b77d5a8 --- /dev/null +++ b/fineract-core/src/test/java/org/apache/fineract/infrastructure/core/config/FineractPropertiesTest.java @@ -0,0 +1,52 @@ +/** + * 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.config; + +import com.google.common.collect.Maps; +import java.util.Map; +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.Test; + +class FineractPropertiesTest { + + @Test + public void testKafkaProperties() { + // valid configs + testKafkaPropertiesParse("|", "=", null, Map.of()); + testKafkaPropertiesParse("|", "=", "", Map.of()); + testKafkaPropertiesParse("|", "=", "key1=value1", Map.of("key1", "value1")); + testKafkaPropertiesParse("|", "=", "key1=value1|key2=value2", Map.of("key1", "value1", "key2", "value2")); + testKafkaPropertiesParse(";", ":", "key1:value1;key2:value2", Map.of("key1", "value1", "key2", "value2")); + + // invalid configs + testKafkaPropertiesParse("||", "=", "key1=value1", Map.of()); + testKafkaPropertiesParse("|", "", "key1=value1", Map.of()); + testKafkaPropertiesParse("", "", "key1=value1", Map.of()); + testKafkaPropertiesParse("|", "=", "key1=value1=value2", Map.of()); + } + + private void testKafkaPropertiesParse(String lineSep, String keyValueSep, String property, Map expected) { + FineractProperties.KafkaProperties kafkaProperties = new FineractProperties.KafkaProperties(); + kafkaProperties.setExtraProperties(property); + kafkaProperties.setExtraPropertiesSeparator(lineSep); + kafkaProperties.setExtraPropertiesKeyValueSeparator(keyValueSep); + Assertions.assertTrue(Maps.difference(expected, kafkaProperties.getExtraPropertiesMap()).areEqual()); + } + +} diff --git a/fineract-provider/dependencies.gradle b/fineract-provider/dependencies.gradle index a9861f614..e4c6d29c9 100644 --- a/fineract-provider/dependencies.gradle +++ b/fineract-provider/dependencies.gradle @@ -43,6 +43,8 @@ dependencies { 'org.springframework.boot:spring-boot-starter-integration', 'org.springframework.integration:spring-integration-event', 'org.springframework.integration:spring-integration-jms', + 'org.springframework.integration:spring-integration-kafka', + 'org.springframework.kafka:spring-kafka', 'org.apache.avro:avro', 'jakarta.ws.rs:jakarta.ws.rs-api', @@ -91,6 +93,10 @@ dependencies { 'io.github.resilience4j:resilience4j-spring-boot2', ) + implementation ('software.amazon.msk:aws-msk-iam-auth:1.1.7') { + exclude group: 'commons-logging' + } + implementation 'io.swagger.core.v3:swagger-jaxrs2-jakarta:2.2.11' implementation ('org.apache.commons:commons-email') { diff --git a/fineract-provider/src/main/java/org/apache/fineract/infrastructure/event/external/config/ExternalEventKafkaConfiguration.java b/fineract-provider/src/main/java/org/apache/fineract/infrastructure/event/external/config/ExternalEventKafkaConfiguration.java new file mode 100644 index 000000000..32453cbab --- /dev/null +++ b/fineract-provider/src/main/java/org/apache/fineract/infrastructure/event/external/config/ExternalEventKafkaConfiguration.java @@ -0,0 +1,61 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package org.apache.fineract.infrastructure.event.external.config; + +import static org.apache.kafka.clients.producer.ProducerConfig.BOOTSTRAP_SERVERS_CONFIG; +import static org.apache.kafka.clients.producer.ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG; +import static org.apache.kafka.clients.producer.ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG; + +import java.util.HashMap; +import java.util.Map; +import org.apache.fineract.infrastructure.core.config.FineractProperties; +import org.apache.kafka.common.serialization.ByteArraySerializer; +import org.apache.kafka.common.serialization.LongSerializer; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import org.springframework.kafka.core.DefaultKafkaProducerFactory; +import org.springframework.kafka.core.KafkaTemplate; +import org.springframework.kafka.core.ProducerFactory; + +@Configuration +@ConditionalOnProperty(value = "fineract.events.external.producer.kafka.enabled", havingValue = "true") +public class ExternalEventKafkaConfiguration { + + @Autowired + private FineractProperties fineractProperties; + + @Bean + public ProducerFactory externalEventsProducerFactory() { + FineractProperties.FineractExternalEventsProducerKafkaProperties kafkaProp = fineractProperties.getEvents().getExternal() + .getProducer().getKafka(); + Map props = new HashMap<>(kafkaProp.getProducer().getExtraPropertiesMap()); + props.put(BOOTSTRAP_SERVERS_CONFIG, kafkaProp.getBootstrapServers()); + props.put(KEY_SERIALIZER_CLASS_CONFIG, LongSerializer.class); + props.put(VALUE_SERIALIZER_CLASS_CONFIG, ByteArraySerializer.class); + return new DefaultKafkaProducerFactory<>(props); + } + + @Bean + public KafkaTemplate externalEventsKafkaTemplate(ProducerFactory externalEventsProducerFactory) { + return new KafkaTemplate<>(externalEventsProducerFactory); + } + +} diff --git a/fineract-provider/src/main/java/org/apache/fineract/infrastructure/event/external/config/ExternalEventsKafkaTopicAutoCreateCondition.java b/fineract-provider/src/main/java/org/apache/fineract/infrastructure/event/external/config/ExternalEventsKafkaTopicAutoCreateCondition.java new file mode 100644 index 000000000..98e942fe8 --- /dev/null +++ b/fineract-provider/src/main/java/org/apache/fineract/infrastructure/event/external/config/ExternalEventsKafkaTopicAutoCreateCondition.java @@ -0,0 +1,36 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package org.apache.fineract.infrastructure.event.external.config; + +import org.springframework.boot.autoconfigure.condition.AllNestedConditions; +import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty; + +public class ExternalEventsKafkaTopicAutoCreateCondition extends AllNestedConditions { + + public ExternalEventsKafkaTopicAutoCreateCondition() { + super(ConfigurationPhase.PARSE_CONFIGURATION); + } + + @ConditionalOnProperty(value = "fineract.events.external.producer.kafka.enabled", havingValue = "true") + static class ExternalEventsKafkaCondition {} + + @ConditionalOnProperty(value = "fineract.events.external.producer.kafka.topic.auto-create", havingValue = "true") + static class TopicAutoCreateCondition {} + +} diff --git a/fineract-provider/src/main/java/org/apache/fineract/infrastructure/event/external/config/KafkaExternalEventTopicConfig.java b/fineract-provider/src/main/java/org/apache/fineract/infrastructure/event/external/config/KafkaExternalEventTopicConfig.java new file mode 100644 index 000000000..0d5e4562d --- /dev/null +++ b/fineract-provider/src/main/java/org/apache/fineract/infrastructure/event/external/config/KafkaExternalEventTopicConfig.java @@ -0,0 +1,57 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package org.apache.fineract.infrastructure.event.external.config; + +import static org.apache.kafka.clients.producer.ProducerConfig.BOOTSTRAP_SERVERS_CONFIG; + +import java.util.HashMap; +import java.util.Map; +import org.apache.fineract.infrastructure.core.config.FineractProperties; +import org.apache.kafka.clients.admin.NewTopic; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Conditional; +import org.springframework.context.annotation.Configuration; +import org.springframework.kafka.config.TopicBuilder; +import org.springframework.kafka.core.KafkaAdmin; + +@Configuration +@Conditional(ExternalEventsKafkaTopicAutoCreateCondition.class) +public class KafkaExternalEventTopicConfig { + + @Autowired + private FineractProperties fineractProperties; + + @Bean + public KafkaAdmin admin() { + Map props = new HashMap<>( + fineractProperties.getEvents().getExternal().getProducer().getKafka().getAdmin().getExtraPropertiesMap()); + props.put(BOOTSTRAP_SERVERS_CONFIG, fineractProperties.getRemoteJobMessageHandler().getKafka().getBootstrapServers()); + return new KafkaAdmin(props); + } + + @Bean + public NewTopic externalEventsTopic() { + FineractProperties.KafkaTopicProperties topicProperties = fineractProperties.getEvents().getExternal().getProducer().getKafka() + .getTopic(); + return TopicBuilder.name(topicProperties.getName()).partitions(topicProperties.getPartitions()) + .replicas(topicProperties.getReplicas()).build(); + } + +} diff --git a/fineract-provider/src/main/java/org/apache/fineract/infrastructure/event/external/producer/kafka/KafkaExternalEventProducer.java b/fineract-provider/src/main/java/org/apache/fineract/infrastructure/event/external/producer/kafka/KafkaExternalEventProducer.java new file mode 100644 index 000000000..9cf815e41 --- /dev/null +++ b/fineract-provider/src/main/java/org/apache/fineract/infrastructure/event/external/producer/kafka/KafkaExternalEventProducer.java @@ -0,0 +1,82 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package org.apache.fineract.infrastructure.event.external.producer.kafka; + +import static org.apache.fineract.infrastructure.core.diagnostics.performance.MeasuringUtil.measure; + +import java.util.ArrayList; +import java.util.Collection; +import java.util.List; +import java.util.Map; +import java.util.Set; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.TimeUnit; +import lombok.AllArgsConstructor; +import lombok.extern.slf4j.Slf4j; +import org.apache.fineract.infrastructure.core.config.FineractProperties; +import org.apache.fineract.infrastructure.event.external.exception.AcknowledgementTimeoutException; +import org.apache.fineract.infrastructure.event.external.producer.ExternalEventProducer; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty; +import org.springframework.kafka.core.KafkaTemplate; +import org.springframework.kafka.support.SendResult; +import org.springframework.stereotype.Component; + +@Component +@Slf4j +@ConditionalOnProperty(value = "fineract.events.external.producer.kafka.enabled", havingValue = "true") +@AllArgsConstructor +public class KafkaExternalEventProducer implements ExternalEventProducer { + + @Autowired + private KafkaTemplate externalEventsKafkaTemplate; + + @Autowired + private FineractProperties fineractProperties; + + @Override + public void sendEvents(Map> partitions) throws AcknowledgementTimeoutException { + FineractProperties.FineractExternalEventsProducerKafkaProperties kafkaProperties = fineractProperties.getEvents().getExternal() + .getProducer().getKafka(); + String topicName = kafkaProperties.getTopic().getName(); + List>> sendResults = new ArrayList<>(); + measure(() -> { + Set keys = partitions.keySet(); + for (Long key : keys) { + List messages = partitions.get(key); + for (byte[] message : messages) { + sendResults.add(externalEventsKafkaTemplate.send(topicName, key, message)); + } + } + + try { + CompletableFuture allOf = CompletableFuture.allOf(sendResults.toArray(new CompletableFuture[0])); + allOf.get(kafkaProperties.getTimeoutInSeconds(), TimeUnit.SECONDS); + } catch (Exception exception) { + throw new RuntimeException("Could not send the messages", exception); + } + }, timeTaken -> { + if (log.isDebugEnabled()) { + int eventCount = partitions.values().stream().map(Collection::size).reduce(0, Integer::sum); + int msgPerSec = (int) (((double) eventCount / timeTaken.toMillis()) * 1000); + log.debug("Sent messages with {} msg/s", msgPerSec); + } + }); + } +} diff --git a/fineract-provider/src/main/java/org/apache/fineract/infrastructure/springbatch/InputChannelInterceptor.java b/fineract-provider/src/main/java/org/apache/fineract/infrastructure/springbatch/InputChannelInterceptor.java index 19f5d5da8..0d372435a 100644 --- a/fineract-provider/src/main/java/org/apache/fineract/infrastructure/springbatch/InputChannelInterceptor.java +++ b/fineract-provider/src/main/java/org/apache/fineract/infrastructure/springbatch/InputChannelInterceptor.java @@ -37,9 +37,12 @@ public class InputChannelInterceptor implements ExecutorChannelInterceptor { } public Message beforeHandleMessage(Message message) { - ContextualMessage castedMessage = (ContextualMessage) message.getPayload(); - ThreadLocalContextUtil.init(castedMessage.getContext()); + return new GenericMessage<>(beforeHandleMessage((ContextualMessage) message.getPayload())); + } + + public StepExecutionRequest beforeHandleMessage(ContextualMessage contextualMessage) { + ThreadLocalContextUtil.init(contextualMessage.getContext()); ThreadLocalContextUtil.setActionContext(ActionContext.COB); - return new GenericMessage<>(castedMessage.getStepExecutionRequest()); + return contextualMessage.getStepExecutionRequest(); } } diff --git a/fineract-provider/src/main/java/org/apache/fineract/infrastructure/springbatch/messagehandler/MessageHandlerConfig.java b/fineract-provider/src/main/java/org/apache/fineract/infrastructure/springbatch/messagehandler/MessageHandlerConfig.java new file mode 100644 index 000000000..741a2d947 --- /dev/null +++ b/fineract-provider/src/main/java/org/apache/fineract/infrastructure/springbatch/messagehandler/MessageHandlerConfig.java @@ -0,0 +1,36 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package org.apache.fineract.infrastructure.springbatch.messagehandler; + +import org.springframework.batch.core.step.StepLocator; +import org.springframework.batch.integration.partition.BeanFactoryStepLocator; +import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; + +@Configuration +@ConditionalOnProperty(value = "fineract.mode.batch-worker-enabled", havingValue = "true") +public class MessageHandlerConfig { + + @Bean + public StepLocator stepLocator() { + return new BeanFactoryStepLocator(); + } + +} diff --git a/fineract-provider/src/main/java/org/apache/fineract/infrastructure/springbatch/messagehandler/StepExecutionRequestHandler.java b/fineract-provider/src/main/java/org/apache/fineract/infrastructure/springbatch/messagehandler/StepExecutionRequestHandler.java index 60844eb5c..126039fd0 100644 --- a/fineract-provider/src/main/java/org/apache/fineract/infrastructure/springbatch/messagehandler/StepExecutionRequestHandler.java +++ b/fineract-provider/src/main/java/org/apache/fineract/infrastructure/springbatch/messagehandler/StepExecutionRequestHandler.java @@ -19,7 +19,6 @@ package org.apache.fineract.infrastructure.springbatch.messagehandler; import lombok.RequiredArgsConstructor; -import org.apache.fineract.infrastructure.springbatch.messagehandler.conditions.JmsWorkerCondition; import org.springframework.batch.core.BatchStatus; import org.springframework.batch.core.JobInterruptedException; import org.springframework.batch.core.Step; @@ -28,13 +27,13 @@ import org.springframework.batch.core.explore.JobExplorer; import org.springframework.batch.core.repository.JobRepository; import org.springframework.batch.core.step.StepLocator; import org.springframework.batch.integration.partition.StepExecutionRequest; -import org.springframework.context.annotation.Conditional; +import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty; import org.springframework.dao.OptimisticLockingFailureException; import org.springframework.stereotype.Component; @Component @RequiredArgsConstructor -@Conditional(JmsWorkerCondition.class) +@ConditionalOnProperty(value = "fineract.mode.batch-worker-enabled", havingValue = "true") public class StepExecutionRequestHandler { private final JobRepository jobRepository; diff --git a/fineract-provider/src/main/java/org/apache/fineract/infrastructure/springbatch/messagehandler/conditions/JmsManagerCondition.java b/fineract-provider/src/main/java/org/apache/fineract/infrastructure/springbatch/messagehandler/conditions/jms/JmsManagerCondition.java similarity index 98% rename from fineract-provider/src/main/java/org/apache/fineract/infrastructure/springbatch/messagehandler/conditions/JmsManagerCondition.java rename to fineract-provider/src/main/java/org/apache/fineract/infrastructure/springbatch/messagehandler/conditions/jms/JmsManagerCondition.java index b8bea3104..2b7b27cc5 100644 --- a/fineract-provider/src/main/java/org/apache/fineract/infrastructure/springbatch/messagehandler/conditions/JmsManagerCondition.java +++ b/fineract-provider/src/main/java/org/apache/fineract/infrastructure/springbatch/messagehandler/conditions/jms/JmsManagerCondition.java @@ -16,7 +16,7 @@ * specific language governing permissions and limitations * under the License. */ -package org.apache.fineract.infrastructure.springbatch.messagehandler.conditions; +package org.apache.fineract.infrastructure.springbatch.messagehandler.conditions.jms; import org.springframework.boot.autoconfigure.condition.AllNestedConditions; import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty; diff --git a/fineract-provider/src/main/java/org/apache/fineract/infrastructure/springbatch/messagehandler/conditions/JmsWorkerCondition.java b/fineract-provider/src/main/java/org/apache/fineract/infrastructure/springbatch/messagehandler/conditions/jms/JmsWorkerCondition.java similarity index 92% rename from fineract-provider/src/main/java/org/apache/fineract/infrastructure/springbatch/messagehandler/conditions/JmsWorkerCondition.java rename to fineract-provider/src/main/java/org/apache/fineract/infrastructure/springbatch/messagehandler/conditions/jms/JmsWorkerCondition.java index 7bdeeca5b..44c98103f 100644 --- a/fineract-provider/src/main/java/org/apache/fineract/infrastructure/springbatch/messagehandler/conditions/JmsWorkerCondition.java +++ b/fineract-provider/src/main/java/org/apache/fineract/infrastructure/springbatch/messagehandler/conditions/jms/JmsWorkerCondition.java @@ -16,7 +16,7 @@ * specific language governing permissions and limitations * under the License. */ -package org.apache.fineract.infrastructure.springbatch.messagehandler.conditions; +package org.apache.fineract.infrastructure.springbatch.messagehandler.conditions.jms; import org.springframework.boot.autoconfigure.condition.AllNestedConditions; import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty; @@ -28,8 +28,8 @@ public class JmsWorkerCondition extends AllNestedConditions { } @ConditionalOnProperty(value = "fineract.mode.batch-worker-enabled", havingValue = "true") - static class WorkerCondition {} + public static class WorkerCondition {} @ConditionalOnProperty(value = "fineract.remote-job-message-handler.jms.enabled", havingValue = "true") - static class JmsCondition {} + public static class JmsCondition {} } diff --git a/fineract-provider/src/main/java/org/apache/fineract/infrastructure/springbatch/messagehandler/conditions/kafka/KafkaManagerCondition.java b/fineract-provider/src/main/java/org/apache/fineract/infrastructure/springbatch/messagehandler/conditions/kafka/KafkaManagerCondition.java new file mode 100644 index 000000000..112fefed5 --- /dev/null +++ b/fineract-provider/src/main/java/org/apache/fineract/infrastructure/springbatch/messagehandler/conditions/kafka/KafkaManagerCondition.java @@ -0,0 +1,35 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package org.apache.fineract.infrastructure.springbatch.messagehandler.conditions.kafka; + +import org.springframework.boot.autoconfigure.condition.AllNestedConditions; +import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty; + +public class KafkaManagerCondition extends AllNestedConditions { + + public KafkaManagerCondition() { + super(ConfigurationPhase.PARSE_CONFIGURATION); + } + + @ConditionalOnProperty(value = "fineract.mode.batch-manager-enabled", havingValue = "true") + static class ManagerCondition {} + + @ConditionalOnProperty(value = "fineract.remote-job-message-handler.kafka.enabled", havingValue = "true") + static class KafkaCondition {} +} diff --git a/fineract-provider/src/main/java/org/apache/fineract/infrastructure/springbatch/messagehandler/conditions/kafka/KafkaRemoteJobTopicAutoCreateCondition.java b/fineract-provider/src/main/java/org/apache/fineract/infrastructure/springbatch/messagehandler/conditions/kafka/KafkaRemoteJobTopicAutoCreateCondition.java new file mode 100644 index 000000000..56325c968 --- /dev/null +++ b/fineract-provider/src/main/java/org/apache/fineract/infrastructure/springbatch/messagehandler/conditions/kafka/KafkaRemoteJobTopicAutoCreateCondition.java @@ -0,0 +1,39 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package org.apache.fineract.infrastructure.springbatch.messagehandler.conditions.kafka; + +import org.springframework.boot.autoconfigure.condition.AllNestedConditions; +import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty; + +public class KafkaRemoteJobTopicAutoCreateCondition extends AllNestedConditions { + + public KafkaRemoteJobTopicAutoCreateCondition() { + super(ConfigurationPhase.PARSE_CONFIGURATION); + } + + @ConditionalOnProperty(value = "fineract.mode.batch-manager-enabled", havingValue = "true") + static class ManagerCondition {} + + @ConditionalOnProperty(value = "fineract.remote-job-message-handler.kafka.enabled", havingValue = "true") + static class KafkaRemoteJobMessageHandlerEnabled {} + + @ConditionalOnProperty(value = "fineract.remote-job-message-handler.kafka.topic.auto-create", havingValue = "true") + static class TopicAutoCreateCondition {} + +} diff --git a/fineract-provider/src/main/java/org/apache/fineract/infrastructure/springbatch/messagehandler/conditions/kafka/KafkaWorkerCondition.java b/fineract-provider/src/main/java/org/apache/fineract/infrastructure/springbatch/messagehandler/conditions/kafka/KafkaWorkerCondition.java new file mode 100644 index 000000000..54d9bfc7a --- /dev/null +++ b/fineract-provider/src/main/java/org/apache/fineract/infrastructure/springbatch/messagehandler/conditions/kafka/KafkaWorkerCondition.java @@ -0,0 +1,35 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package org.apache.fineract.infrastructure.springbatch.messagehandler.conditions.kafka; + +import org.springframework.boot.autoconfigure.condition.AllNestedConditions; +import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty; + +public class KafkaWorkerCondition extends AllNestedConditions { + + public KafkaWorkerCondition() { + super(ConfigurationPhase.PARSE_CONFIGURATION); + } + + @ConditionalOnProperty(value = "fineract.mode.batch-worker-enabled", havingValue = "true") + static class WorkerCondition {} + + @ConditionalOnProperty(value = "fineract.remote-job-message-handler.kafka.enabled", havingValue = "true") + static class KafkaCondition {} +} diff --git a/fineract-provider/src/main/java/org/apache/fineract/infrastructure/springbatch/messagehandler/conditions/SpringEventManagerCondition.java b/fineract-provider/src/main/java/org/apache/fineract/infrastructure/springbatch/messagehandler/conditions/spring/SpringEventManagerCondition.java similarity index 98% rename from fineract-provider/src/main/java/org/apache/fineract/infrastructure/springbatch/messagehandler/conditions/SpringEventManagerCondition.java rename to fineract-provider/src/main/java/org/apache/fineract/infrastructure/springbatch/messagehandler/conditions/spring/SpringEventManagerCondition.java index b0fe92071..3a4e39d7a 100644 --- a/fineract-provider/src/main/java/org/apache/fineract/infrastructure/springbatch/messagehandler/conditions/SpringEventManagerCondition.java +++ b/fineract-provider/src/main/java/org/apache/fineract/infrastructure/springbatch/messagehandler/conditions/spring/SpringEventManagerCondition.java @@ -16,7 +16,7 @@ * specific language governing permissions and limitations * under the License. */ -package org.apache.fineract.infrastructure.springbatch.messagehandler.conditions; +package org.apache.fineract.infrastructure.springbatch.messagehandler.conditions.spring; import org.springframework.boot.autoconfigure.condition.AllNestedConditions; import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty; diff --git a/fineract-provider/src/main/java/org/apache/fineract/infrastructure/springbatch/messagehandler/conditions/SpringEventWorkerCondition.java b/fineract-provider/src/main/java/org/apache/fineract/infrastructure/springbatch/messagehandler/conditions/spring/SpringEventWorkerCondition.java similarity index 98% rename from fineract-provider/src/main/java/org/apache/fineract/infrastructure/springbatch/messagehandler/conditions/SpringEventWorkerCondition.java rename to fineract-provider/src/main/java/org/apache/fineract/infrastructure/springbatch/messagehandler/conditions/spring/SpringEventWorkerCondition.java index 38f243d89..741e9c495 100644 --- a/fineract-provider/src/main/java/org/apache/fineract/infrastructure/springbatch/messagehandler/conditions/SpringEventWorkerCondition.java +++ b/fineract-provider/src/main/java/org/apache/fineract/infrastructure/springbatch/messagehandler/conditions/spring/SpringEventWorkerCondition.java @@ -16,7 +16,7 @@ * specific language governing permissions and limitations * under the License. */ -package org.apache.fineract.infrastructure.springbatch.messagehandler.conditions; +package org.apache.fineract.infrastructure.springbatch.messagehandler.conditions.spring; import org.springframework.boot.autoconfigure.condition.AllNestedConditions; import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty; diff --git a/fineract-provider/src/main/java/org/apache/fineract/infrastructure/springbatch/messagehandler/JmsBatchWorkerMessageListener.java b/fineract-provider/src/main/java/org/apache/fineract/infrastructure/springbatch/messagehandler/jms/JmsBatchWorkerMessageListener.java similarity index 95% rename from fineract-provider/src/main/java/org/apache/fineract/infrastructure/springbatch/messagehandler/JmsBatchWorkerMessageListener.java rename to fineract-provider/src/main/java/org/apache/fineract/infrastructure/springbatch/messagehandler/jms/JmsBatchWorkerMessageListener.java index 7b61388f6..f2d81e9e8 100644 --- a/fineract-provider/src/main/java/org/apache/fineract/infrastructure/springbatch/messagehandler/JmsBatchWorkerMessageListener.java +++ b/fineract-provider/src/main/java/org/apache/fineract/infrastructure/springbatch/messagehandler/jms/JmsBatchWorkerMessageListener.java @@ -16,7 +16,7 @@ * specific language governing permissions and limitations * under the License. */ -package org.apache.fineract.infrastructure.springbatch.messagehandler; +package org.apache.fineract.infrastructure.springbatch.messagehandler.jms; import jakarta.jms.JMSException; import jakarta.jms.MessageListener; @@ -24,7 +24,8 @@ import lombok.RequiredArgsConstructor; import lombok.extern.slf4j.Slf4j; import org.apache.fineract.infrastructure.springbatch.ContextualMessage; import org.apache.fineract.infrastructure.springbatch.InputChannelInterceptor; -import org.apache.fineract.infrastructure.springbatch.messagehandler.conditions.JmsWorkerCondition; +import org.apache.fineract.infrastructure.springbatch.messagehandler.StepExecutionRequestHandler; +import org.apache.fineract.infrastructure.springbatch.messagehandler.conditions.jms.JmsWorkerCondition; import org.springframework.batch.integration.partition.StepExecutionRequest; import org.springframework.beans.factory.InitializingBean; import org.springframework.context.annotation.Conditional; diff --git a/fineract-provider/src/main/java/org/apache/fineract/infrastructure/springbatch/messagehandler/JmsBrokerConfiguration.java b/fineract-provider/src/main/java/org/apache/fineract/infrastructure/springbatch/messagehandler/jms/JmsBrokerConfiguration.java similarity index 99% rename from fineract-provider/src/main/java/org/apache/fineract/infrastructure/springbatch/messagehandler/JmsBrokerConfiguration.java rename to fineract-provider/src/main/java/org/apache/fineract/infrastructure/springbatch/messagehandler/jms/JmsBrokerConfiguration.java index ba579b011..f22900118 100644 --- a/fineract-provider/src/main/java/org/apache/fineract/infrastructure/springbatch/messagehandler/JmsBrokerConfiguration.java +++ b/fineract-provider/src/main/java/org/apache/fineract/infrastructure/springbatch/messagehandler/jms/JmsBrokerConfiguration.java @@ -16,7 +16,7 @@ * specific language governing permissions and limitations * under the License. */ -package org.apache.fineract.infrastructure.springbatch.messagehandler; +package org.apache.fineract.infrastructure.springbatch.messagehandler.jms; import org.apache.activemq.ActiveMQConnectionFactory; import org.apache.activemq.ActiveMQPrefetchPolicy; diff --git a/fineract-provider/src/main/java/org/apache/fineract/infrastructure/springbatch/messagehandler/JmsManagerConfig.java b/fineract-provider/src/main/java/org/apache/fineract/infrastructure/springbatch/messagehandler/jms/JmsManagerConfig.java similarity index 93% rename from fineract-provider/src/main/java/org/apache/fineract/infrastructure/springbatch/messagehandler/JmsManagerConfig.java rename to fineract-provider/src/main/java/org/apache/fineract/infrastructure/springbatch/messagehandler/jms/JmsManagerConfig.java index 6ed0f1022..a2621c82d 100644 --- a/fineract-provider/src/main/java/org/apache/fineract/infrastructure/springbatch/messagehandler/JmsManagerConfig.java +++ b/fineract-provider/src/main/java/org/apache/fineract/infrastructure/springbatch/messagehandler/jms/JmsManagerConfig.java @@ -16,12 +16,12 @@ * specific language governing permissions and limitations * under the License. */ -package org.apache.fineract.infrastructure.springbatch.messagehandler; +package org.apache.fineract.infrastructure.springbatch.messagehandler.jms; import jakarta.jms.ConnectionFactory; import org.apache.fineract.infrastructure.core.config.FineractProperties; import org.apache.fineract.infrastructure.springbatch.OutputChannelInterceptor; -import org.apache.fineract.infrastructure.springbatch.messagehandler.conditions.JmsManagerCondition; +import org.apache.fineract.infrastructure.springbatch.messagehandler.conditions.jms.JmsManagerCondition; import org.springframework.batch.integration.config.annotation.EnableBatchIntegration; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.context.annotation.Bean; @@ -30,7 +30,6 @@ import org.springframework.context.annotation.Configuration; import org.springframework.context.annotation.Import; import org.springframework.integration.channel.DirectChannel; import org.springframework.integration.dsl.IntegrationFlow; -import org.springframework.integration.dsl.IntegrationFlows; import org.springframework.integration.handler.LoggingHandler; import org.springframework.integration.jms.dsl.Jms; @@ -49,7 +48,7 @@ public class JmsManagerConfig { @Bean public IntegrationFlow outboundFlow(ConnectionFactory connectionFactory) { - return IntegrationFlows.from(outboundRequests) // + return IntegrationFlow.from(outboundRequests) // .intercept(outputInterceptor) // .log(LoggingHandler.Level.DEBUG) // .handle(Jms.outboundAdapter(connectionFactory) diff --git a/fineract-provider/src/main/java/org/apache/fineract/infrastructure/springbatch/messagehandler/JmsWorkerConfig.java b/fineract-provider/src/main/java/org/apache/fineract/infrastructure/springbatch/messagehandler/jms/JmsWorkerConfig.java similarity index 91% rename from fineract-provider/src/main/java/org/apache/fineract/infrastructure/springbatch/messagehandler/JmsWorkerConfig.java rename to fineract-provider/src/main/java/org/apache/fineract/infrastructure/springbatch/messagehandler/jms/JmsWorkerConfig.java index f25c6efa2..7ea304402 100644 --- a/fineract-provider/src/main/java/org/apache/fineract/infrastructure/springbatch/messagehandler/JmsWorkerConfig.java +++ b/fineract-provider/src/main/java/org/apache/fineract/infrastructure/springbatch/messagehandler/jms/JmsWorkerConfig.java @@ -16,14 +16,12 @@ * specific language governing permissions and limitations * under the License. */ -package org.apache.fineract.infrastructure.springbatch.messagehandler; +package org.apache.fineract.infrastructure.springbatch.messagehandler.jms; import jakarta.jms.ConnectionFactory; import jakarta.jms.Session; import org.apache.fineract.infrastructure.core.config.FineractProperties; -import org.apache.fineract.infrastructure.springbatch.messagehandler.conditions.JmsWorkerCondition; -import org.springframework.batch.core.step.StepLocator; -import org.springframework.batch.integration.partition.BeanFactoryStepLocator; +import org.apache.fineract.infrastructure.springbatch.messagehandler.conditions.jms.JmsWorkerCondition; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Qualifier; import org.springframework.context.annotation.Bean; @@ -63,8 +61,4 @@ public class JmsWorkerConfig { return factory.createListenerContainer(endpoint); } - @Bean - public StepLocator stepLocator() { - return new BeanFactoryStepLocator(); - } } diff --git a/fineract-provider/src/main/java/org/apache/fineract/infrastructure/springbatch/messagehandler/kafka/KafkaJobTopicConfig.java b/fineract-provider/src/main/java/org/apache/fineract/infrastructure/springbatch/messagehandler/kafka/KafkaJobTopicConfig.java new file mode 100644 index 000000000..a32d66dd5 --- /dev/null +++ b/fineract-provider/src/main/java/org/apache/fineract/infrastructure/springbatch/messagehandler/kafka/KafkaJobTopicConfig.java @@ -0,0 +1,57 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package org.apache.fineract.infrastructure.springbatch.messagehandler.kafka; + +import static org.apache.kafka.clients.producer.ProducerConfig.BOOTSTRAP_SERVERS_CONFIG; + +import java.util.HashMap; +import java.util.Map; +import org.apache.fineract.infrastructure.core.config.FineractProperties; +import org.apache.fineract.infrastructure.springbatch.messagehandler.conditions.kafka.KafkaRemoteJobTopicAutoCreateCondition; +import org.apache.kafka.clients.admin.NewTopic; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Conditional; +import org.springframework.context.annotation.Configuration; +import org.springframework.kafka.config.TopicBuilder; +import org.springframework.kafka.core.KafkaAdmin; + +@Configuration +@Conditional(KafkaRemoteJobTopicAutoCreateCondition.class) +public class KafkaJobTopicConfig { + + @Autowired + private FineractProperties fineractProperties; + + @Bean + public KafkaAdmin admin() { + Map props = new HashMap<>( + fineractProperties.getRemoteJobMessageHandler().getKafka().getAdmin().getExtraPropertiesMap()); + props.put(BOOTSTRAP_SERVERS_CONFIG, fineractProperties.getRemoteJobMessageHandler().getKafka().getBootstrapServers()); + return new KafkaAdmin(props); + } + + @Bean + public NewTopic jobTopic() { + FineractProperties.KafkaTopicProperties topicProperties = fineractProperties.getRemoteJobMessageHandler().getKafka().getTopic(); + return TopicBuilder.name(topicProperties.getName()).partitions(topicProperties.getPartitions()) + .replicas(topicProperties.getReplicas()).build(); + } + +} diff --git a/fineract-provider/src/main/java/org/apache/fineract/infrastructure/springbatch/messagehandler/kafka/KafkaManagerConfig.java b/fineract-provider/src/main/java/org/apache/fineract/infrastructure/springbatch/messagehandler/kafka/KafkaManagerConfig.java new file mode 100644 index 000000000..3e39d0ad3 --- /dev/null +++ b/fineract-provider/src/main/java/org/apache/fineract/infrastructure/springbatch/messagehandler/kafka/KafkaManagerConfig.java @@ -0,0 +1,91 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package org.apache.fineract.infrastructure.springbatch.messagehandler.kafka; + +import static org.apache.kafka.clients.producer.ProducerConfig.BOOTSTRAP_SERVERS_CONFIG; +import static org.apache.kafka.clients.producer.ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG; +import static org.apache.kafka.clients.producer.ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG; + +import java.util.HashMap; +import java.util.Map; +import org.apache.fineract.infrastructure.core.config.FineractProperties; +import org.apache.fineract.infrastructure.springbatch.ContextualMessage; +import org.apache.fineract.infrastructure.springbatch.OutputChannelInterceptor; +import org.apache.fineract.infrastructure.springbatch.messagehandler.conditions.kafka.KafkaManagerCondition; +import org.apache.kafka.common.serialization.IntegerSerializer; +import org.springframework.batch.integration.config.annotation.EnableBatchIntegration; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Conditional; +import org.springframework.context.annotation.Configuration; +import org.springframework.expression.common.LiteralExpression; +import org.springframework.integration.channel.DirectChannel; +import org.springframework.integration.dsl.IntegrationFlow; +import org.springframework.integration.expression.FunctionExpression; +import org.springframework.integration.handler.LoggingHandler; +import org.springframework.integration.kafka.outbound.KafkaProducerMessageHandler; +import org.springframework.kafka.core.DefaultKafkaProducerFactory; +import org.springframework.kafka.core.KafkaTemplate; +import org.springframework.kafka.core.ProducerFactory; +import org.springframework.kafka.support.serializer.JsonSerializer; +import org.springframework.messaging.support.GenericMessage; + +@Configuration +@EnableBatchIntegration +@Conditional(KafkaManagerCondition.class) +public class KafkaManagerConfig { + + @Autowired + private DirectChannel outboundRequests; + @Autowired + private OutputChannelInterceptor outputInterceptor; + @Autowired + private FineractProperties fineractProperties; + + @Bean + public ProducerFactory producerFactory() { + Map props = new HashMap<>( + fineractProperties.getRemoteJobMessageHandler().getKafka().getProducer().getExtraPropertiesMap()); + props.put(BOOTSTRAP_SERVERS_CONFIG, fineractProperties.getRemoteJobMessageHandler().getKafka().getBootstrapServers()); + props.put(KEY_SERIALIZER_CLASS_CONFIG, IntegerSerializer.class); + props.put(VALUE_SERIALIZER_CLASS_CONFIG, JsonSerializer.class); + return new DefaultKafkaProducerFactory<>(props); + } + + @Bean + public KafkaTemplate kafkaTemplate(ProducerFactory producerFactory) { + return new KafkaTemplate<>(producerFactory); + } + + @Bean + public IntegrationFlow outboundFlow(KafkaTemplate kafkaTemplate) { + FineractProperties.KafkaTopicProperties topic = fineractProperties.getRemoteJobMessageHandler().getKafka().getTopic(); + + KafkaProducerMessageHandler messageHandler = new KafkaProducerMessageHandler<>(kafkaTemplate); + messageHandler.setTopicExpression(new LiteralExpression(topic.getName())); + messageHandler.setPartitionIdExpression(new FunctionExpression>( + message -> message.getPayload().getStepExecutionRequest().getStepExecutionId() % topic.getPartitions())); + + return IntegrationFlow.from(outboundRequests) // + .intercept(outputInterceptor) // + .log(LoggingHandler.Level.DEBUG) // + .handle(messageHandler).get(); + } + +} diff --git a/fineract-provider/src/main/java/org/apache/fineract/infrastructure/springbatch/messagehandler/kafka/KafkaRemoteMessageListener.java b/fineract-provider/src/main/java/org/apache/fineract/infrastructure/springbatch/messagehandler/kafka/KafkaRemoteMessageListener.java new file mode 100644 index 000000000..efd668d97 --- /dev/null +++ b/fineract-provider/src/main/java/org/apache/fineract/infrastructure/springbatch/messagehandler/kafka/KafkaRemoteMessageListener.java @@ -0,0 +1,57 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package org.apache.fineract.infrastructure.springbatch.messagehandler.kafka; + +import lombok.extern.slf4j.Slf4j; +import org.apache.fineract.infrastructure.springbatch.ContextualMessage; +import org.apache.fineract.infrastructure.springbatch.InputChannelInterceptor; +import org.apache.fineract.infrastructure.springbatch.messagehandler.StepExecutionRequestHandler; +import org.apache.fineract.infrastructure.springbatch.messagehandler.conditions.kafka.KafkaWorkerCondition; +import org.springframework.batch.integration.partition.StepExecutionRequest; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.context.annotation.Conditional; +import org.springframework.kafka.annotation.KafkaListener; +import org.springframework.kafka.support.Acknowledgment; +import org.springframework.messaging.handler.annotation.Payload; +import org.springframework.stereotype.Component; + +@Component +@Slf4j +@Conditional(KafkaWorkerCondition.class) +public class KafkaRemoteMessageListener { + + @Autowired + private StepExecutionRequestHandler stepExecutionRequestHandler; + + @Autowired + private InputChannelInterceptor inputInterceptor; + + @KafkaListener(topics = "${fineract.remote-job-message-handler.kafka.topic.name}") + public void onMessage(@Payload ContextualMessage contextualMessage, Acknowledgment acknowledgment) { + try { + log.debug("Received Kafka partition message {}", contextualMessage); + StepExecutionRequest stepExecutionRequest = inputInterceptor.beforeHandleMessage(contextualMessage); + stepExecutionRequestHandler.handle(stepExecutionRequest); + } catch (Exception e) { + log.error("Exception while processing Kafka message", e); + } + acknowledgment.acknowledge(); + log.debug("Message was acknowledged {}", acknowledgment); + } +} diff --git a/fineract-provider/src/main/java/org/apache/fineract/infrastructure/springbatch/messagehandler/kafka/KafkaWorkerConfig.java b/fineract-provider/src/main/java/org/apache/fineract/infrastructure/springbatch/messagehandler/kafka/KafkaWorkerConfig.java new file mode 100644 index 000000000..babc7f549 --- /dev/null +++ b/fineract-provider/src/main/java/org/apache/fineract/infrastructure/springbatch/messagehandler/kafka/KafkaWorkerConfig.java @@ -0,0 +1,79 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package org.apache.fineract.infrastructure.springbatch.messagehandler.kafka; + +import static org.apache.kafka.clients.consumer.ConsumerConfig.AUTO_OFFSET_RESET_CONFIG; +import static org.apache.kafka.clients.consumer.ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG; +import static org.apache.kafka.clients.consumer.ConsumerConfig.GROUP_ID_CONFIG; +import static org.apache.kafka.clients.consumer.ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG; +import static org.apache.kafka.clients.consumer.ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG; +import static org.springframework.kafka.support.serializer.JsonDeserializer.TRUSTED_PACKAGES; + +import java.util.HashMap; +import java.util.Map; +import lombok.extern.slf4j.Slf4j; +import org.apache.fineract.infrastructure.core.config.FineractProperties; +import org.apache.fineract.infrastructure.core.config.FineractProperties.FineractRemoteJobMessageHandlerKafkaProperties; +import org.apache.fineract.infrastructure.springbatch.messagehandler.conditions.kafka.KafkaWorkerCondition; +import org.apache.kafka.common.serialization.IntegerDeserializer; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Conditional; +import org.springframework.context.annotation.Configuration; +import org.springframework.integration.channel.QueueChannel; +import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory; +import org.springframework.kafka.core.ConsumerFactory; +import org.springframework.kafka.core.DefaultKafkaConsumerFactory; +import org.springframework.kafka.listener.ContainerProperties; +import org.springframework.kafka.support.serializer.JsonDeserializer; + +@Configuration +@Conditional(KafkaWorkerCondition.class) +@Slf4j +public class KafkaWorkerConfig { + + @Autowired + private QueueChannel inboundRequests; + + @Autowired + private FineractProperties fineractProperties; + + @Bean + public ConsumerFactory consumerFactory() { + FineractRemoteJobMessageHandlerKafkaProperties kafkaProperties = fineractProperties.getRemoteJobMessageHandler().getKafka(); + Map props = new HashMap<>(kafkaProperties.getConsumer().getExtraPropertiesMap()); + props.put(BOOTSTRAP_SERVERS_CONFIG, kafkaProperties.getBootstrapServers()); + props.put(GROUP_ID_CONFIG, kafkaProperties.getConsumer().getGroupId()); + props.put(KEY_DESERIALIZER_CLASS_CONFIG, IntegerDeserializer.class); + props.put(VALUE_DESERIALIZER_CLASS_CONFIG, JsonDeserializer.class); + props.put(TRUSTED_PACKAGES, "*"); + props.put(AUTO_OFFSET_RESET_CONFIG, "earliest"); + return new DefaultKafkaConsumerFactory<>(props); + } + + @Bean + public ConcurrentKafkaListenerContainerFactory kafkaListenerContainerFactory( + ConsumerFactory consumerFactory) { + ConcurrentKafkaListenerContainerFactory factory = new ConcurrentKafkaListenerContainerFactory<>(); + factory.setConsumerFactory(consumerFactory); + factory.getContainerProperties().setAckMode(ContainerProperties.AckMode.MANUAL); + return factory; + } + +} diff --git a/fineract-provider/src/main/java/org/apache/fineract/infrastructure/springbatch/messagehandler/SpringEventManagerConfig.java b/fineract-provider/src/main/java/org/apache/fineract/infrastructure/springbatch/messagehandler/spring/SpringEventManagerConfig.java similarity index 92% rename from fineract-provider/src/main/java/org/apache/fineract/infrastructure/springbatch/messagehandler/SpringEventManagerConfig.java rename to fineract-provider/src/main/java/org/apache/fineract/infrastructure/springbatch/messagehandler/spring/SpringEventManagerConfig.java index 42cdd35c1..b17476f64 100644 --- a/fineract-provider/src/main/java/org/apache/fineract/infrastructure/springbatch/messagehandler/SpringEventManagerConfig.java +++ b/fineract-provider/src/main/java/org/apache/fineract/infrastructure/springbatch/messagehandler/spring/SpringEventManagerConfig.java @@ -16,10 +16,10 @@ * specific language governing permissions and limitations * under the License. */ -package org.apache.fineract.infrastructure.springbatch.messagehandler; +package org.apache.fineract.infrastructure.springbatch.messagehandler.spring; import org.apache.fineract.infrastructure.springbatch.OutputChannelInterceptor; -import org.apache.fineract.infrastructure.springbatch.messagehandler.conditions.SpringEventManagerCondition; +import org.apache.fineract.infrastructure.springbatch.messagehandler.conditions.spring.SpringEventManagerCondition; import org.springframework.batch.integration.config.annotation.EnableBatchIntegration; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.context.annotation.Bean; @@ -27,7 +27,6 @@ import org.springframework.context.annotation.Conditional; import org.springframework.context.annotation.Configuration; import org.springframework.integration.channel.DirectChannel; import org.springframework.integration.dsl.IntegrationFlow; -import org.springframework.integration.dsl.IntegrationFlows; import org.springframework.integration.event.outbound.ApplicationEventPublishingMessageHandler; import org.springframework.integration.handler.LoggingHandler; @@ -44,7 +43,7 @@ public class SpringEventManagerConfig { @Bean public IntegrationFlow outboundFlow() { ApplicationEventPublishingMessageHandler handler = new ApplicationEventPublishingMessageHandler(); - return IntegrationFlows.from(outboundRequests) // + return IntegrationFlow.from(outboundRequests) // .intercept(outputInterceptor) // .log(LoggingHandler.Level.DEBUG) // .handle(handler) // diff --git a/fineract-provider/src/main/java/org/apache/fineract/infrastructure/springbatch/messagehandler/SpringEventWorkerConfig.java b/fineract-provider/src/main/java/org/apache/fineract/infrastructure/springbatch/messagehandler/spring/SpringEventWorkerConfig.java similarity index 93% rename from fineract-provider/src/main/java/org/apache/fineract/infrastructure/springbatch/messagehandler/SpringEventWorkerConfig.java rename to fineract-provider/src/main/java/org/apache/fineract/infrastructure/springbatch/messagehandler/spring/SpringEventWorkerConfig.java index 7691e24f0..f1ba77320 100644 --- a/fineract-provider/src/main/java/org/apache/fineract/infrastructure/springbatch/messagehandler/SpringEventWorkerConfig.java +++ b/fineract-provider/src/main/java/org/apache/fineract/infrastructure/springbatch/messagehandler/spring/SpringEventWorkerConfig.java @@ -16,10 +16,10 @@ * specific language governing permissions and limitations * under the License. */ -package org.apache.fineract.infrastructure.springbatch.messagehandler; +package org.apache.fineract.infrastructure.springbatch.messagehandler.spring; import org.apache.fineract.infrastructure.springbatch.InputChannelInterceptor; -import org.apache.fineract.infrastructure.springbatch.messagehandler.conditions.SpringEventWorkerCondition; +import org.apache.fineract.infrastructure.springbatch.messagehandler.conditions.spring.SpringEventWorkerCondition; import org.springframework.batch.integration.config.annotation.EnableBatchIntegration; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.context.annotation.Bean; @@ -27,7 +27,6 @@ import org.springframework.context.annotation.Conditional; import org.springframework.context.annotation.Configuration; import org.springframework.integration.channel.QueueChannel; import org.springframework.integration.dsl.IntegrationFlow; -import org.springframework.integration.dsl.IntegrationFlows; import org.springframework.integration.event.core.MessagingEvent; import org.springframework.integration.event.inbound.ApplicationEventListeningMessageProducer; import org.springframework.integration.handler.LoggingHandler; @@ -44,7 +43,7 @@ public class SpringEventWorkerConfig { @Bean public IntegrationFlow inboundFlow() { - return IntegrationFlows.from(eventListener()) // + return IntegrationFlow.from(eventListener()) // .log(LoggingHandler.Level.DEBUG) // .channel(inboundRequests) // .intercept(inputInterceptor) // diff --git a/fineract-provider/src/main/resources/application.properties b/fineract-provider/src/main/resources/application.properties index b6e2411b4..b3d997fb5 100644 --- a/fineract-provider/src/main/resources/application.properties +++ b/fineract-provider/src/main/resources/application.properties @@ -74,6 +74,26 @@ fineract.remote-job-message-handler.jms.broker-url=${FINERACT_REMOTE_JOB_MESSAGE fineract.remote-job-message-handler.jms.broker-username=${FINERACT_REMOTE_JOB_MESSAGE_HANDLER_JMS_BROKER_USERNAME:} fineract.remote-job-message-handler.jms.broker-password=${FINERACT_REMOTE_JOB_MESSAGE_HANDLER_JMS_BROKER_PASSWORD:} +fineract.remote-job-message-handler.kafka.enabled=${FINERACT_REMOTE_JOB_MESSAGE_HANDLER_KAFKA_ENABLED:false} +fineract.remote-job-message-handler.kafka.topic.auto-create=${FINERACT_REMOTE_JOB_MESSAGE_HANDLER_KAFKA_TOPIC_AUTO_CREATE:true} +fineract.remote-job-message-handler.kafka.topic.name=${FINERACT_REMOTE_JOB_MESSAGE_HANDLER_KAFKA_TOPIC_NAME:job-topic} +fineract.remote-job-message-handler.kafka.topic.replicas=${FINERACT_REMOTE_JOB_MESSAGE_HANDLER_KAFKA_TOPIC_REPLICAS:1} +fineract.remote-job-message-handler.kafka.topic.partitions=${FINERACT_REMOTE_JOB_MESSAGE_HANDLER_KAFKA_TOPIC_PARTITIONS:10} +fineract.remote-job-message-handler.kafka.bootstrap-servers=${FINERACT_REMOTE_JOB_MESSAGE_HANDLER_KAFKA_BOOTSTRAP_SERVERS:localhost:9092} +fineract.remote-job-message-handler.kafka.consumer.group-id=${FINERACT_REMOTE_JOB_MESSAGE_HANDLER_KAFKA_CONSUMER_GROUPID:fineract-consumer-group-id} +#defines the key and value separator, e.g.: key=value +fineract.remote-job-message-handler.kafka.consumer.extra-properties-key-value-separator=${FINERACT_REMOTE_JOB_MESSAGE_HANDLER_KAFKA_CONSUMER_EXTRA_PROPERTIES_SEPARATOR:=} +#defines the item separator, e.g.: key1=value1|key2=value2 +fineract.remote-job-message-handler.kafka.consumer.extra-properties-separator=${FINERACT_REMOTE_JOB_MESSAGE_HANDLER_KAFKA_CONSUMER_EXTRA_PROPERTIES_SEPARATOR:|} +#holds list of key value pairs using the above defined separators: key1=value1|key2=value2|...|keyn=valuen +fineract.remote-job-message-handler.kafka.consumer.extra-properties=${FINERACT_REMOTE_JOB_MESSAGE_HANDLER_KAFKA_CONSUMER_EXTRA_PROPERTIES:} +fineract.remote-job-message-handler.kafka.producer.extra-properties-separator=${FINERACT_REMOTE_JOB_MESSAGE_HANDLER_KAFKA_PRODUCER_EXTRA_PROPERTIES_SEPARATOR:|} +fineract.remote-job-message-handler.kafka.producer.extra-properties-key-value-separator=${FINERACT_REMOTE_JOB_MESSAGE_HANDLER_KAFKA_PRODUCER_EXTRA_PROPERTIES_SEPARATOR:=} +fineract.remote-job-message-handler.kafka.producer.extra-properties=${FINERACT_REMOTE_JOB_MESSAGE_HANDLER_KAFKA_PRODUCER_EXTRA_PROPERTIES:} +fineract.remote-job-message-handler.kafka.admin.extra-properties-separator=${FINERACT_REMOTE_JOB_MESSAGE_HANDLER_KAFKA_ADMIN_EXTRA_PROPERTIES_SEPARATOR:|} +fineract.remote-job-message-handler.kafka.admin.extra-properties-key-value-separator=${FINERACT_REMOTE_JOB_MESSAGE_HANDLER_KAFKA_ADMIN_EXTRA_PROPERTIES_KEY_VALUE_SEPARATOR:=} +fineract.remote-job-message-handler.kafka.admin.extra-properties=${FINERACT_REMOTE_JOB_MESSAGE_HANDLER_KAFKA_ADMIN_EXTRA_PROPERTIES:} + fineract.events.external.enabled=${FINERACT_EXTERNAL_EVENTS_ENABLED:false} fineract.events.external.producer.jms.enabled=${FINERACT_EXTERNAL_EVENTS_PRODUCER_JMS_ENABLED:false} fineract.events.external.producer.jms.async-send-enabled=${FINERACT_EXTERNAL_EVENTS_PRODUCER_JMS_ASYNC_SEND_ENABLED:false} @@ -86,6 +106,21 @@ fineract.events.external.producer.jms.producer-count=${FINERACT_EXTERNAL_EVENTS_ fineract.events.external.producer.jms.thread-pool-task-executor-core-pool-size=${FINERACT_EVENT_TASK_EXECUTOR_CORE_POOL_SIZE:10} fineract.events.external.producer.jms.thread-pool-task-executor-max-pool-size=${FINERACT_EVENT_TASK_EXECUTOR_MAX_POOL_SIZE:100} +fineract.events.external.producer.kafka.enabled=${FINERACT_EXTERNAL_EVENTS_KAFKA_ENABLED:false} +fineract.events.external.producer.kafka.timeout-in-seconds=${FINERACT_EXTERNAL_EVENTS_KAFKA_TIMEOUT_IN_SECONDS:10} +fineract.events.external.producer.kafka.topic.auto-create=${FINERACT_EXTERNAL_EVENTS_KAFKA_TOPIC_AUTO_CREATE:true} +fineract.events.external.producer.kafka.topic.name=${FINERACT_EXTERNAL_EVENTS_KAFKA_TOPIC_NAME:external-events} +fineract.events.external.producer.kafka.topic.replicas=${FINERACT_EXTERNAL_EVENTS_KAFKA_TOPIC_REPLICAS:1} +fineract.events.external.producer.kafka.topic.partitions=${FINERACT_EXTERNAL_EVENTS_KAFKA_TOPIC_PARTITIONS:10} +fineract.events.external.producer.kafka.bootstrap-servers=${FINERACT_EXTERNAL_EVENTS_KAFKA_BOOTSTRAP_SERVERS:localhost:9092} +fineract.events.external.producer.kafka.producer.extra-properties-separator=${FINERACT_EXTERNAL_EVENTS_KAFKA_PRODUCER_EXTRA_PROPERTIES_SEPARATOR:|} +fineract.events.external.producer.kafka.producer.extra-properties-key-value-separator=${FINERACT_EXTERNAL_EVENTS_KAFKA_PRODUCER_EXTRA_PROPERTIES_SEPARATOR:=} +fineract.events.external.producer.kafka.producer.extra-properties=${FINERACT_EXTERNAL_EVENTS_KAFKA_PRODUCER_EXTRA_PROPERTIES:linger.ms=10|batch.size=16384} +fineract.events.external.producer.kafka.admin.extra-properties-separator=${FINERACT_EXTERNAL_EVENTS_KAFKA_ADMIN_EXTRA_PROPERTIES_SEPARATOR:|} +fineract.events.external.producer.kafka.admin.extra-properties-key-value-separator=${FINERACT_EXTERNAL_EVENTS_KAFKA_ADMIN_EXTRA_PROPERTIES_KEY_VALUE_SEPARATOR:=} +fineract.events.external.producer.kafka.admin.extra-properties=${FINERACT_EXTERNAL_EVENTS_KAFKA_ADMIN_EXTRA_PROPERTIES:} + + fineract.task-executor.default-task-executor-core-pool-size=${FINERACT_DEFAULT_TASK_EXECUTOR_CORE_POOL_SIZE:10} fineract.task-executor.default-task-executor-max-pool-size=${FINERACT_DEFAULT_TASK_EXECUTOR_MAX_POOL_SIZE:100} diff --git a/fineract-provider/src/test/java/org/apache/fineract/infrastructure/event/external/producer/kafka/KafkaExternalEventProducerTest.java b/fineract-provider/src/test/java/org/apache/fineract/infrastructure/event/external/producer/kafka/KafkaExternalEventProducerTest.java new file mode 100644 index 000000000..21ab34de8 --- /dev/null +++ b/fineract-provider/src/test/java/org/apache/fineract/infrastructure/event/external/producer/kafka/KafkaExternalEventProducerTest.java @@ -0,0 +1,137 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package org.apache.fineract.infrastructure.event.external.producer.kafka; + +import static org.mockito.Mockito.times; + +import java.nio.charset.Charset; +import java.util.List; +import java.util.Map; +import java.util.concurrent.CompletableFuture; +import org.apache.fineract.infrastructure.core.config.FineractProperties; +import org.jetbrains.annotations.NotNull; +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.extension.ExtendWith; +import org.mockito.Mock; +import org.mockito.Mockito; +import org.mockito.junit.jupiter.MockitoExtension; +import org.springframework.kafka.core.KafkaTemplate; +import org.springframework.kafka.support.SendResult; + +@ExtendWith(MockitoExtension.class) +class KafkaExternalEventProducerTest { + + public static final String TOPIC_NAME = "unit-test"; + @Mock + private KafkaTemplate kafkaTemplate; + + @Mock + private SendResult sendResult1; + + @Mock + private SendResult sendResult2; + + @Mock + private SendResult sendResult3; + + private static final byte[] FIRST = "first".getBytes(Charset.defaultCharset()); + private static final byte[] SECOND = "second".getBytes(Charset.defaultCharset()); + private static final byte[] THIRD = "third".getBytes(Charset.defaultCharset()); + + @Test + public void testSendOK() { + // given + KafkaExternalEventProducer underTest = new KafkaExternalEventProducer(kafkaTemplate, createProperties()); + Mockito.when(kafkaTemplate.send(TOPIC_NAME, 1L, FIRST)).thenReturn(CompletableFuture.completedFuture(sendResult1)); + Mockito.when(kafkaTemplate.send(TOPIC_NAME, 1L, SECOND)).thenReturn(CompletableFuture.completedFuture(sendResult2)); + Mockito.when(kafkaTemplate.send(TOPIC_NAME, 2L, THIRD)).thenReturn(CompletableFuture.completedFuture(sendResult2)); + + // when + underTest.sendEvents(Map.of(1L, List.of(FIRST, SECOND), 2L, List.of(THIRD))); + + // then + Mockito.verify(kafkaTemplate, times(1)).send(TOPIC_NAME, 1L, FIRST); + Mockito.verify(kafkaTemplate, times(1)).send(TOPIC_NAME, 1L, SECOND); + Mockito.verify(kafkaTemplate, times(1)).send(TOPIC_NAME, 2L, THIRD); + Mockito.verifyNoMoreInteractions(kafkaTemplate); + } + + @Test + public void testSendOneFails() { + // given + KafkaExternalEventProducer underTest = new KafkaExternalEventProducer(kafkaTemplate, createProperties()); + Mockito.when(kafkaTemplate.send(TOPIC_NAME, 1L, FIRST)).thenReturn(CompletableFuture.completedFuture(sendResult1)); + Mockito.when(kafkaTemplate.send(TOPIC_NAME, 1L, SECOND)).thenReturn(CompletableFuture.completedFuture(sendResult2)); + Mockito.when(kafkaTemplate.send(TOPIC_NAME, 2L, THIRD)) + .thenReturn(CompletableFuture.failedFuture(new RuntimeException("Kafka error"))); + + // when + Assertions.assertThrows(RuntimeException.class, () -> underTest.sendEvents(Map.of(1L, List.of(FIRST, SECOND), 2L, List.of(THIRD)))); + + // then + Mockito.verify(kafkaTemplate, times(1)).send(TOPIC_NAME, 1L, FIRST); + Mockito.verify(kafkaTemplate, times(1)).send(TOPIC_NAME, 1L, SECOND); + Mockito.verify(kafkaTemplate, times(1)).send(TOPIC_NAME, 2L, THIRD); + Mockito.verifyNoMoreInteractions(kafkaTemplate); + } + + @Test + public void testTimeOut() { + // given + KafkaExternalEventProducer underTest = new KafkaExternalEventProducer(kafkaTemplate, createProperties()); + Mockito.when(kafkaTemplate.send(TOPIC_NAME, 1L, FIRST)).thenReturn(CompletableFuture.completedFuture(sendResult1)); + Mockito.when(kafkaTemplate.send(TOPIC_NAME, 1L, SECOND)).thenReturn(CompletableFuture.completedFuture(sendResult2)); + Mockito.when(kafkaTemplate.send(TOPIC_NAME, 2L, THIRD)).thenReturn(new CompletableFuture<>()); + + // when + Assertions.assertThrows(RuntimeException.class, () -> underTest.sendEvents(Map.of(1L, List.of(FIRST, SECOND), 2L, List.of(THIRD)))); + + // then + Mockito.verify(kafkaTemplate, times(1)).send(TOPIC_NAME, 1L, FIRST); + Mockito.verify(kafkaTemplate, times(1)).send(TOPIC_NAME, 1L, SECOND); + Mockito.verify(kafkaTemplate, times(1)).send(TOPIC_NAME, 2L, THIRD); + Mockito.verifyNoMoreInteractions(kafkaTemplate); + } + + @NotNull + private static FineractProperties createProperties() { + FineractProperties props = new FineractProperties(); + + FineractProperties.FineractEventsProperties fineractEventsProperties = new FineractProperties.FineractEventsProperties(); + props.setEvents(fineractEventsProperties); + + FineractProperties.FineractExternalEventsProperties externalEventsProperties = new FineractProperties.FineractExternalEventsProperties(); + fineractEventsProperties.setExternal(externalEventsProperties); + + FineractProperties.FineractExternalEventsProducerProperties producer = new FineractProperties.FineractExternalEventsProducerProperties(); + externalEventsProperties.setProducer(producer); + + FineractProperties.FineractExternalEventsProducerKafkaProperties kafkaProperties = new FineractProperties.FineractExternalEventsProducerKafkaProperties(); + producer.setKafka(kafkaProperties); + + FineractProperties.KafkaTopicProperties kafkaTopicProperties = new FineractProperties.KafkaTopicProperties(); + kafkaProperties.setTopic(kafkaTopicProperties); + kafkaProperties.setTimeoutInSeconds(1); + + kafkaTopicProperties.setName(TOPIC_NAME); + return props; + } + +} diff --git a/integration-tests/build.gradle b/integration-tests/build.gradle index d6795ddb3..1703ffcdd 100644 --- a/integration-tests/build.gradle +++ b/integration-tests/build.gradle @@ -90,9 +90,11 @@ cargoRunLocal.dependsOn ':fineract-war:war' cargoStartLocal.dependsOn ':fineract-war:war' cargoStartLocal.mustRunAfter 'testClasses' -test { - dependsOn(cargoStartLocal) - finalizedBy cargoStopLocal +if (!project.hasProperty('cargoDisabled')) { + test { + dependsOn(cargoStartLocal) + finalizedBy cargoStopLocal + } } // NOTE: Gradle suggested these dependencies