chore(cdp): remove the legacy Kafka batch hogflow consumer and routing

nodejs/src/capabilities.ts

@@ -24,7 +24,6 @@ export const CAPABILITIES_CDP: PluginServerCapabilities = {
 /** CDP + Workflows - full CDP with HogFlow workflow automation */
 export const CAPABILITIES_CDP_WORKFLOWS: PluginServerCapabilities = {
     ...CAPABILITIES_CDP,
-    cdpBatchHogFlow: true,
     cdpCyclotronWorkerBatchResolve: true,
     cdpCyclotronWorkerHogFlow: true,
     cdpCyclotronWorkerEmail: true,
@@ -175,10 +174,6 @@ export function getPluginServerCapabilities(
             return {
                 errorTrackingIngestion: true,
             }
-        case PluginServerMode.cdp_batch_hogflow_requests:
-            return {
-                cdpBatchHogFlow: true,
-            }
         case PluginServerMode.cdp_cyclotron_worker_batch_resolve:
             return {
                 cdpCyclotronWorkerBatchResolve: true,

nodejs/src/cdp/cdp-api.test.ts

@@ -1,5 +1,4 @@
 import { createMockJobQueue } from '../../tests/helpers/mocks/job-queue.mock'
-import { mockProducer } from '../../tests/helpers/mocks/producer.mock'
 import { mockFetch } from '../../tests/helpers/mocks/request.mock'
 
 import { Server } from 'http'
@@ -774,14 +773,8 @@ describe('CDP API', () => {
 
     describe('batch hogflow invocations', () => {
         let batchHogFlow: HogFlow
-        let produceSpy: jest.SpyInstance
 
         beforeEach(async () => {
-            // The batch hogflow route now goes through the outputs registry, which in
-            // tests routes every CDP producer slot at the shared `mockProducer`. Spying
-            // on its `produce` intercepts the produced message without reconstructing
-            // the api.
-            produceSpy = jest.spyOn(mockProducer, 'produce')
             batchHogFlow = await insertHogFlow({
                 id: new UUIDT().toString(),
                 name: 'test batch hog flow',
@@ -806,10 +799,6 @@ describe('CDP API', () => {
             })
         })
 
-        afterEach(() => {
-            produceSpy.mockRestore()
-        })
-
         it('errors if missing team', async () => {
             const nonExistentTeamId = new UUIDT().toString()
             const res = await supertest(app)
@@ -855,73 +844,12 @@ describe('CDP API', () => {
             expect(res.body.error).toEqual('Only batch Workflows are supported for batch jobs')
         })
 
-        it('queues batch job request to kafka', async () => {
-            produceSpy.mockResolvedValue(undefined)
-
-            const res = await supertest(app)
-                .post(`/api/projects/${batchHogFlow.team_id}/hog_flows/${batchHogFlow.id}/batch_invocations/job-123`)
-                .send({
-                    filters: {
-                        filter_test_accounts: true,
-                    },
-                })
-
-            expect(res.status).toEqual(200)
-            expect(res.body).toEqual({ status: 'queued' })
-            expect(produceSpy).toHaveBeenCalledWith({
-                topic: 'cdp_batch_hogflow_requests_test',
-                value: Buffer.from(
-                    JSON.stringify({
-                        teamId: batchHogFlow.team_id,
-                        hogFlowId: batchHogFlow.id,
-                        parentRunId: 'job-123',
-                        filters: {
-                            properties: (batchHogFlow as any).trigger.filters.properties,
-                            filter_test_accounts: true,
-                        },
-                    })
-                ),
-                key: `${batchHogFlow.team_id}_${batchHogFlow.id}`,
-            })
-        })
-
-        it('queues batch job with filters from hog flow config when not provided', async () => {
-            produceSpy.mockResolvedValue(undefined)
-
-            const res = await supertest(app)
-                .post(`/api/projects/${batchHogFlow.team_id}/hog_flows/${batchHogFlow.id}/batch_invocations/job-456`)
-                .send({})
-
-            expect(res.status).toEqual(200)
-            expect(res.body).toEqual({ status: 'queued' })
-            expect(produceSpy).toHaveBeenCalledWith({
-                topic: 'cdp_batch_hogflow_requests_test',
-                value: Buffer.from(
-                    JSON.stringify({
-                        teamId: batchHogFlow.team_id,
-                        hogFlowId: batchHogFlow.id,
-                        parentRunId: 'job-456',
-                        filters: {
-                            properties: (batchHogFlow as any).trigger.filters.properties,
-                            filter_test_accounts: false,
-                        },
-                    })
-                ),
-                key: `${batchHogFlow.team_id}_${batchHogFlow.id}`,
-            })
-        })
-
-        it('routes to the cyclotron resolver when CDP_BATCH_RESOLVER_ROUTING matches the team', async () => {
-            // Stub a producer in place of the real CyclotronV2Manager; assert
-            // it gets the right createJob payload (queue name, parentRunId,
-            // serialized state) without standing up a real cyclotron pool.
+        it('queues batch job to the cyclotron resolver', async () => {
             const createJobMock = jest.fn().mockResolvedValue('resolver-job-id')
             api['batchResolverProducer'] = {
                 createJob: createJobMock,
                 disconnect: jest.fn().mockResolvedValue(undefined),
             }
-            const originalMatcher = api['batchResolverRoutingMatcher']
-            api['batchResolverRoutingMatcher'] = () => true
 
             try {
                 const res = await supertest(app)
@@ -937,9 +865,6 @@ describe('CDP API', () => {
                 expect(res.status).toEqual(200)
                 expect(res.body).toEqual({ status: 'queued' })
 
-                // Kafka path stays untouched
-                expect(produceSpy).not.toHaveBeenCalled()
-
                 expect(createJobMock).toHaveBeenCalledTimes(1)
                 const arg = createJobMock.mock.calls[0][0]
                 expect(arg).toMatchObject({
@@ -954,14 +879,17 @@ describe('CDP API', () => {
                     batchJobId: 'job-789',
                     teamId: batchHogFlow.team_id,
                     hogFlowId: batchHogFlow.id,
+                    filters: {
+                        properties: (batchHogFlow as any).trigger.filters.properties,
+                        filter_test_accounts: true,
+                    },
                     maxAudienceSize: 1234,
                     variables: { foo: 'bar' },
                     cursor: null,
                     totalEnqueued: 0,
                     pagesProcessed: 0,
                 })
             } finally {
-                api['batchResolverRoutingMatcher'] = originalMatcher
                 api['batchResolverProducer'] = null
             }
         })

nodejs/src/cdp/cdp-api.ts

@@ -2,7 +2,6 @@ import { DateTime } from 'luxon'
 import express from 'ultimate-express'
 
 import { ModifiedRequest } from '~/common/api/router'
-import { buildIntegerMatcherWithPercentage } from '~/common/config/config'
 import { logger } from '~/common/utils/logger'
 import { UUID, UUIDT, delay } from '~/common/utils/utils'
 import { PluginEvent } from '~/plugin-scaffold'
@@ -13,19 +12,17 @@ import {
     HealthCheckResultOk,
     PluginServerService,
     PluginsServerConfig,
-    ValueMatcher,
 } from '../types'
 import { getAsyncFunctionHandler, getRegisteredAsyncFunctionNames } from './async-function-registry'
 import './async-functions'
-import { CdpOutputs, createCdpCoreServices } from './cdp-services'
+import { createCdpCoreServices } from './cdp-services'
 import { CdpConsumerBaseDeps } from './consumers/cdp-base.consumer'
 import {
     CdpSourceWebhooksConsumer,
     HogFunctionWebhookResult,
     SourceWebhookError,
 } from './consumers/cdp-source-webhooks.consumer'
 import { HogTransformerService, createHogTransformerService } from './hog-transformations/hog-transformer.service'
-import { BATCH_HOGFLOW_REQUESTS_OUTPUT } from './outputs/outputs'
 import { RerunJobManager } from './rerun/rerun-job.manager'
 import { RerunRequest } from './rerun/rerun-job.types'
 import { BatchExportHogFunctionService, NotFoundError, ParseError } from './services/batch-export-hog-function.service'
@@ -103,11 +100,9 @@ export class CdpApi {
     private hogflowQueue: JobQueue
     private emailTrackingService: EmailTrackingService
     private recipientTokensService: RecipientTokensService
-    private outputs: CdpOutputs
     private batchExportHogFunctionService: BatchExportHogFunctionService
     private groupsManager: GroupsManagerService
     private batchResolverProducer: CyclotronV2JobProducer | null
-    private batchResolverRoutingMatcher: ValueMatcher<number>
 
     constructor(
         private config: PluginsServerConfig,
@@ -126,7 +121,6 @@ export class CdpApi {
         this.segmentDestinationExecutorService = services.segmentDestinationExecutorService
         this.hogWatcher = services.hogWatcher
         this.invocationResultsService = services.invocationResultsService
-        this.outputs = services.outputs
 
         // API-only services. The hog-transformer's monitoring service reuses the same
         // resolved outputs registry as the core CDP services — no separate construction.
@@ -157,7 +151,6 @@ export class CdpApi {
             this.invocationResultsService
         )
         this.batchResolverProducer = batchResolverProducer
-        this.batchResolverRoutingMatcher = buildIntegerMatcherWithPercentage(config.CDP_BATCH_RESOLVER_ROUTING)
     }
 
     public get service(): PluginServerService {
@@ -781,49 +774,34 @@ export class CdpApi {
             const maxAudienceSize =
                 typeof req.body.max_audience_size === 'number' ? req.body.max_audience_size : undefined
 
-            const batchHogFlowRequest = {
+            if (!this.batchResolverProducer) {
+                throw new Error('Batch resolver producer is not configured (missing CYCLOTRON_NODE_DATABASE_URL)')
+            }
+
+            const initialState: BatchResolverState = {
+                batchJobId: parent_run_id,
                 teamId: team.id,
                 hogFlowId: hogFlow.id,
-                parentRunId: parent_run_id,
                 filters: {
                     properties: hogFlow.trigger.filters.properties || [],
                     filter_test_accounts: req.body.filters?.filter_test_accounts || false,
                 },
-                maxAudienceSize,
-            }
-
-            if (this.batchResolverRoutingMatcher(team.id)) {
-                if (!this.batchResolverProducer) {
-                    throw new Error('CDP_BATCH_RESOLVER_ROUTING matched team but no producer is configured')
-                }
-                const initialState: BatchResolverState = {
-                    batchJobId: parent_run_id,
-                    teamId: team.id,
-                    hogFlowId: hogFlow.id,
-                    filters: batchHogFlowRequest.filters,
-                    variables: req.body.variables ?? {},
-                    groupTypeIndex:
-                        typeof req.body.group_type_index === 'number' ? req.body.group_type_index : undefined,
-                    maxAudienceSize: maxAudienceSize ?? this.config.CDP_BATCH_WORKFLOW_MAX_AUDIENCE_SIZE,
-                    cursor: null,
-                    totalEnqueued: 0,
-                    pagesProcessed: 0,
-                    attempts: 0,
-                    startedAt: new Date().toISOString(),
-                }
-                await this.batchResolverProducer.createJob({
-                    teamId: team.id,
-                    queueName: HOGFLOW_BATCH_RESOLVE_QUEUE,
-                    parentRunId: parent_run_id,
-                    functionId: hogFlow.id,
-                    state: serializeResolverState(initialState),
-                })
-            } else {
-                await this.outputs.produce(BATCH_HOGFLOW_REQUESTS_OUTPUT, {
-                    value: Buffer.from(JSON.stringify(batchHogFlowRequest)),
-                    key: `${team.id}_${hogFlow.id}`,
-                })
-            }
+                variables: req.body.variables ?? {},
+                groupTypeIndex: typeof req.body.group_type_index === 'number' ? req.body.group_type_index : undefined,
+                maxAudienceSize: maxAudienceSize ?? this.config.CDP_BATCH_WORKFLOW_MAX_AUDIENCE_SIZE,
+                cursor: null,
+                totalEnqueued: 0,
+                pagesProcessed: 0,
+                attempts: 0,
+                startedAt: new Date().toISOString(),
+            }
+            await this.batchResolverProducer.createJob({
+                teamId: team.id,
+                queueName: HOGFLOW_BATCH_RESOLVE_QUEUE,
+                parentRunId: parent_run_id,
+                functionId: hogFlow.id,
+                state: serializeResolverState(initialState),
+            })
 
             res.json({ status: 'queued' })
         } catch (e) {

nodejs/src/cdp/cdp-services.ts

@@ -12,7 +12,6 @@ import type { CommonConfig } from '../common/config'
 import { InternalCaptureService } from '../common/services/internal-capture'
 import type { CdpConfig } from './config'
 import {
-    BatchHogflowRequestsOutput,
     PrecalculatedPersonPropertiesOutput,
     PrefilteredEventsOutput,
     WarehouseSourceWebhooksOutput,
@@ -51,7 +50,6 @@ export type CdpOutput =
     | HogInvocationResultsOutput
     | PrefilteredEventsOutput
     | PrecalculatedPersonPropertiesOutput
-    | BatchHogflowRequestsOutput
     | WarehouseSourceWebhooksOutput
 
 export type CdpOutputs = IngestionOutputs<CdpOutput>
@@ -158,8 +156,6 @@ export type CdpCoreServicesConfig = Pick<
         | 'CDP_PREFILTERED_EVENTS_PRODUCER'
         | 'CDP_PRECALCULATED_PERSON_PROPERTIES_TOPIC'
         | 'CDP_PRECALCULATED_PERSON_PROPERTIES_PRODUCER'
-        | 'CDP_BATCH_HOGFLOW_REQUESTS_TOPIC'
-        | 'CDP_BATCH_HOGFLOW_REQUESTS_PRODUCER'
         | 'CDP_WAREHOUSE_SOURCE_WEBHOOKS_TOPIC'
         | 'CDP_WAREHOUSE_SOURCE_WEBHOOKS_PRODUCER'
     >

nodejs/src/cdp/config.ts

@@ -1,6 +1,5 @@
 import {
     KAFKA_APP_METRICS_2,
-    KAFKA_CDP_BATCH_HOGFLOW_REQUESTS,
     KAFKA_CDP_CLICKHOUSE_PRECALCULATED_PERSON_PROPERTIES,
     KAFKA_CDP_CLICKHOUSE_PREFILTERED_EVENTS,
     KAFKA_EVENTS_JSON,
@@ -113,8 +112,6 @@ export type CdpConfig = ClickhouseConfig & {
     CDP_PREFILTERED_EVENTS_PRODUCER: CdpProducerName
     CDP_PRECALCULATED_PERSON_PROPERTIES_TOPIC: string
     CDP_PRECALCULATED_PERSON_PROPERTIES_PRODUCER: CdpProducerName
-    CDP_BATCH_HOGFLOW_REQUESTS_TOPIC: string
-    CDP_BATCH_HOGFLOW_REQUESTS_PRODUCER: CdpProducerName
     CDP_WAREHOUSE_SOURCE_WEBHOOKS_TOPIC: string
     CDP_WAREHOUSE_SOURCE_WEBHOOKS_PRODUCER: CdpProducerName
 
@@ -133,18 +130,8 @@ export type CdpConfig = ClickhouseConfig & {
     // Destination migration diffing
     DESTINATION_MIGRATION_DIFFING_ENABLED: boolean
 
-    CDP_BATCH_WORKFLOW_PRODUCER_BATCH_SIZE: number
     CDP_BATCH_WORKFLOW_MAX_AUDIENCE_SIZE: number
 
-    // Per-team routing for postHogFlowBatchInvocation: teams matched here
-    // dispatch to a cyclotron resolver job (queue=hogflow_batch_resolve)
-    // instead of producing to the cdp_batch_hogflow_requests Kafka topic.
-    // Same string format as CDP_EMAIL_QUEUE_ROUTING — '' / '*' / '1,2' /
-    // '*:0.1' for percentage. The Kafka consumer path stays alive in
-    // parallel until this defaults to '*' everywhere and the legacy path
-    // is removed.
-    CDP_BATCH_RESOLVER_ROUTING: string
-
     // Cyclotron Node (node postgres job queue)
     CYCLOTRON_NODE_MAX_CONNECTIONS: number
     CYCLOTRON_NODE_IDLE_TIMEOUT_MS: number
@@ -250,8 +237,6 @@ export function getDefaultCdpConfig(): CdpConfig {
         CDP_PREFILTERED_EVENTS_PRODUCER: WARPSTREAM_CALCULATED_EVENTS_PRODUCER,
         CDP_PRECALCULATED_PERSON_PROPERTIES_TOPIC: KAFKA_CDP_CLICKHOUSE_PRECALCULATED_PERSON_PROPERTIES,
         CDP_PRECALCULATED_PERSON_PROPERTIES_PRODUCER: WARPSTREAM_CALCULATED_EVENTS_PRODUCER,
-        CDP_BATCH_HOGFLOW_REQUESTS_TOPIC: KAFKA_CDP_BATCH_HOGFLOW_REQUESTS,
-        CDP_BATCH_HOGFLOW_REQUESTS_PRODUCER: WARPSTREAM_CYCLOTRON_PRODUCER,
         CDP_WAREHOUSE_SOURCE_WEBHOOKS_TOPIC: KAFKA_WAREHOUSE_SOURCE_WEBHOOKS,
         CDP_WAREHOUSE_SOURCE_WEBHOOKS_PRODUCER: WAREHOUSE_PRODUCER,
 
@@ -277,9 +262,7 @@ export function getDefaultCdpConfig(): CdpConfig {
         // Destination migration diffing
         DESTINATION_MIGRATION_DIFFING_ENABLED: false,
 
-        CDP_BATCH_WORKFLOW_PRODUCER_BATCH_SIZE: 1,
         CDP_BATCH_WORKFLOW_MAX_AUDIENCE_SIZE: 5000,
-        CDP_BATCH_RESOLVER_ROUTING: '',
 
         // Cyclotron Node
         CYCLOTRON_NODE_MAX_CONNECTIONS: 10,