diff --git a/auth-service/Dockerfile b/auth-service/Dockerfile index 1693e41..5626902 100644 --- a/auth-service/Dockerfile +++ b/auth-service/Dockerfile @@ -3,8 +3,7 @@ FROM eclipse-temurin:21-jre WORKDIR /app COPY auth-service/target/*.jar app.jar -COPY database/migrations /app/database/migrations EXPOSE 8082 -ENTRYPOINT ["java","-jar","app.jar"] \ No newline at end of file +ENTRYPOINT ["java", "-jar", "app.jar"] \ No newline at end of file diff --git a/auth-service/src/main/resources/application.yaml b/auth-service/src/main/resources/application.yaml index df8885d..86f1fd6 100644 --- a/auth-service/src/main/resources/application.yaml +++ b/auth-service/src/main/resources/application.yaml @@ -20,6 +20,7 @@ spring: properties: hibernate: format_sql: true + default_schema: auth flyway: enabled: true diff --git a/auth-service/target/classes/application.yaml b/auth-service/target/classes/application.yaml index df8885d..86f1fd6 100644 --- a/auth-service/target/classes/application.yaml +++ b/auth-service/target/classes/application.yaml @@ -20,6 +20,7 @@ spring: properties: hibernate: format_sql: true + default_schema: auth flyway: enabled: true diff --git a/database/migrations/V2__create_worker_tasks.sql b/database/migrations/V2__create_worker_tasks.sql new file mode 100644 index 0000000..f7ffc40 --- /dev/null +++ b/database/migrations/V2__create_worker_tasks.sql @@ -0,0 +1,20 @@ +CREATE TABLE worker_tasks ( + id UUID PRIMARY KEY DEFAULT gen_random_uuid(), + workflow_execution_id UUID, + node_id UUID, + type VARCHAR(50) NOT NULL, + payload TEXT, + status VARCHAR(20) NOT NULL DEFAULT 'PENDING', + result TEXT, + error TEXT, + created_at TIMESTAMP, + updated_at TIMESTAMP +); + +CREATE TABLE dead_letter_tasks ( + id UUID PRIMARY KEY DEFAULT gen_random_uuid(), + original_topic VARCHAR(255) NOT NULL, + task_payload TEXT NOT NULL, + exception_message TEXT, + failed_at TIMESTAMP NOT NULL +); \ No newline at end of file diff --git a/database/migrations/V3__create_worker_tasks.sql b/database/migrations/V3__create_worker_tasks.sql deleted file mode 100644 index 39bac3c..0000000 --- a/database/migrations/V3__create_worker_tasks.sql +++ /dev/null @@ -1,10 +0,0 @@ -CREATE TABLE worker_tasks ( - id VARCHAR(36) PRIMARY KEY, - type VARCHAR(50) NOT NULL, - payload TEXT NOT NULL, - status VARCHAR(20) NOT NULL DEFAULT 'PENDING', - result TEXT, - error TEXT, - created_at TIMESTAMP NOT NULL DEFAULT NOW(), - updated_at TIMESTAMP NOT NULL DEFAULT NOW() -); \ No newline at end of file diff --git a/database/migrations/V4__add_worker_task_execution_columns.sql b/database/migrations/V4__add_worker_task_execution_columns.sql deleted file mode 100644 index 872ab34..0000000 --- a/database/migrations/V4__add_worker_task_execution_columns.sql +++ /dev/null @@ -1,6 +0,0 @@ -ALTER TABLE worker_tasks - ADD COLUMN IF NOT EXISTS workflow_execution_id VARCHAR(36), - ADD COLUMN IF NOT EXISTS node_id VARCHAR(36); - -ALTER TABLE worker_tasks - ALTER COLUMN payload DROP NOT NULL; diff --git a/database/migrations/V2__create_workflow_schema.sql b/database/migrations/V4__create_workflow_schema.sql similarity index 100% rename from database/migrations/V2__create_workflow_schema.sql rename to database/migrations/V4__create_workflow_schema.sql diff --git a/database/migrations/V5__create_worker_tables.sql b/database/migrations/V5__create_worker_tables.sql deleted file mode 100644 index c5d1dfa..0000000 --- a/database/migrations/V5__create_worker_tables.sql +++ /dev/null @@ -1,22 +0,0 @@ --- Create worker_tasks table -CREATE TABLE IF NOT EXISTS worker_tasks ( - id UUID PRIMARY KEY, - type VARCHAR(50) NOT NULL, - payload TEXT, - status VARCHAR(20) NOT NULL DEFAULT 'PENDING', - result TEXT, - error TEXT, - workflow_execution_id UUID, - node_id UUID, - created_at TIMESTAMP NOT NULL DEFAULT NOW(), - updated_at TIMESTAMP NOT NULL DEFAULT NOW() -); - --- Create dead_letter_tasks table -CREATE TABLE IF NOT EXISTS dead_letter_tasks ( - id UUID PRIMARY KEY, - original_topic VARCHAR(255) NOT NULL, - task_payload TEXT NOT NULL, - exception_message TEXT, - failed_at TIMESTAMP NOT NULL -); \ No newline at end of file diff --git a/docker-compose.yml b/docker-compose.yml index 4810e9d..60117d9 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -28,7 +28,7 @@ services: ports: - "8083:8082" volumes: - - ./database/auth:/app/database/migrations + - ./database/migrations:/app/database/migrations db: image: postgres:16-alpine @@ -67,10 +67,7 @@ services: - ./database:/app/database worker-service: - build: - context: . - dockerfile: worker-service/Dockerfile - image: flowforge/worker-service:local + image: aswinrj/worker-service container_name: worker-service restart: always depends_on: @@ -80,6 +77,8 @@ services: - mailpit ports: - "8084:8084" + volumes: + - ./database/migrations:/app/database/migrations kafka: image: apache/kafka:latest @@ -92,7 +91,7 @@ services: KAFKA_NODE_ID: 1 KAFKA_PROCESS_ROLES: broker,controller KAFKA_LISTENERS: PLAINTEXT://:9092,PLAINTEXT_HOST://:29092,CONTROLLER://:9093 - KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://flowforge-kafka:9092,PLAINTEXT_HOST://localhost:29092 + KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka:9092,PLAINTEXT_HOST://localhost:9092 KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT,CONTROLLER:PLAINTEXT KAFKA_INTER_BROKER_LISTENER_NAME: PLAINTEXT KAFKA_CONTROLLER_LISTENER_NAMES: CONTROLLER @@ -104,4 +103,4 @@ services: volumes: postgres_data: - kafka_data: + kafka_data: \ No newline at end of file diff --git a/worker-service/src/main/java/com/flowforge/worker_service/config/KafkaConfig.java b/worker-service/src/main/java/com/flowforge/worker_service/config/KafkaConfig.java index 3e68c92..4518a1c 100644 --- a/worker-service/src/main/java/com/flowforge/worker_service/config/KafkaConfig.java +++ b/worker-service/src/main/java/com/flowforge/worker_service/config/KafkaConfig.java @@ -10,6 +10,8 @@ import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory; import org.springframework.kafka.core.ConsumerFactory; import org.springframework.kafka.core.DefaultKafkaConsumerFactory; +import org.springframework.scheduling.TaskScheduler; +import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler; import java.util.HashMap; import java.util.Map; @@ -35,4 +37,13 @@ public ConcurrentKafkaListenerContainerFactory kafkaListenerCont factory.setConsumerFactory(consumerFactory); return factory; } + + @Bean + public TaskScheduler taskScheduler() { + ThreadPoolTaskScheduler scheduler = new ThreadPoolTaskScheduler(); + scheduler.setPoolSize(5); + scheduler.setThreadNamePrefix("retry-topic-scheduler-"); + scheduler.initialize(); + return scheduler; + } } diff --git a/worker-service/src/main/java/com/flowforge/worker_service/domain/model/WorkerTask.java b/worker-service/src/main/java/com/flowforge/worker_service/domain/model/WorkerTask.java index 58d5a02..52c5218 100644 --- a/worker-service/src/main/java/com/flowforge/worker_service/domain/model/WorkerTask.java +++ b/worker-service/src/main/java/com/flowforge/worker_service/domain/model/WorkerTask.java @@ -1,6 +1,7 @@ package com.flowforge.worker_service.domain.model; import com.flowforge.worker_service.domain.enums.TaskStatus; +import jakarta.persistence.Table; import lombok.AllArgsConstructor; import lombok.Builder; import lombok.Data; diff --git a/worker-service/src/main/resources/application.yaml b/worker-service/src/main/resources/application.yaml index a262725..0a50286 100644 --- a/worker-service/src/main/resources/application.yaml +++ b/worker-service/src/main/resources/application.yaml @@ -18,11 +18,13 @@ spring: flyway: enabled: true - locations: filesystem:/app/database/migrations + locations: classpath:db/migration + schemas: worker + default-schema: worker create-schemas: true kafka: - bootstrap-servers: flowforge-kafka:9092 + bootstrap-servers: flowforge-kafka:29092 producer: key-serializer: org.apache.kafka.common.serialization.StringSerializer value-serializer: org.springframework.kafka.support.serializer.JsonSerializer