feat: enhance Kafka configuration with dynamic topics prefix support

Author: yakud

apps/chaingraph-execution-api/.env.example

@@ -10,6 +10,7 @@ EXECUTION_MODE=distributed
 
 # Kafka (required for distributed mode)
 KAFKA_BROKERS=localhost:9092
+KAFKA_TOPICS_PREFIX=dev.
 
 # Logging
 # debug, info, warn, error

apps/chaingraph-execution-api/src/config.ts

@@ -27,7 +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',
+    topicsPrefix: process.env.KAFKA_TOPICS_PREFIX || '',
   },
 
   // Logging

apps/chaingraph-execution-worker/.env.example

@@ -19,6 +19,7 @@ DATABASE_URL_EXECUTIONS=postgres://postgres@0.0.0.0:5435/postgres?sslmode=disabl
 
 # Kafka (required for distributed mode)
 KAFKA_BROKERS=localhost:9092
+KAFKA_TOPICS_PREFIX=dev.
 
 # Health Monitoring
 # Port for health endpoint

apps/chaingraph-execution-worker/src/config.ts

@@ -32,6 +32,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 || '',
   },
 
   // Health and monitoring

apps/chaingraph-execution-worker/src/worker.ts

@@ -25,6 +25,7 @@ export async function startWorker(): Promise<void> {
     process.env.DATABASE_URL = config.databaseUrl
     process.env.DATABASE_URL_EXECUTIONS = config.databaseUrlExecutions
     process.env.KAFKA_BROKERS = config.kafka.brokers.join(',')
+    process.env.KAFKA_TOPICS_PREFIX = config.kafka.topicsPrefix
     process.env.WORKER_ID = workerId
 
     // Create services (EventBus, TaskQueue, ExecutionStore)

packages/chaingraph-executor/server/utils/config.ts

@@ -28,6 +28,7 @@ export const config = {
   kafka: {
     brokers: (process.env.KAFKA_BROKERS || 'localhost:9092,localhost:9093,localhost:9094').split(','),
     clientId: process.env.KAFKA_CLIENT_ID || 'chaingraph-executor',
+    topicsPrefix: process.env.KAFKA_TOPICS_PREFIX || '',
     groupId: {
       worker: process.env.KAFKA_GROUP_ID_WORKER || 'chaingraph-execution-workers',
       stream: process.env.KAFKA_GROUP_ID_STREAM || 'chaingraph-event-stream',

packages/chaingraph-executor/server/workers/ExecutionWorker.ts

@@ -373,9 +373,7 @@ export class ExecutionWorker {
         heartbeatInterval: 3000,
         // Ultra-low latency optimizations
         maxWaitTimeInMs: 1, // 1ms for ultra-low latency
-        fetchMinBytes: 1, // Don't wait for data accumulation
-        fetchBatchSize: 1024 * 1024, // 1MB batch size
-        maxBytesPerPartition: 1024 * 1024, // 1MB per partition
+        allowAutoTopicCreation: false,
       })
 
       await this.commandConsumer.connect()

packages/chaingraph-executor/types/messages.ts

@@ -12,6 +12,7 @@ import type {
   ExecutionOptions,
   IntegrationContext,
 } from '@badaitech/chaingraph-types'
+import process from 'node:process'
 
 export enum ExecutionCommandType {
   CREATE = 'CREATE',
@@ -79,13 +80,15 @@ export interface ExecutionEventMessage {
   workerId: string // Which worker produced this
 }
 
+const topicsPrefix = process.env.KAFKA_TOPICS_PREFIX || ''
+
 /**
  * Kafka topic names
  */
 export const KafkaTopics = {
-  COMMANDS: '2.chaingraph.execution.commands',
-  EVENTS: '2.chaingraph.execution.events',
-  TASKS: '2.chaingraph.execution.tasks',
+  COMMANDS: `${topicsPrefix}chaingraph.execution.commands`,
+  EVENTS: `${topicsPrefix}chaingraph.execution.events`,
+  TASKS: `${topicsPrefix}chaingraph.execution.tasks`,
 } as const
 
 export type KafkaTopicName = typeof KafkaTopics[keyof typeof KafkaTopics]