Merge pull request #2703 from pikasTech/fix/workbench-worker-kafka-producer-config

fix: 为 Workbench worker 注入独立 Kafka producer 配置

deploy/deploy.yaml

@@ -1161,6 +1161,9 @@ lanes:
           WORKBENCH_ACTIVITY_RETRY_INITIAL_INTERVAL_MS: "1000"
           WORKBENCH_ACTIVITY_RETRY_MAXIMUM_INTERVAL_MS: "30000"
           WORKBENCH_ACTIVITY_RETRY_MAXIMUM_ATTEMPTS: "5"
+          HWLAB_KAFKA_BOOTSTRAP_SERVERS: platform-infra-kafka-kafka-bootstrap.platform-infra.svc.cluster.local:9092
+          HWLAB_KAFKA_EVENT_TOPIC: hwlab.event.v1
+          HWLAB_KAFKA_CLIENT_ID: hwlab-v03-workbench-worker
           OTEL_SERVICE_NAME: hwlab-workbench-worker
   production:
     name: Production

internal/workbench/kafka-event-publisher.test.ts

@@ -60,16 +60,8 @@ test("L0 publisher writes semantic activity to hwlab.event.v1", async () => {
 
 function kafkaEnvironment(): Record<string, string> {
   return {
-    HWLAB_KAFKA_DIRECT_PUBLISH_ENABLED: "false",
-    HWLAB_WORKBENCH_LIVE_KAFKA_SSE_ENABLED: "true",
-    HWLAB_WORKBENCH_KAFKA_REFRESH_REPLAY_ENABLED: "false",
-    HWLAB_KAFKA_TRANSACTIONAL_PROJECTOR_ENABLED: "false",
-    HWLAB_KAFKA_PROJECTION_OUTBOX_RELAY_ENABLED: "false",
-    HWLAB_WORKBENCH_PROJECTION_REALTIME_ENABLED: "false",
     HWLAB_KAFKA_BOOTSTRAP_SERVERS: "127.0.0.1:9092",
-    HWLAB_KAFKA_AGENTRUN_EVENT_TOPIC: "agentrun.event.v1",
     HWLAB_KAFKA_EVENT_TOPIC: "hwlab.event.v1",
     HWLAB_KAFKA_CLIENT_ID: "hwlab-workbench-l0",
-    HWLAB_KAFKA_HWLAB_EVENT_GROUP_ID: "hwlab-workbench-l0-live",
   };
 }

internal/workbench/kafka-event-publisher.ts

@@ -3,7 +3,6 @@ import { randomUUID } from "node:crypto";
 import {
   buildHwlabKafkaProducerMessage,
   defaultKafkaFactory,
-  kafkaEventBridgeConfig,
   projectAgentRunKafkaEventToHwlabEvent,
 } from "../cloud/kafka-event-bridge.ts";
 import type { WorkbenchEventPublisher } from "./contracts.ts";
@@ -13,8 +12,7 @@ export function createWorkbenchKafkaEventPublisher(options: {
   kafkaFactory?: typeof defaultKafkaFactory;
   now?: () => string;
 }): WorkbenchEventPublisher {
-  const config = kafkaEventBridgeConfig(options.env);
-  if (!config) throw codedError("workbench_kafka_publisher_unconfigured", "Workbench event publisher requires Kafka realtime configuration.");
+  const config = workbenchKafkaPublisherConfig(options.env);
   const kafka = (options.kafkaFactory ?? defaultKafkaFactory)(config);
   const producer = kafka.producer({ allowAutoTopicCreation: false });
   let ready: Promise<unknown> | null = null;
@@ -63,6 +61,26 @@ export function createWorkbenchKafkaEventPublisher(options: {
   };
 }
 
+function workbenchKafkaPublisherConfig(env: Record<string, string | undefined>) {
+  const brokers = csv(env.HWLAB_KAFKA_BOOTSTRAP_SERVERS);
+  const hwlabTopic = text(env.HWLAB_KAFKA_EVENT_TOPIC);
+  const clientId = text(env.HWLAB_KAFKA_CLIENT_ID);
+  const missing = [
+    brokers.length === 0 ? "HWLAB_KAFKA_BOOTSTRAP_SERVERS" : null,
+    hwlabTopic ? null : "HWLAB_KAFKA_EVENT_TOPIC",
+    clientId ? null : "HWLAB_KAFKA_CLIENT_ID",
+  ].filter(Boolean);
+  if (missing.length > 0) throw codedError("workbench_kafka_publisher_unconfigured", `Workbench event publisher configuration is incomplete: ${missing.join(", ")}`);
+  return {
+    brokers,
+    hwlabTopic,
+    clientId,
+    dnsServers: csv(env.HWLAB_KAFKA_DNS_SERVERS),
+    dnsSearchDomains: csv(env.HWLAB_KAFKA_DNS_SEARCH_DOMAINS),
+  };
+}
+
+function csv(value: unknown): string[] { return text(value).split(",").map((item) => item.trim()).filter(Boolean); }
 function text(value: unknown): string { return String(value ?? "").trim(); }
 function timestamp(value: unknown): string | null { const result = text(value); return result && Number.isFinite(Date.parse(result)) ? new Date(Date.parse(result)).toISOString() : null; }
 function codedError(code: string, message: string) { return Object.assign(new Error(message), { code }); }