Dekaf: namespace group IDs and invalidate topic state when binding is backfilled or collection is reset #2093
Conversation
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
jshearer
force-pushed
the
dekaf/topic_name_translation_falliable
branch
3 times, most recently
from
abb05dc to
f57382b
Compare
jshearer
force-pushed
the
dekaf/topic_name_translation_falliable
branch
from
f57382b to
9bc1da2
Compare
We've seen cases where this has panic'd in prod and it would be useful to know the topic name that it's failing with.
If `serve()` panics, it shouldn't result in an ever-growing `dekaf_total_connections` metric. So instead of directly incrementing/decrementing the connection counter, let's just periodically observe the semaphore's state and report that instead.
jshearer
force-pushed
the
dekaf/topic_name_translation_falliable
branch
from
9bc1da2 to
32bd5cc
Compare
This flag will cause topics and groups to be namespaced when reported to/from upstream Kafka. This fixes a few problems described in #2060 Specifically: * Stop passing `token` to `to_upstream_topic_name`, as the net effect was breaking your consumer group if you changed your token. * Namespacing group IDs to the materialization so that multiple consumers of different tasks can share the same group ID without worrying about conflicts * Include the partition template name and binding state key in the mapped topic names. This causes backfills and collection resets to propagate to consumers as a fresh topic with no committed offsets. Enabling `namespaced_ids` will cause all of your committed offset state to be effectively reset, so it's disabled by default. After this change is deployed, we'll want to manually go and publish all Dekaf materializations with this field explicitly disabled. Then we can change its default to enabled, and the net effect will be that all _new_ Dekaf materializations will have this turned on by default, leaving existing ones alone.
jshearer
force-pushed
the
dekaf/topic_name_translation_falliable
branch
from
32bd5cc to
e59af8a
Compare
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
Add
advanced.namespaced_idsto Dekaf endpoint configThis flag will cause topics and groups to be namespaced when reported to/from upstream Kafka. This fixes a few problems described in #2060
Specifically:
tokentoto_upstream_topic_name, as the net effect was breaking your consumer group if you changed your token.Post-deploy steps:
Setting
advanced.namespaced_idsto true will cause all of your committed offset state to be effectively reset, so it's set to false by default if not specified. After this change is deployed, we'll want to manually go and publish all Dekaf materializations with this field explicitly set to false. Then we can change its default value to true, and the net effect will be that all new Dekaf materializations will have this turned on by default, leaving existing ones alone.Other changes
We currently have a couple of consumers of materializations where the token changed after the group/topics had offsets committed. This is causing sessions to panic, and while we're actually handling that correctly because they're each an isolated Tokio task, I realized that we are "leaking"
dekaf_total_connections-- we increment the counter when the connection is established, but panic before we can decrement it. So I refactored the way we report that metric to handle panic'd sessions.I also fixed the panic by removing a couple of
unwrap()s which I thought wouldn't fail, but do in this circumstance.Fixes #2083, fixes #2060
This change is