Scheduler setup

This guide will walk you through the process of setting up the Scheduler service.

Infrastructure prerequisites

• MongoDB
• Kafka
• OpenID Connect Settings

Dependencies

• MongoDB database
• Ability to connect to a Kafka instance used by the FlowX Engine
• Scheduler service account - required for using Start Timer event node - see here for more details.

The service comes with most of the required configuration properties pre-filled. However, certain custom environment variables need to be set up.

This service needs to connect to a Mongo database that has replicas, in order to work correctly.

Scheduler configuration

Scheduler

scheduler:
  thread-count: 30  # Configure the number of threads to be used for sending expired messages.
  callbacks-thread-count: 60 # Configure the number of threads for handling Kafka responses, whether the message was successfully sent or not
  cronExpression: "*/10 * * * * *" #every 10 seconds
  retry: # new retry mechanism
    max-attempts: 3
    seconds: 1
    thread-count: 3
    cronExpression: "*/10 * * * * *" #every 10 seconds
  cleanup:
    cronExpression: "*/25 * * * * *" #every 25 seconds

• SCHEDULER_THREAD_COUNT: Used to configure the number of threads to be used for sending expired.
• SCHEDULER_CALLBACKS_THREAD_COUNT: Used to configure the number of threads for handling Kafka responses, whether the message was successfully sent or not.

The “scheduler.cleanup.cronExpression” is valid for both scheduler and timer event scheduler.

Retry mechanism

• SCHEDULER_RETRY_THREAD_COUNT: Specify the number of threads to use for resending messages that need to be retried.
• SCHEDULER_RETRY_MAX_ATTEMPTS: This configuration parameter sets the number of retry attempts. For instance, if it’s set to 3, it means that the system will make a maximum of three retry attempts for message resending.
• SCHEDULER_RETRY_SECONDS: This configuration parameter defines the time interval, in seconds, for retry attempts. For example, when set to 1, it indicates that the system will retry the operation after a one-second delay.

Cleanup

• SCHEDULER_CLEANUP_CRONEXPRESSION: It specifies how often, in seconds, events that have already been processed should be cleaned up from the database.

Recovery mechanism

flowx:
  timer-calculator:
    delay-max-repetitions: 1000000

You have a “next execution” set for 10:25, and the cycle step is 10 minutes. If the instance goes down for 2 hours, the next execution time should be 12:25, not 10:35. To calculate this, you add 10 minutes repeatedly to 10:25 until you reach the current time. So, it would be 10:25 + 10 min + 10 min + 10 min, until you reach the current time of 12:25. This ensures that the next execution time is adjusted correctly after the downtime.

• FLOWX_TIMER_CALCULATOR_DELAY_MAX_REPETITIONS: This means that, for example, if our cycle step is set to one second and the system experiences a downtime of two weeks, which is equivalent to 1,209,600 seconds, and we have the “max repetitions” set to 1,000,000, it will attempt to calculate the next schedule. However, when it reaches the maximum repetitions, an exception is thrown, making it impossible to calculate the next schedule. As a result, the entry remains locked and needs to be rescheduled. This scenario represents a critical case where the system experiences extended downtime, and the cycle step is very short (e.g., 1 second), leading to the inability to determine the next scheduled event.

Timer event scheduler

Configuration for Timer Event scheduler designed to manage timer events. Similar configuration to scheduler.

timer-event-scheduler:
  thread-count: 30
  callbacks-thread-count: 60
  cronExpression: "*/1 * * * * *" #every 1 seconds
  retry:
    max-attempts: 3
    seconds: 1
    thread-count: 3
    cronExpression: "*/5 * * * * *" #every 5 seconds

OpenID connect settings

Changed in v5.5.0 — The default SECURITY_TYPE has changed from oauth2 to jwt-public-key. Opaque-token introspection has been removed. Service account client IDs now follow the flowx-{service-name}-sa pattern.

Environment variable	Description	Default value
SECURITY_TYPE	Security type	jwt-public-key
SECURITY_OAUTH2_BASE_SERVER_URL	Base URL of the OpenID server
SECURITY_PATHAUTHORIZATIONS_0_PATH	Security path pattern	/api/**
SECURITY_PATHAUTHORIZATIONS_0_ROLESALLOWED	Roles allowed for path access	ANY_AUTHENTICATED_USER
SPRING_SECURITY_OAUTH2_CLIENT_REGISTRATION_MAINIDENTITY_CLIENTID	Client ID for service account	flowx-scheduler-core-sa
SPRING_SECURITY_OAUTH2_CLIENT_REGISTRATION_MAINIDENTITY_CLIENTSECRET	Client secret for service account
SPRING_SECURITY_OAUTH2_CLIENT_PROVIDER_MAINAUTHPROVIDER_TOKEN_URI	Provider token URI	${SECURITY_OAUTH2_BASE_SERVER_URL}/realms/${SECURITY_OAUTH2_SA_REALM}/protocol/openid-connect/token

The service account is essential for enabling the Start Timer event node.

More details about the necessary service account, here:

Scheduler service account

Configuring datasoruce (MongoDB)

The MongoDB database is used to persist scheduled messages until they are sent back. The following configurations need to be set using environment variables:

• SPRING_DATA_MONGODB_URI: The URI for the MongoDB database.

Configuring Kafka

Core Kafka settings

Environment Variable	Description	Default Value
SPRING_KAFKA_BOOTSTRAPSERVERS	Address of the Kafka server	localhost:9092
SPRING_KAFKA_SECURITY_PROTOCOL	Security protocol for Kafka connections	PLAINTEXT
SPRING_KAFKA_CONSUMER_GROUPID	Consumer group identifier	scheduler-consumer
KAFKA_MESSAGE_MAX_BYTES	Maximum message size (bytes)	52428800 (50 MB)
KAFKA_AUTHEXCEPTIONRETRYINTERVAL	Retry interval after authorization exceptions (seconds)	10

Consumer configuration

Environment Variable	Description	Default Value
KAFKA_CONSUMER_THREADS	Number of Kafka consumer threads	1
KAFKA_CONSUMER_SCHEDULEDTIMEREVENTS_THREADS	Number of threads for starting Timer Events	1
KAFKA_CONSUMER_SCHEDULEDTIMEREVENTS_GROUPID	Consumer group for starting timer events	scheduled-timer-events
KAFKA_CONSUMER_STOPSCHEDULEDTIMEREVENTS_THREADS	Number of threads for stopping Timer Events	1
KAFKA_CONSUMER_STOPSCHEDULEDTIMEREVENTS_GROUPID	Consumer group for stopping timer events	stop-scheduled-timer-events

OAuth authentication (when using SASL_PLAINTEXT)

Environment Variable	Description	Default Value
KAFKA_OAUTH_CLIENT_ID	OAuth client ID	kafka
KAFKA_OAUTH_CLIENT_SECRET	OAuth client secret	kafka-secret
KAFKA_OAUTH_TOKEN_ENDPOINT_URI	OAuth token endpoint	kafka.auth.localhost

Topic naming configuration

Environment Variable	Description	Default Value
KAFKA_TOPIC_NAMING_PACKAGE	Package prefix for topic names	ai.flowx.
KAFKA_TOPIC_NAMING_ENVIRONMENT	Environment segment for topic names
KAFKA_TOPIC_NAMING_VERSION	Version suffix for topic names	.v1
KAFKA_TOPIC_NAMING_SEPARATOR	Primary separator for topic names	.
KAFKA_TOPIC_NAMING_SEPARATOR2	Secondary separator for topic names	-

Kafka topics

Schedule topics

Environment Variable	Description	Default Value
KAFKA_TOPIC_SCHEDULE_IN_SET	Receives scheduled message setting requests	ai.flowx.core.trigger.set.schedule.v1
KAFKA_TOPIC_SCHEDULE_IN_STOP	Handles requests to terminate scheduled messages	ai.flowx.core.trigger.stop.schedule.v1

Timer events topics

Environment Variable	Description	Default Value
KAFKA_TOPIC_SCHEDULEDTIMEREVENTS_IN_SET	Topic for setting timer events	ai.flowx.core.trigger.set.timer-event-schedule.v1
KAFKA_TOPIC_SCHEDULEDTIMEREVENTS_IN_STOP	Topic for stopping timer events	ai.flowx.core.trigger.stop.timer-event-schedule.v1

Make sure the topics configured for this service don’t follow the engine pattern.

Configuring logging

The following environment variables can be set to control log levels:

• LOGGING_LEVEL_ROOT: Root Spring Boot microservice logs (Default: INFO)
• LOGGING_LEVEL_APP: App level logs (Default: DEBUG)