Merge pull request #300 from badaitech/feat/flow-initialize-enhance

yakud · web-flow · commit f8318a8eee26 · 2025-09-19T16:30:12.000+03:00
feat: enhance Kafka configuration with topics prefix and improve error handling in task publishing
diff --git a/.changeset/pre.json b/.changeset/pre.json
@@ -19,6 +19,7 @@
     "beige-geese-show",
     "chubby-numbers-enjoy",
     "cold-news-fold",
+    "easy-candies-smile",
     "full-readers-itch",
     "hungry-words-hang",
     "loud-gifts-tell",
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.14
+
+### Patch Changes
+
+- feat: enhance Kafka configuration with topics prefix and improve error handling in task publishing
+
 ## 0.5.5-dev.13
 
 ### 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.13",
+  "version": "0.5.5-dev.14",
   "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.14
+
+### Patch Changes
+
+- feat: enhance Kafka configuration with topics prefix and improve error handling in task publishing
+- Updated dependencies
+  - @badaitech/chaingraph-executor@0.5.5-dev.14
+  - @badaitech/chaingraph-types@0.5.5-dev.14
+  - @badaitech/chaingraph-trpc@0.5.5-dev.14
+
 ## 0.5.5-dev.13
 
 ### 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.13",
+  "version": "0.5.5-dev.14",
   "private": false,
   "description": "Chaingraph tRPC Server - Scalable API server for Chaingraph execution management",
   "license": "BUSL-1.1",
diff --git a/apps/chaingraph-execution-api/src/config.ts b/apps/chaingraph-execution-api/src/config.ts
@@ -27,6 +27,7 @@ export const config = {
   // Kafka configuration (needed for distributed mode)
   kafka: {
     brokers: (process.env.KAFKA_BROKERS || 'localhost:9092').split(','),
+    topicsPrefix: process.env.KAFKA_TOPICS_PREFIX || 'chaingraph',
   },
 
   // Logging
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.14
+
+### Patch Changes
+
+- feat: enhance Kafka configuration with topics prefix and improve error handling in task publishing
+- Updated dependencies
+  - @badaitech/chaingraph-executor@0.5.5-dev.14
+  - @badaitech/chaingraph-nodes@0.5.5-dev.14
+  - @badaitech/chaingraph-types@0.5.5-dev.14
+  - @badaitech/chaingraph-trpc@0.5.5-dev.14
+
 ## 0.5.5-dev.13
 
 ### 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.13",
+  "version": "0.5.5-dev.14",
   "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-frontend/CHANGELOG.md b/apps/chaingraph-frontend/CHANGELOG.md
@@ -1,5 +1,17 @@
 # @badaitech/chaingraph-frontend
 
+## 0.5.5-dev.14
+
+### Patch Changes
+
+- feat: enhance Kafka configuration with topics prefix and improve error handling in task publishing
+- Updated dependencies
+  - @badaitech/chaingraph-executor@0.5.5-dev.14
+  - @badaitech/chaingraph-nodes@0.5.5-dev.14
+  - @badaitech/chaingraph-types@0.5.5-dev.14
+  - @badaitech/chaingraph-trpc@0.5.5-dev.14
+  - @badaitech/badai-api@0.5.5-dev.14
+
 ## 0.5.5-dev.13
 
 ### 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.13",
+  "version": "0.5.5-dev.14",
   "private": false,
   "description": "Frontend application for the Chaingraph project",
   "license": "BUSL-1.1",
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.14
+
+### Patch Changes
+
+- feat: enhance Kafka configuration with topics prefix and improve error handling in task publishing
+
 ## 0.5.5-dev.13
 
 ### 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.13",
+  "version": "0.5.5-dev.14",
   "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.13",
+  "version": "0.5.5-dev.14",
   "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
@@ -257,6 +257,7 @@ export class KafkaEventBus implements IEventBus {
               // Parse message
               // Use safe parsing to handle potential issues
               const parsedMessage = safeSuperJSONParse<any>(messageValue)
+              // TODO: Add schema validation for parsedMessage (zod schema or similar)
 
               // Filter by executionId
               if (parsedMessage.executionId !== executionId)
diff --git a/packages/chaingraph-executor/server/implementations/distributed/KafkaTaskQueue.ts b/packages/chaingraph-executor/server/implementations/distributed/KafkaTaskQueue.ts
@@ -31,7 +31,7 @@ export class KafkaTaskQueue implements ITaskQueue {
       this.producer = await getTaskProducer()
     }
 
-    await this.producer.send({
+    const records = await this.producer.send({
       topic: KafkaTopics.TASKS,
       messages: [{
         key: task.executionId,
@@ -41,6 +41,25 @@ export class KafkaTaskQueue implements ITaskQueue {
       acks: -1,
       timeout: 30000,
     })
+
+    for (const record of records) {
+      if (record.errorCode && record.errorCode !== 0) {
+        logger.error({
+          topicName: record.topicName,
+          partition: record.partition,
+          errorCode: record.errorCode,
+        }, 'Failed to publish task to Kafka')
+        throw new Error(`Failed to publish task to Kafka, error code: ${record.errorCode}`)
+      }
+
+      logger.debug({
+        topicName: record.topicName,
+        partition: record.partition,
+        baseOffset: record.baseOffset,
+        logAppendTime: record.logAppendTime,
+        logStartOffset: record.logStartOffset,
+      }, 'Task published to Kafka successfully')
+    }
   }
 
   consumeTasks = async (handler: (task: ExecutionTask) => Promise<void>): Promise<void> => {
@@ -75,6 +94,8 @@ export class KafkaTaskQueue implements ITaskQueue {
         try {
           const task = safeSuperJSONParse<ExecutionTask>(message.value.toString())
 
+          // TODO: Add schema validation for task (zod schema or similar)
+
           if (!task) {
             logger.error({
               messageValue: message.value.toString().substring(0, 200),
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: false,
+      allowAutoTopicCreation: true,
       idempotent: true,
       createPartitioner: Partitioners.DefaultPartitioner,
       maxInFlightRequests: 5,
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: false,
+      allowAutoTopicCreation: true,
       idempotent: true,
       createPartitioner: Partitioners.DefaultPartitioner,
       maxInFlightRequests: 5,
diff --git a/packages/chaingraph-executor/server/kafka/producers/task-producer.ts b/packages/chaingraph-executor/server/kafka/producers/task-producer.ts
@@ -24,7 +24,7 @@ export async function getTaskProducer(): Promise<Producer> {
   if (!producer) {
     const kafka = getKafkaClient()
     producer = kafka.producer({
-      allowAutoTopicCreation: false,
+      allowAutoTopicCreation: true,
       idempotent: true,
       createPartitioner: Partitioners.DefaultPartitioner,
       maxInFlightRequests: 5,
diff --git a/packages/chaingraph-executor/server/trpc/router.ts b/packages/chaingraph-executor/server/trpc/router.ts
@@ -138,13 +138,15 @@ export const executionRouter = router({
 
       const flowMetadata = await flowStore.getFlowMetadata(input.flowId)
       if (!flowMetadata) {
+        logger.warn({ flowId: input.flowId }, 'Flow not found when creating execution')
         throw new TRPCError({
           code: 'NOT_FOUND',
           message: `Flow with id ${input.flowId} not found`,
         })
       }
 
       if (!flowMetadata.id) {
+        logger.error({ flowMetadata }, 'Flow metadata is missing ID')
         throw new TRPCError({
           code: 'INTERNAL_SERVER_ERROR',
           message: `Flow metadata is missing ID`,
diff --git a/packages/chaingraph-nodes/CHANGELOG.md b/packages/chaingraph-nodes/CHANGELOG.md
@@ -1,5 +1,13 @@
 # @badaitech/chaingraph-nodes
 
+## 0.5.5-dev.14
+
+### Patch Changes
+
+- feat: enhance Kafka configuration with topics prefix and improve error handling in task publishing
+- Updated dependencies
+  - @badaitech/chaingraph-types@0.5.5-dev.14
+
 ## 0.5.5-dev.13
 
 ### Patch Changes
diff --git a/packages/chaingraph-nodes/package.json b/packages/chaingraph-nodes/package.json
@@ -1,7 +1,7 @@
 {
   "name": "@badaitech/chaingraph-nodes",
   "type": "module",
-  "version": "0.5.5-dev.13",
+  "version": "0.5.5-dev.14",
   "private": false,
   "description": "Node components for the Chaingraph project",
   "license": "BUSL-1.1",
diff --git a/packages/chaingraph-trpc/CHANGELOG.md b/packages/chaingraph-trpc/CHANGELOG.md
@@ -1,5 +1,11 @@
 # @badaitech/chaingraph-trpc
 
+## 0.5.5-dev.14
+
+### Patch Changes
+
+- feat: enhance Kafka configuration with topics prefix and improve error handling in task publishing
+
 ## 0.5.5-dev.13
 
 ### Patch Changes
diff --git a/packages/chaingraph-trpc/package.json b/packages/chaingraph-trpc/package.json
@@ -1,7 +1,7 @@
 {
   "name": "@badaitech/chaingraph-trpc",
   "type": "module",
-  "version": "0.5.5-dev.13",
+  "version": "0.5.5-dev.14",
   "private": false,
   "description": "tRPC client and server for the Chaingraph V2",
   "license": "BUSL-1.1",
diff --git a/packages/chaingraph-types/CHANGELOG.md b/packages/chaingraph-types/CHANGELOG.md
@@ -1,5 +1,11 @@
 # @badaitech/chaingraph-types
 
+## 0.5.5-dev.14
+
+### Patch Changes
+
+- feat: enhance Kafka configuration with topics prefix and improve error handling in task publishing
+
 ## 0.5.5-dev.13
 
 ### Patch Changes
diff --git a/packages/chaingraph-types/package.json b/packages/chaingraph-types/package.json
@@ -1,7 +1,7 @@
 {
   "name": "@badaitech/chaingraph-types",
   "type": "module",
-  "version": "0.5.5-dev.13",
+  "version": "0.5.5-dev.14",
   "private": false,
   "description": "Core type definitions for the Chaingraph project",
   "license": "BUSL-1.1",
diff --git a/packages/typescript-config/CHANGELOG.md b/packages/typescript-config/CHANGELOG.md
@@ -1,5 +1,11 @@
 # @badaitech/typescript-config
 
+## 0.5.5-dev.14
+
+### Patch Changes
+
+- feat: enhance Kafka configuration with topics prefix and improve error handling in task publishing
+
 ## 0.5.5-dev.13
 
 ### Patch Changes
diff --git a/packages/typescript-config/package.json b/packages/typescript-config/package.json
@@ -1,6 +1,6 @@
 {
   "name": "@badaitech/typescript-config",
-  "version": "0.5.5-dev.13",
+  "version": "0.5.5-dev.14",
   "private": true,
   "license": "MIT",
   "files": [

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.13",`
	`4`	`+ "version": "0.5.5-dev.14",`
`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.13",`
	`4`	`+ "version": "0.5.5-dev.14",`
`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.13",`
	`4`	`+ "version": "0.5.5-dev.14",`
`5`	`5`	`"private": false,`
`6`	`6`	`"description": "Frontend application for the Chaingraph project",`
`7`	`7`	`"license": "BUSL-1.1",`