Skip to content

chore(cohorts): make the cohort membership consumer's source configurable - #87866

Closed
gustavohstrassburger wants to merge 1 commit into
masterfrom
posthog/keep-cdp-cohort-membership-consumer
Closed

gustavohstrassburger wants to merge 1 commit into
masterfrom
posthog/keep-cdp-cohort-membership-consumer

Conversation

@gustavohstrassburger

Copy link
Copy Markdown
Contributor

Problem

The cdp-cohort-membership consumer is the only writer to the Postgres cohort_membership table, and that table is what the feature-flags realtime cohort read path queries (rust/feature-flags/src/cohorts/membership/realtime_provider.rs). The new Rust cohort-stream-processor only produces membership changes to Kafka; nothing else persists them.

#79808 proposed deleting the consumer as part of decommissioning the warpstream-calculated-events cluster. That would remove the sink the realtime behavioral cohorts path needs, so it is closed in favour of this PR: retire the cluster, keep the consumer.

The consumer's topic was hardcoded, so it could not follow its producer to a different cluster and topic without a code change.

Refs #82624

Changes

  • The cohort membership consumer's Kafka source is now set per deployment via CDP_COHORT_MEMBERSHIP_CONSUMER_TOPIC and CDP_COHORT_MEMBERSHIP_CONSUMER_GROUP_ID, following the existing CDP_LEGACY_EVENT_CONSUMER_* pattern. Defaults match today's behaviour, so nothing changes until a deployment sets them.
  • Mechanical: added the cohort_membership_changed_shadow topic constant, and dropped the cohort_membership_changed entry from the Django Kafka routing map — no Python producer remains and it pointed at the cluster being destroyed.

How did you test this code?

  • npx tsc --noEmit in nodejs/: no errors in any file this PR touches. The remaining errors are pre-existing and come from unbuilt workspace packages (@posthog/hogvm, @posthog/replay-anonymizer).
  • ruff check and ruff format --check on the two changed Python files: clean.
  • No new tests. The existing cdp-cohort-membership.consumer.test.ts already builds the consumer from config and produces to the resulting topic, so a hardcoded or mis-wired topic fails it.
  • Not done: no manual or cluster testing, and the Jest and Python suites were not run locally.

Automatic notifications

  • Publish to changelog?

Docs update

None.

🤖 Agent context

Autonomy: Human-driven (agent-assisted)

Authored with Claude Code (PostHog Desktop). No repo skills were invoked.

The starting point was a review of #82624's plan rather than a code change: tracing every reader and writer of the Postgres cohort_membership table showed the consumer is load-bearing for the realtime path, not dead weight. The alternative of leaving the topic hardcoded and repointing only the brokers was rejected — the producer writes to a different topic name, so the consumer needs both.


Created with PostHog Desktop

…able

The cdp-cohort-membership consumer is the only writer to the Postgres
cohort_membership table, which the feature-flags realtime cohort path reads.
It stays, but its Kafka source has to move off the calculated-events cluster
that is being decommissioned.

Add CDP_COHORT_MEMBERSHIP_CONSUMER_TOPIC and _GROUP_ID, following the existing
CDP_LEGACY_EVENT_CONSUMER_* precedent, so a deployment can point the consumer at
the cohort-stream-processor's output topic without a code change. Defaults keep
today's behaviour.

Also drop the now-dead cohort_membership_changed entry from the Django Kafka
routing map: no Python producer remains, and it pointed at the cluster going
away.

Generated-By: PostHog Desktop
Task-Id: dcb5016e-2bdc-425b-9b29-c455bbee721e
@trunk-io

trunk-io Bot commented Aug 24, 2026

Copy link
Copy Markdown

Merging to master in this repository is managed by Trunk.

  • To merge this pull request, check the box to the left or comment /trunk merge below.

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

@github-actions

Copy link
Copy Markdown
Contributor

🤖 CI report

⚠️ Trunk lane — backend Python lane

This PR is assigned to the backend Python lane. It runs backend Python tests and may merge in parallel with PRs in other lanes.

Copilot AI left a comment

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Pull request overview

Makes the cohort membership consumer’s Kafka topic and group configurable while removing obsolete routing.

Changes:

  • Adds deployment-configurable topic and group defaults.
  • Adds the shadow topic constant.
  • Removes stale Django Kafka routing.

Reviewed changes

Copilot reviewed 5 out of 5 changed files in this pull request and generated 1 comment.

Show a summary per file
File Summary
posthog/settings/kafka.py Updates routing configuration examples.
posthog/kafka_client/routing.py Removes obsolete cohort topic routing.
nodejs/src/common/config/kafka-topics.ts Adds the shadow topic constant.
nodejs/src/cdp/consumers/cdp-cohort-membership.consumer.ts Uses configurable Kafka source settings; custom-value coverage is still needed.
nodejs/src/cdp/config.ts Defines configurable consumer settings and defaults.

💡 Add a code-review agent skill or configure MCP servers for context-aware, tailored reviews. Learn more in the docs.

Comment on lines +34 to +35
groupId: config.CDP_COHORT_MEMBERSHIP_CONSUMER_GROUP_ID,
topic: config.CDP_COHORT_MEMBERSHIP_CONSUMER_TOPIC,
@trunk-io

trunk-io Bot commented Aug 24, 2026

Copy link
Copy Markdown

Static Badge   Static Badge   Static Badge

View Full Report ↗︎ ⋅ Docs

@gustavohstrassburger

Copy link
Copy Markdown
Contributor Author

Instead of renaming, we'll create a fresh topic with the same hardcoded name.

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

Projects

Archived in project

Development

Successfully merging this pull request may close these issues.

2 participants