feat: enhance flow initialization with Kafka topic creation and optimize producer configurations

yakud · yakud · commit 4e945eeab8b9 · 2025-09-21T14:29:30.000+03:00
diff --git a/.changeset/pre.json b/.changeset/pre.json
@@ -32,6 +32,7 @@
     "rude-parts-heal",
     "sixty-humans-smile",
     "tiny-things-report",
-    "tired-days-sleep"
+    "tired-days-sleep",
+    "tough-rice-spend"
   ]
 }
diff --git a/apps/chaingraph-backend/CHANGELOG.md b/apps/chaingraph-backend/CHANGELOG.md
@@ -1,5 +1,11 @@
 # @badaitech/chaingraph-backend
 
+## 0.5.5-dev.19
+
+### Patch Changes
+
+- feat: enhance flow initialization with Kafka topic creation and optimize producer configurations
+
 ## 0.5.5-dev.18
 
 ### Patch Changes
diff --git a/apps/chaingraph-backend/package.json b/apps/chaingraph-backend/package.json
@@ -1,7 +1,7 @@
 {
   "name": "@badaitech/chaingraph-backend",
   "type": "module",
-  "version": "0.5.5-dev.18",
+  "version": "0.5.5-dev.19",
   "private": false,
   "description": "Backend server for the Chaingraph project",
   "license": "BUSL-1.1",
diff --git a/apps/chaingraph-execution-api/CHANGELOG.md b/apps/chaingraph-execution-api/CHANGELOG.md
@@ -1,5 +1,15 @@
 # @badaitech/chaingraph-execution-api
 
+## 0.5.5-dev.19
+
+### Patch Changes
+
+- feat: enhance flow initialization with Kafka topic creation and optimize producer configurations
+- Updated dependencies
+  - @badaitech/chaingraph-executor@0.5.5-dev.19
+  - @badaitech/chaingraph-types@0.5.5-dev.19
+  - @badaitech/chaingraph-trpc@0.5.5-dev.19
+
 ## 0.5.5-dev.18
 
 ### Patch Changes
diff --git a/apps/chaingraph-execution-api/package.json b/apps/chaingraph-execution-api/package.json
@@ -1,7 +1,7 @@
 {
   "name": "@badaitech/chaingraph-execution-api",
   "type": "module",
-  "version": "0.5.5-dev.18",
+  "version": "0.5.5-dev.19",
   "private": false,
   "description": "Chaingraph tRPC Server - Scalable API server for Chaingraph execution management",
   "license": "BUSL-1.1",
diff --git a/apps/chaingraph-execution-worker/CHANGELOG.md b/apps/chaingraph-execution-worker/CHANGELOG.md
@@ -1,5 +1,16 @@
 # @badaitech/chaingraph-execution-worker
 
+## 0.5.5-dev.19
+
+### Patch Changes
+
+- feat: enhance flow initialization with Kafka topic creation and optimize producer configurations
+- Updated dependencies
+  - @badaitech/chaingraph-executor@0.5.5-dev.19
+  - @badaitech/chaingraph-nodes@0.5.5-dev.19
+  - @badaitech/chaingraph-types@0.5.5-dev.19
+  - @badaitech/chaingraph-trpc@0.5.5-dev.19
+
 ## 0.5.5-dev.18
 
 ### Patch Changes
diff --git a/apps/chaingraph-execution-worker/package.json b/apps/chaingraph-execution-worker/package.json
@@ -1,7 +1,7 @@
 {
   "name": "@badaitech/chaingraph-execution-worker",
   "type": "module",
-  "version": "0.5.5-dev.18",
+  "version": "0.5.5-dev.19",
   "private": false,
   "description": "Chaingraph Execution Worker Service which connected to Kafka and process the flow executions",
   "license": "BUSL-1.1",
diff --git a/apps/chaingraph-execution-worker/src/index.ts b/apps/chaingraph-execution-worker/src/index.ts
@@ -8,22 +8,36 @@
 
 import cluster from 'node:cluster'
 import process from 'node:process'
+import { createTopicsIfNotExist } from '@badaitech/chaingraph-executor/server'
 import { config } from './config'
 import { logger } from './logger'
 import { startMaster } from './master'
 import { startWorker } from './worker'
 
 // Determine if this is the master process or a worker
-if (cluster.isPrimary) {
-  logger.info('🚀 Starting Chaingraph Execution Worker Service')
-  logger.info({
-    masterPid: process.pid,
-    workers: config.workers.count,
-    mode: config.executionMode,
-  }, 'Service configuration')
+async function main() {
+  if (cluster.isPrimary) {
+    logger.info('🚀 Starting Chaingraph Execution Worker Service')
+    logger.info({
+      masterPid: process.pid,
+      workers: config.workers.count,
+      mode: config.executionMode,
+    }, 'Service configuration')
 
-  startMaster()
-} else {
-  // This is a worker process
-  startWorker()
+    // Create Kafka topics if in distributed mode
+    if (config.executionMode === 'distributed') {
+      logger.info('Creating Kafka topics if they don\'t exist')
+      await createTopicsIfNotExist()
+    }
+
+    startMaster()
+  } else {
+    // This is a worker process
+    await startWorker()
+  }
 }
+
+main().catch((error) => {
+  logger.error({ error }, 'Failed to start service')
+  process.exit(1)
+})
diff --git a/apps/chaingraph-frontend/CHANGELOG.md b/apps/chaingraph-frontend/CHANGELOG.md
@@ -1,5 +1,17 @@
 # @badaitech/chaingraph-frontend
 
+## 0.5.5-dev.19
+
+### Patch Changes
+
+- feat: enhance flow initialization with Kafka topic creation and optimize producer configurations
+- Updated dependencies
+  - @badaitech/chaingraph-executor@0.5.5-dev.19
+  - @badaitech/chaingraph-nodes@0.5.5-dev.19
+  - @badaitech/chaingraph-types@0.5.5-dev.19
+  - @badaitech/chaingraph-trpc@0.5.5-dev.19
+  - @badaitech/badai-api@0.5.5-dev.19
+
 ## 0.5.5-dev.18
 
 ### Patch Changes
diff --git a/apps/chaingraph-frontend/package.json b/apps/chaingraph-frontend/package.json
@@ -1,7 +1,7 @@
 {
   "name": "@badaitech/chaingraph-frontend",
   "type": "module",
-  "version": "0.5.5-dev.18",
+  "version": "0.5.5-dev.19",
   "private": false,
   "description": "Frontend application for the Chaingraph project",
   "license": "BUSL-1.1",
diff --git a/docker-compose.yaml b/docker-compose.yaml
@@ -14,5 +14,28 @@ services:
       - postgres_data:/var/lib/postgresql/data
     restart: always
 
+  kafka-broker-1:
+    image: apache/kafka:latest
+    hostname: broker
+    container_name: broker
+    ports:
+      - '9092:9092'
+    environment:
+      KAFKA_BROKER_ID: 1
+      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT,CONTROLLER:PLAINTEXT
+      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://broker:29092,PLAINTEXT_HOST://localhost:9092
+      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
+      KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS: 0
+      KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 1
+      KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 1
+      KAFKA_PROCESS_ROLES: broker,controller
+      KAFKA_NODE_ID: 1
+      KAFKA_CONTROLLER_QUORUM_VOTERS: 1@broker:29093
+      KAFKA_LISTENERS: PLAINTEXT://broker:29092,CONTROLLER://broker:29093,PLAINTEXT_HOST://0.0.0.0:9092
+      KAFKA_INTER_BROKER_LISTENER_NAME: PLAINTEXT
+      KAFKA_CONTROLLER_LISTENER_NAMES: CONTROLLER
+      KAFKA_LOG_DIRS: /tmp/kraft-combined-logs
+      CLUSTER_ID: MkU3OEVBNTcwNTJENDM2Qk
+
 volumes:
   postgres_data:
diff --git a/packages/badai-api/CHANGELOG.md b/packages/badai-api/CHANGELOG.md
@@ -1,5 +1,11 @@
 # @badaitech/badai-api
 
+## 0.5.5-dev.19
+
+### Patch Changes
+
+- feat: enhance flow initialization with Kafka topic creation and optimize producer configurations
+
 ## 0.5.5-dev.18
 
 ### Patch Changes
diff --git a/packages/badai-api/package.json b/packages/badai-api/package.json
@@ -1,7 +1,7 @@
 {
   "name": "@badaitech/badai-api",
   "type": "module",
-  "version": "0.5.5-dev.18",
+  "version": "0.5.5-dev.19",
   "private": false,
   "description": "Bad AI GraphQL API client",
   "license": "BUSL-1.1",
diff --git a/packages/chaingraph-executor/package.json b/packages/chaingraph-executor/package.json
@@ -1,7 +1,7 @@
 {
   "name": "@badaitech/chaingraph-executor",
   "type": "module",
-  "version": "0.5.5-dev.18",
+  "version": "0.5.5-dev.19",
   "private": false,
   "description": "Chaingraph Message Bus for inter-service communication",
   "license": "BUSL-1.1",
diff --git a/packages/chaingraph-executor/server/implementations/distributed/KafkaEventBus.ts b/packages/chaingraph-executor/server/implementations/distributed/KafkaEventBus.ts
@@ -72,7 +72,7 @@ export class KafkaEventBus implements IEventBus {
         value,
         timestamp: Date.now().toString(),
       }],
-      acks: 1,
+      acks: -1,
       timeout: 30000,
     })
   }
@@ -91,10 +91,6 @@ export class KafkaEventBus implements IEventBus {
       heartbeatInterval: 3000,
       maxWaitTimeInMs: 1, // Ultra-low latency: reduced from 10ms to 1ms
       allowAutoTopicCreation: false,
-      maxBytesPerPartition: 1024 * 1024, // 1MB per partition
-      minBytes: 1, // Fetch as soon as data is available
-      maxInFlightRequests: 1, // Ensure order by limiting to 1 in-flight request
-      readUncommitted: true, // Read uncommitted for lower latency
       retry: {
         initialRetryTime: 100, // Start with 100ms retry delay
         maxRetryTime: 5000, // Max 5 seconds between retries
diff --git a/packages/chaingraph-executor/server/implementations/distributed/KafkaTaskQueue.ts b/packages/chaingraph-executor/server/implementations/distributed/KafkaTaskQueue.ts
@@ -31,15 +31,21 @@ export class KafkaTaskQueue implements ITaskQueue {
       this.producer = await getTaskProducer()
     }
 
+    logger.debug({
+      taskExecutionId: task.executionId,
+      topic: KafkaTopics.TASKS,
+    }, 'Publishing task to Kafka')
+
     const records = await this.producer.send({
       topic: KafkaTopics.TASKS,
       messages: [{
         key: task.executionId,
         value: safeSuperJSONStringify(task),
         timestamp: Date.now().toString(),
       }],
-      acks: 1,
-      timeout: 30000,
+      acks: 1, // Only wait for leader acknowledgment (faster)
+      timeout: 5000, // 5 second timeout for lower latency
+      compression: 0, // No compression for lowest latency
     })
 
     for (const record of records) {
@@ -70,14 +76,19 @@ export class KafkaTaskQueue implements ITaskQueue {
     const kafka = getKafkaClient()
     this.consumer = kafka.consumer({
       groupId: config.kafka.groupId.worker,
-      sessionTimeout: 30000, // Increased from 10000 for better stability
+      sessionTimeout: 30000,
       heartbeatInterval: 3000,
-      rebalanceTimeout: 30000, // Increased from 10000 for better stability
-      maxWaitTimeInMs: 1, // Reduced from 100ms to 1ms for ultra-low latency
+      rebalanceTimeout: 30000,
+      maxWaitTimeInMs: 100, // fetch.max.wait.ms equivalent - wait up to 100ms
+      minBytes: 1, // fetch.min.bytes equivalent - fetch even single bytes
+      maxBytes: 52428800, // fetch.max.bytes equivalent - 50MB max fetch
+      maxBytesPerPartition: 10485760, // max.partition.fetch.bytes equivalent - 10MB per partition
+      readUncommitted: true, // Read only committed messages
       allowAutoTopicCreation: false,
-      minBytes: 1, // Fetch as soon as data is available
-      maxInFlightRequests: 1, // Ensure order by limiting to 1 in-flight request
-      readUncommitted: true, // Read uncommitted for lower latency
+      retry: {
+        initialRetryTime: 100,
+        retries: 5,
+      },
     })
 
     await this.consumer.connect()
diff --git a/packages/chaingraph-executor/server/kafka/client.ts b/packages/chaingraph-executor/server/kafka/client.ts
@@ -58,7 +58,6 @@ export function getKafkaClient(): Kafka {
         initialRetryTime: 50, // Reduced from 100ms for faster recovery
         maxRetryTime: 1000, // Max 1 second between retries (faster than default 30s)
         retries: 5, // Reduced from 8 to fail faster and avoid long delays
-        multiplier: 1.2, // Gentler exponential backoff
       },
       // Connection timeouts for faster failure detection
       connectionTimeout: 3000, // 3 second connection timeout
diff --git a/packages/chaingraph-executor/server/kafka/producers/command-producer.ts b/packages/chaingraph-executor/server/kafka/producers/command-producer.ts
@@ -24,7 +24,7 @@ export async function getCommandProducer(): Promise<Producer> {
   if (!producer) {
     const kafka = getKafkaClient()
     producer = kafka.producer({
-      allowAutoTopicCreation: true,
+      allowAutoTopicCreation: false,
       idempotent: true,
       createPartitioner: Partitioners.DefaultPartitioner,
       maxInFlightRequests: 10, // Increased from 5 for higher throughput
@@ -66,8 +66,8 @@ export async function publishExecutionCommand(command: ExecutionCommand): Promis
         value: safeSuperJSONStringify(command),
         timestamp: Date.now().toString(),
       }],
-      acks: 1, // Only wait for leader acknowledgment (optimized for single broker)
-      timeout: 5000, // 5 second timeout (reduced from 30s for faster failure detection)
+      acks: -1, // Wait for all in-sync replicas to acknowledge
+      timeout: 30000, // 30 second timeout
     })
 
     logger.debug({
diff --git a/packages/chaingraph-executor/server/kafka/producers/event-producer.ts b/packages/chaingraph-executor/server/kafka/producers/event-producer.ts
@@ -34,7 +34,7 @@ export async function getEventProducer(): Promise<Producer> {
   if (!producer) {
     const kafka = getKafkaClient()
     producer = kafka.producer({
-      allowAutoTopicCreation: true,
+      allowAutoTopicCreation: false,
       idempotent: true,
       createPartitioner: Partitioners.DefaultPartitioner,
       maxInFlightRequests: 10, // Increased from 5 for higher throughput
diff --git a/packages/chaingraph-executor/server/kafka/producers/task-producer.ts b/packages/chaingraph-executor/server/kafka/producers/task-producer.ts
@@ -24,10 +24,15 @@ export async function getTaskProducer(): Promise<Producer> {
   if (!producer) {
     const kafka = getKafkaClient()
     producer = kafka.producer({
-      allowAutoTopicCreation: true,
-      idempotent: true,
+      allowAutoTopicCreation: false,
+      idempotent: false, // Disable for lower latency (no deduplication overhead)
       createPartitioner: Partitioners.DefaultPartitioner,
-      maxInFlightRequests: 10, // Increased from 5 for higher throughput
+      maxInFlightRequests: 5, // Balance between throughput and ordering
+      transactionTimeout: 30000,
+      retry: {
+        initialRetryTime: 100,
+        retries: 2, // Fewer retries for faster failure
+      },
     })
 
     // Store the connection promise to avoid multiple connection attempts
@@ -61,8 +66,9 @@ export async function publishExecutionTask(task: ExecutionTask): Promise<void> {
         value: safeSuperJSONStringify(task),
         timestamp: Date.now().toString(),
       }],
-      acks: 1, // Only wait for leader acknowledgment (optimized for single broker)
-      timeout: 5000, // 5 second timeout (reduced from 30s for faster failure detection)
+      acks: 1, // Only wait for leader acknowledgment (faster)
+      timeout: 5000, // 5 second timeout for lower latency
+      compression: 0, // No compression (0 = none) for lowest latency
     })
   } catch (error) {
     logger.error({ error, executionId: task.executionId }, 'Failed to publish task')
diff --git a/packages/chaingraph-executor/server/kafka/topics.ts b/packages/chaingraph-executor/server/kafka/topics.ts
@@ -23,7 +23,7 @@ interface TopicConfig {
 const topicConfigs: TopicConfig[] = [
   {
     topic: KafkaTopics.COMMANDS,
-    numPartitions: 10,
+    numPartitions: 100,
     replicationFactor: 1, // Changed from 3 to 1 for local development
     configEntries: [
       { name: 'retention.ms', value: '604800000' }, // 7 days
@@ -32,7 +32,7 @@ const topicConfigs: TopicConfig[] = [
   },
   {
     topic: KafkaTopics.EVENTS,
-    numPartitions: 50,
+    numPartitions: 100,
     replicationFactor: 1, // Changed from 3 to 1 for local development
     configEntries: [
       { name: 'retention.ms', value: '2592000000' }, // 30 days
diff --git a/packages/chaingraph-executor/server/workers/ExecutionWorker.ts b/packages/chaingraph-executor/server/workers/ExecutionWorker.ts
@@ -368,7 +368,7 @@ export class ExecutionWorker {
     try {
       const kafka = getKafkaClient()
       this.commandConsumer = kafka.consumer({
-        groupId: `${config.kafka.groupId.worker}-commands`, // Removed workerId for proper load balancing
+        groupId: `${config.kafka.groupId.worker}-commands-${this.workerId}`,
         sessionTimeout: 30000, // Increased from 10000 for better stability
         heartbeatInterval: 3000,
         // Ultra-low latency optimizations
diff --git a/packages/chaingraph-executor/types/messages.ts b/packages/chaingraph-executor/types/messages.ts
@@ -80,7 +80,12 @@ export interface ExecutionEventMessage {
   workerId: string // Which worker produced this
 }
 
-const topicsPrefix = process.env.KAFKA_TOPICS_PREFIX || ''
+let topicsPrefix = ''
+try {
+  topicsPrefix = process?.env?.KAFKA_TOPICS_PREFIX || ''
+} catch {
+  topicsPrefix = ''
+}
 
 /**
  * Kafka topic names
diff --git a/packages/chaingraph-nodes/CHANGELOG.md b/packages/chaingraph-nodes/CHANGELOG.md
diff --git a/packages/chaingraph-nodes/package.json b/packages/chaingraph-nodes/package.json
diff --git a/packages/chaingraph-trpc/CHANGELOG.md b/packages/chaingraph-trpc/CHANGELOG.md
diff --git a/packages/chaingraph-trpc/package.json b/packages/chaingraph-trpc/package.json
diff --git a/packages/chaingraph-types/CHANGELOG.md b/packages/chaingraph-types/CHANGELOG.md
diff --git a/packages/chaingraph-types/package.json b/packages/chaingraph-types/package.json
diff --git a/packages/typescript-config/CHANGELOG.md b/packages/typescript-config/CHANGELOG.md
diff --git a/packages/typescript-config/package.json b/packages/typescript-config/package.json

Original file line number	Diff line number	Diff line change
`@@ -32,6 +32,7 @@`
`32`	`32`	`"rude-parts-heal",`
`33`	`33`	`"sixty-humans-smile",`
`34`	`34`	`"tiny-things-report",`
`35`		`- "tired-days-sleep"`
	`35`	`+ "tired-days-sleep",`
	`36`	`+ "tough-rice-spend"`
`36`	`37`	`]`
`37`	`38`	`}`
Original file line number	Diff line number	Diff line change
`@@ -1,7 +1,7 @@`
`1`	`1`	`{`
`2`	`2`	`"name": "@badaitech/chaingraph-backend",`
`3`	`3`	`"type": "module",`
`4`		`- "version": "0.5.5-dev.18",`
	`4`	`+ "version": "0.5.5-dev.19",`
`5`	`5`	`"private": false,`
`6`	`6`	`"description": "Backend server for the Chaingraph project",`
`7`	`7`	`"license": "BUSL-1.1",`
Original file line number	Diff line number	Diff line change
`@@ -1,7 +1,7 @@`
`1`	`1`	`{`
`2`	`2`	`"name": "@badaitech/chaingraph-execution-api",`
`3`	`3`	`"type": "module",`
`4`		`- "version": "0.5.5-dev.18",`
	`4`	`+ "version": "0.5.5-dev.19",`
`5`	`5`	`"private": false,`
`6`	`6`	`"description": "Chaingraph tRPC Server - Scalable API server for Chaingraph execution management",`
`7`	`7`	`"license": "BUSL-1.1",`
Original file line number	Diff line number	Diff line change
`@@ -1,7 +1,7 @@`
`1`	`1`	`{`
`2`	`2`	`"name": "@badaitech/chaingraph-frontend",`
`3`	`3`	`"type": "module",`
`4`		`- "version": "0.5.5-dev.18",`
	`4`	`+ "version": "0.5.5-dev.19",`
`5`	`5`	`"private": false,`
`6`	`6`	`"description": "Frontend application for the Chaingraph project",`
`7`	`7`	`"license": "BUSL-1.1",`