Repository navigation
chore(cohorts): remove the cdp cohort membership consumer - #79808
gustavohstrassburger wants to merge 5 commits into
Conversation
|
Merging to
After your PR is submitted to the merge queue, this comment will be automatically updated with its status. If the PR fails, failure details will also be posted here |
|
@greptile review |
|
Reviews (1): Last reviewed commit: "chore(cohorts): remove the cdp cohort me..." | Re-trigger Greptile |
|
|
5dc6caf to
b42e2a6
Compare
The warpstream-calculated-events cluster is being decommissioned. The cdp-cohort-membership consumer read cohort_membership_changed from it, and that topic lost its producer when the legacy precalculated realtime-cohort path was deleted, so the consumer has had nothing to read. Removes the consumer and its test, the topic constant and its routing entry, and a stale routing-overrides example that named the topic. PluginServerMode.cdp_cohort_membership stays as a stub returning no capabilities, following the existing cdp_precalculated_filters precedent: an unknown mode throws at boot, so the member cannot go until charts stops setting PLUGIN_SERVER_MODE for that deployment. The ClickHouse SQL definitions for the WarpStream Kafka engine tables are left in place; the tables themselves are dropped by a separate migration-only PR. Generated-By: PostHog Code Task-Id: f7c0c71a-192b-485b-9cef-879e131761bc
ad47c35 to
4edeff3
Compare
👀 Auto-assigned reviewersThese soft owners were skipped because they only have minor changes here. Nothing blocks merge, so self-assign if you'd like a look:
Soft owners come from each directory's |
|
Reviews (2): Last reviewed commit: "chore(cohorts): remove the cdp cohort me..." | Re-trigger Greptile |
There was a problem hiding this comment.
Pull request overview
Removes the deprecated cdp-cohort-membership Kafka consumer and associated wiring now that the cohort_membership_changed topic no longer has a producer and the warpstream-calculated-events cluster is being decommissioned. This cleanup eliminates an idle deployment component and prunes now-stale topic/routing configuration.
Changes:
- Deleted the
cdp-cohort-membershipconsumer implementation and its Jest test, plus related CDP test fixtures. - Removed the
cohort_membership_changedKafka topic constant and its Python/Node routing/import references; updated the example routing override comment. - Kept
PluginServerMode.cdp_cohort_membershipas a no-op stub capability mode to avoid crash-looping until charts stop setting it; updated monitoring skill docs to reflect remaining WarpStream topics.
Reviewed changes
Copilot reviewed 13 out of 13 changed files in this pull request and generated 1 comment.
Show a summary per file
| File | Description |
|---|---|
| posthog/settings/kafka.py | Updates the KAFKA_TOPIC_ROUTING_OVERRIDES example away from the removed topic. |
| posthog/kafka_client/topics.py | Removes KAFKA_COHORT_MEMBERSHIP_CHANGED topic constant. |
| posthog/kafka_client/routing.py | Removes routing entry/import for the removed topic. |
| nodejs/tests/helpers/kafka.ts | Drops the removed topic from test topic setup lists. |
| nodejs/src/types.ts | Removes cdpCohortMembership from PluginServerCapabilities. |
| nodejs/src/server.ts | Removes import and startup wiring for the deleted cohort membership consumer. |
| nodejs/src/common/config/kafka-topics.ts | Removes the Node topic constant for cohort_membership_changed. |
| nodejs/src/common/config.ts | Leaves PluginServerMode.cdp_cohort_membership enum member with TODO for later removal. |
| nodejs/src/cdp/consumers/cdp-cohort-membership.consumer.ts | Deletes the consumer implementation. |
| nodejs/src/cdp/consumers/cdp-cohort-membership.consumer.test.ts | Deletes the consumer end-to-end test suite. |
| nodejs/src/cdp/_tests/fixtures.ts | Removes cohort-membership-specific test helpers/types. |
| nodejs/src/capabilities.ts | Removes the realtime cohorts capability group and makes cdp_cohort_membership return no capabilities. |
| .agents/skills/monitoring-ingestion-pipeline/SKILL.md | Updates the WarpStream cluster table to remove the deleted topic and note decommissioning. |
💡 Add a code-review agent skill or configure MCP servers for context-aware, tailored reviews. Learn more in the docs.
🤖 CI report✅ Bundle size — no changeUncompressed size of every built Total: 67.86 MiB · no change No file changed by more than 1000 B. Posted automatically by build-bundle-size-report · uncompressed bytes from dist-report ✅ Eager graph — within budgetHow much code each root ships on the eager path — downloaded and parsed before the surface is interactive. Measured from the esbuild output chunks (post-tree-shake, static imports only); lazy
🟢 Largest files eagerly shipped from
|
| Size | File |
|---|---|
| 126.8 KiB | ../node_modules/.pnpm/react-dom@18.3.1_react@18.3.1/node_modules/react-dom/cjs/react-dom.production.min.js |
| 24.6 KiB | ../node_modules/.pnpm/buffer@6.0.3/node_modules/buffer/index.js |
| 6.3 KiB | ../node_modules/.pnpm/react@18.3.1/node_modules/react/cjs/react.production.min.js |
| 4.5 KiB | ../node_modules/.pnpm/@jspm+core@2.1.0/node_modules/@jspm/core/nodelibs/browser/process.js |
| 3.9 KiB | ../node_modules/.pnpm/scheduler@0.23.2/node_modules/scheduler/cjs/scheduler.production.min.js |
| 1.4 KiB | ../node_modules/.pnpm/base64-js@1.5.1/node_modules/base64-js/index.js |
| 1.3 KiB | src/RootErrorBoundary.tsx |
| 912 B | ../node_modules/.pnpm/ieee754@1.2.1/node_modules/ieee754/index.js |
| 789 B | src/scenes/ChunkLoadErrorBoundary.tsx |
| 762 B | src/index.tsx |
Largest files eagerly shipped from src/scenes/AuthenticatedShell.tsx
| Size | File |
|---|---|
| 304.9 KiB | ../node_modules/.pnpm/posthog-js@1.417.0/node_modules/posthog-js/dist/rrweb.js |
| 267.7 KiB | ../node_modules/.pnpm/@posthog+icons@0.38.0_react-dom@18.3.1_react@18.3.1__react@18.3.1/node_modules/@posthog/icons/dist/posthog-icons.es.js |
| 243.8 KiB | ../node_modules/.pnpm/posthog-js@1.417.0/node_modules/posthog-js/dist/module.js |
| 238.4 KiB | src/taxonomy/core-filter-definitions-by-group.json |
| 154.3 KiB | ../node_modules/.pnpm/re2js@0.4.1/node_modules/re2js/build/index.esm.js |
| 126.8 KiB | ../node_modules/.pnpm/react-dom@18.3.1_react@18.3.1/node_modules/react-dom/cjs/react-dom.production.min.js |
| 103.8 KiB | src/lib/api.ts |
| 94.6 KiB | ../packages/quill/packages/quill/dist/index.js |
| 93.3 KiB | ../node_modules/.pnpm/prosemirror-view@1.40.1/node_modules/prosemirror-view/dist/index.js |
| 90.6 KiB | ../node_modules/.pnpm/@tiptap+core@3.20.6_@tiptap+pm@3.20.6/node_modules/@tiptap/core/dist/index.js |
Posted automatically by check-eager-graph · sizes are eager output bytes (shipped, post-tree-shake) from the esbuild metafile · part of #32479
✅ Toolbar bundle — eager 2.22 MiB within budget
What the toolbar ships to customer pages, measured from the esbuild output (minified, post-tree-shake). The eager set is the entry plus everything statically imported from it — fetched before any feature runs; deferred chunks load lazily. The eager guardrail is 5.72 MiB. Each output file must also stay below 10 MB, where CloudFront stops compressing it. The module boundary is enforced separately by check-toolbar-graph.
| Metric | Size | Δ vs base | Budget |
|---|---|---|---|
| Eager (shipped) entry + static imports |
2.22 MiB · 17 files | no change | ████░░░░░░ 38.7% of 5.72 MiB |
| Deferred (lazy) | 2.08 MiB · 33 files | no change | n/a — loads on demand |
Loader dist/toolbar.js |
1.1 KiB | no change | █░░░░░░░░░ 5.8% of 19.5 KiB |
Largest eagerly-shipped chunks
| Size | File |
|---|---|
| 729.9 KiB | dist/toolbar/toolbar-app-5IFNCIRZ.css |
| 565.4 KiB | dist/toolbar/chunk-chunk-VRLC7S2G.js |
| 484.6 KiB | dist/toolbar/chunk-chunk-2OORENDY.js |
| 133.6 KiB | dist/toolbar/chunk-chunk-GIJIAMYW.js |
| 131.8 KiB | dist/toolbar/chunk-chunk-T5KY5WYR.js |
| 71.2 KiB | dist/toolbar/toolbar-app-2AZP2GQA.js |
| 69.0 KiB | dist/toolbar/chunk-chunk-27JL52RE.js |
| 35.6 KiB | dist/toolbar/chunk-chunk-QI6ALZJ3.js |
| 20.9 KiB | dist/toolbar/chunk-chunk-CH5JACWD.js |
| 12.2 KiB | dist/toolbar/chunk-chunk-PIK3PADE.js |
Posted automatically by check-toolbar-size · sizes are toolbar output bytes (shipped, post-tree-shake) from the esbuild metafile
✅ Dist folder size — no change
Total size of the built frontend/dist folder (all assets), compared against the base branch.
Total: 1429.37 MiB · no change
⚠️ Playwright — 1 failed
🎭 Playwright report · View test results →
❌ 1 failed test:
- Can submit the signup form multiple times if there is a generic email set (chromium)
These issues are not necessarily caused by your changes.
Annoyed by this section? Help fix flakies and failures and it will go green!
haacked
left a comment
haacked
left a comment
There was a problem hiding this comment.
Looks good, the deletion is clean. Two questions inline worth a reply before merge.
| } | ||
| // The consumer is gone. Boot with no capabilities so a pod that charts still | ||
| // launches in this mode idles instead of crash-looping on an unknown mode. | ||
| return {} |
There was a problem hiding this comment.
question: After this PR nothing in this repo can write the Postgres cohort_membership table that the flags service reads. Is something outside the monorepo meant to fill it, or is the Postgres read path being retired too?
The description names the Rust cohort-stream-processor as the replacement writer, but that service only produces Kafka messages. Its output topic defaults to cohort_membership_changed_shadow (config.rs:310), and nothing in Rust, Python, or the ClickHouse DDL consumes that topic into Postgres. The only INSERT INTO cohort_membership left in the tree is a test helper. The consumer deleted here was schema-compatible with what the processor emits: its Zod schema matched CohortMembershipChange field for field, so this removes the one component that could have been re-pointed at the shadow topic.
Nothing serving is affected today and your reasoning there holds. The table was already frozen when the topic lost its producer, and REALTIME_COHORT_EVALUATION_TEAM_IDS defaults to none, so server.rs:218 installs the no-op provider. The exposure is whoever turns that gate on next: the flags service then reads a table nobody fills and treats every person as a non-member.
| KAFKA_DWH_CDP_RAW_TABLE: KafkaClusterProfile.CYCLOTRON, | ||
| # --- CALCULATED_EVENTS (Warpstream calculated-events) --- | ||
| KAFKA_COHORT_MEMBERSHIP_CHANGED: KafkaClusterProfile.CALCULATED_EVENTS, | ||
| # --- REPLAY (Session replay) --- |
There was a problem hiding this comment.
question: Does the teardown runbook wait for #79807's drops to be applied, not just merged? ClickHouse is still bound to cohort_membership_changed on the calculated-events cluster after this PR and after the charts PR, so stopping the deployment isn't enough before the cluster is destroyed.
cohortmembership/sql.py:105-124 defines kafka_cohort_membership_ws against CLICKHOUSE_KAFKA_WARPSTREAM_CALCULATED_EVENTS_NAMED_COLLECTION, and migration 0245 creates it on US, EU, and DEV. No drop migration exists in this tree, which matches your plan to leave the drops to #79807.
A Kafka engine table pointed at destroyed brokers keeps erroring and retrying until it's dropped. That makes "this consumer is the last application binding" true for the deployment but not for ClickHouse, so the ordering section is worth updating. It currently gates teardown on the charts PR alone.
There was a problem hiding this comment.
It waits. Ordering is documented in the stack: #79807 must be applied in all three regions before #9879 removes the named collection, and #9865 deletes the topics after that. Merge alone isn't the gate — a Kafka engine table whose named collection has vanished error-loops on every poll.
Confirmed that #79807's drop set covers kafka_cohort_membership_ws / cohort_membership_ws_mv, so ClickHouse is off cohort_membership_changed before the cluster goes.
| }) | ||
| } | ||
|
|
||
| if (capabilities.cdpHogflowSubscriptionMatcher) { |
There was a problem hiding this comment.
suggestion: removing this consumer leaves the nodejs behavioral-cohorts Postgres route with no production reader, so the next person auditing which pods need those credentials will get the wrong answer. PostgresUse.BEHAVIORAL_COHORTS_RW now appears nowhere under nodejs/src except its own plumbing: the enum member, the fallback map entry, and the pool built from BEHAVIORAL_COHORTS_DATAhttps://github.com/PostHog/posthog/blob/4edeff3d9032d54f61db909f4548cb31ce3172cf_URL. Behind it sit four config keys and the Omit<PostgresRouterConfig, 'BEHAVIORAL_COHORTS_DATAhttps://github.com/PostHog/posthog/blob/4edeff3d9032d54f61db909f4548cb31ce3172cf_URL'> that exists only to keep this out of the ingestion server.
Nothing breaks at runtime, since createPostgresPool returns a lazy pool that never dials until queried, so an unused route costs no connections. The cost is a config surface that reads as live.
Removing it is safe across repos: overrideWithEnv only reads env vars matching keys already in the config object, so charts can keep setting POSTGRES_BEHAVIORAL_COHORTS_* with no effect, and the Rust readers use their own BEHAVIORAL_COHORTS_READ_DATAhttps://github.com/PostHog/posthog/blob/4edeff3d9032d54f61db909f4548cb31ce3172cf_URL. If a nodejs reader of that database is planned, leaving it is the right call. Say so and this drops.
There was a problem hiding this comment.
Leaving BEHAVIORAL_COHORTS_RW in place for now rather than removing it in this PR — it's coupled to the open question on the other thread. If the Postgres read path is being retired the whole surface should go in one change; if a replacement writer is coming it should stay. Happy to strip it here once that's settled.
… helpers Generated-By: PostHog Code Task-Id: 08dc6a1f-fda7-4722-af9e-a292f90c1f09
Generated-By: PostHog Desktop Task-Id: 37d4f498-b810-4c89-beea-0ee2cd18e719
haacked
left a comment
haacked
left a comment
There was a problem hiding this comment.
Delta since my last look is clean and every fix holds. One blocking thing inline: the cluster still has a CDP producer bound to it, so the teardown ordering isn't safe as written.
| | `warpstream-traces` | `ingestion-traces`, `clickhouse_traces` | `kminion-warpstream-traces` | | ||
| | `warpstream-shared` | `clickhouse_document_embeddings`, error tracking fingerprint/issue topics, `document_embeddings_input` | `kminion-warpstream-shared` | | ||
| | `warpstream-calculated-events` | `cohort_membership_changed` (US only) | `kminion-warpstream-calculated-events` | | ||
| | `warpstream-calculated-events` | `clickhouse_prefiltered_events`, `clickhouse_precalculated_person_properties` (US only; being decommissioned) | `kminion-warpstream-calculated-events` | |
There was a problem hiding this comment.
blocking: Destroying this cluster will crash-loop most of the CDP fleet. The ordering in the description isn't safe as written. This consumer isn't the last application binding to warpstream-calculated-events. Every CDP pod opens a Kafka producer against that cluster at boot, and the connection is boot-critical: if it fails, the pod exits.
The registry registers a producer at producer-registry.ts:32, and server.ts:117 builds that registry whenever needsCdpProducer is true (every CDP mode except cdp-hogflow-scheduler, plus the cyclotron janitor). build() creates all four producers up front and awaits each one. createWithConfig rejects when connect() fails. That rejection leaves startServices(), gets caught at base-server.ts:108, and ends at process.exit(1). node-rdkafka fetches metadata with a 30-second default timeout, so a pod takes about 30 seconds to die and then backs off.
Dropping the env vars instead doesn't help. metadata.broker.list falls back to kafka:9092, which doesn't resolve in production. The comment above the registration says the producer is kept because registration is lazy, which is true of register() but not the build() that follows.
Already-running pods survive since nothing produces to this producer and a dead connection only logs. The damage lands on the next rollout: pods that restart after the cluster goes away crash-loop, the rollout stalls, and capacity bleeds as pods cycle.
This diff is safe on its own. The ordering section needs to change, and the registration must be deleted and deployed in all three regions before charts#14082 removes the cluster wiring. Nothing routes to it, so the deletion is straightforward:
export function createCdpProducerRegistry(kafkaClientRack: string | undefined) {
return new KafkaProducerRegistryBuilder(kafkaClientRack)
.register(WARPSTREAM_INGESTION_PRODUCER, WARPSTREAM_INGESTION_PRODUCER_CONFIG_MAP)
- .register(WARPSTREAM_CALCULATED_EVENTS_PRODUCER, WARPSTREAM_CALCULATED_EVENTS_PRODUCER_CONFIG_MAP)
.register(WARPSTREAM_CYCLOTRON_PRODUCER, WARPSTREAM_CYCLOTRON_PRODUCER_CONFIG_MAP)
.register(WAREHOUSE_PRODUCER, WAREHOUSE_PRODUCER_CONFIG_MAP)
}Delete the two imports along with WARPSTREAM_CALCULATED_EVENTS_PRODUCER and its config map in producers.ts.
| } | ||
| // The consumer is gone. Boot with no capabilities so a pod that charts still | ||
| // launches in this mode idles instead of crash-looping on an unknown mode. | ||
| return {} |
There was a problem hiding this comment.
question: A flag set to "roll out to everyone except cohort X" would stop excluding anyone once the realtime cohort table is frozen. This is the failure mode on the realtime cohort read path, opposite from the fail-closed behavior described in the thread above. When a realtime cohort lookup returns nothing, the flags service treats the person as a non-member (flag_matching.rs:2218-2222), which is safe for an IN condition. But apply_cohort_membership_logic maps NotIn to !matches (cohort_operations.rs:568), and a nested cohort filter inverts the same way via negation (cohort_operations.rs:397), so the exclusion matches every person the frozen table has no row for. Anyone who entered a cohort after the freeze is wrongly included by a NOT_IN condition, not just missing from an IN one.
This PR doesn't change flag evaluation, and REALTIME_COHORT_EVALUATION_TEAM_IDS still defaults to none, so the provider answers false for everything. This removal is worth noting because this PR removes the last component that could have refilled the table. If the read path is being retired too, this concern dissolves. If the gate is meant to turn on once a new writer exists, backfill the table before the flip rather than after.
| celery_scheduler, | ||
| nodejs_realtime_cohorts, | ||
| ] | ||
| capabilities: [event_ingestion, property_definitions, property_values, celery_workers, celery_scheduler] |
There was a problem hiding this comment.
suggestion: A developer who selects only the cohorts intent now gets a much heavier local Node.js process than before, roughly thirteen services where they used to get one.
The shared nodejs proc still starts because it declares capability: event_ingestion (mprocs.yaml:66-69) and cohorts still requires event_ingestion. With no nodejs_* capability left in the resolved set, the generator leaves NODEJS_CAPABILITY_GROUPS unset (generator.py:448-450), so getPluginServerCapabilities falls through to the default local-dev branch and merges CDP workflows, error tracking, and feature flags. That was one consumer before.
web_analytics, llm_analytics, and data_warehouse already resolve to zero nodejs_* capabilities and already hit this default. A cohorts developer is probably better served by a working CDP stack than by the dead consumer they had. If you want a narrower process, add a specific group:
| capabilities: [event_ingestion, property_definitions, property_values, celery_workers, celery_scheduler] | |
| capabilities: [event_ingestion, property_definitions, property_values, celery_workers, celery_scheduler, nodejs_cdp] |
| # Node.js capabilities are controlled via NODEJS_CAPABILITY_GROUPS env var | ||
| # Set by hogli dev:generate based on selected nodejs_* capabilities | ||
| # Valid groups: cdp_workflows, realtime_cohorts, session_replay, logs, traces, metrics, feature_flags | ||
| # Valid groups: cdp_workflows, session_replay, logs, traces, metrics, feature_flags |
There was a problem hiding this comment.
nit: Four names on this line (session_replay, logs, traces, metrics) aren't in CAPABILITY_GROUP_MAP (capabilities.ts:58-62), so they hit the unknown-group warning and get dropped, while cdp works but isn't listed. That drift predates this PR, but you're editing the line anyway, and the example below it still uses session_replay.
| # Valid groups: cdp_workflows, session_replay, logs, traces, metrics, feature_flags | |
| # Valid groups: cdp, cdp_workflows, feature_flags |
|
Closing this one. The The cluster still goes away. The consumer follows its new producer instead of being deleted: #87866 makes its topic and group id configurable so a deployment can point it at the cohort-stream-processor's output topic, and PostHog/charts#14082 keeps the app and repoints it. Issue #82624 has been updated with the revised sequence. |
Problem
The
cdp-cohort-membershipconsumer readscohort_membership_changedfrom thewarpstream-calculated-eventscluster. That topic lost its producer when the legacy precalculated realtime-cohort path was deleted, so the consumer has been running against a silent topic — two pods per region doing nothing.The cluster is now being decommissioned, and this consumer is the last application binding to it.
Changes
cohort_membership_changedtopic constant and its Kafka cluster routing entry, plus a stale routing-overrides example that named the topic.monitoring-ingestion-pipelineskill, which still listed the topic.PluginServerMode.cdp_cohort_membershipstays as a stub returning no capabilities, matching the existingcdp_precalculated_filtersprecedent. An unknownPLUGIN_SERVER_MODEthrows at boot, so removing the enum member before charts stops setting it would crash-loop the deployment rather than retire it. TheTODOmarks it for deletion once the charts side lands.The ClickHouse SQL definitions for the WarpStream Kafka engine tables are deliberately left in place. Removing them would force a rewrite of the already-applied migration that imports them; #79807 drops the actual tables, which is what the decommission needs.
Ordering
This is safe to merge on its own, but the deployment must stop before the cluster is destroyed. The companion
PostHog/chartsPR removes the Argo app; this PR only removes the code behind it.Worth stating plainly: this removes the last writer to the Postgres
cohort_membershiptable. Nothing serving is affected today, because the feature-flags realtime cohort read path is gated byREALTIME_COHORT_EVALUATION_TEAM_IDS, which defaults to none and is not set anywhere. The Rustcohort-stream-processoris the intended replacement writer and is untouched here.How did you test this code?
ruff check/ruff format --checkon the three changed Python files: clean.npx tsc --noEmitinnodejs/afterpnpm install --frozen-lockfile: zero errors in any file this PR touches. The 23 errors reported are pre-existing, from unbuilt workspace packages (@posthog/hogvm,@posthog/cyclotron,@posthog/replay-anonymizer) and their implicit-any cascade.PluginServerModestub and unrelated Rust code (the shadow topic,cohort-stream-processor,cohort-seeder,feature-flags).Not done: the Python and Jest suites were not run, and no manual testing was performed. No tests were added — the change only deletes a consumer and its test.
Automatic notifications
Docs update
None.
🤖 Agent context
Autonomy: Human-driven (agent-assisted)
Authored with Claude Code (PostHog Code).
Originally one PR combining this with the ClickHouse drop migration. Split because
AGENTS.mdrequires migration PRs to be migration-only; #79807 now carries the migration and the two are independent of each other.Two things were deliberately left alone after investigation. The
PluginServerModemember had to stay to avoid a boot crash before the charts deployment is removed. And the MSK-fedkafka_cohort_membershiptable was left in place: it reads the same topic name but on a different cluster, migration0248already drops it in cloud, and what remains is the local-CI-and-hobby definition — removing it would be schema churn outside this decommission.Part of a cross-repo decommission alongside PRs in
PostHog/chartsandposthog-cloud-infra.