dekaf: keep connection open on document validation error - #3359
Merged
Conversation
danielnelson
changed the title
danielnelson
changed the title
danielnelson
force-pushed
the
dbn/dekaf-schema-leader-not-available
branch
from
August 14, 2026 16:09
1521000 to
5bffb38
Compare
danielnelson
marked this pull request as ready for review
August 14, 2026 16:20
Collection schemas are refreshed every spec_ttl. Currently if a document does not validate against its collection schema the session would log an error and close the connection. This would interrupt all partitions on this connection and the client would need to reconnect. If the error is corrected by the inferred schema you it should be updated on the next spec_ttl refresh, but if the client reconnects before then it would result in further disconnections. With this change, when a partition fails to validate against the schema, it is placed into "cooldown". LEADER_NOT_AVAILABLE will be reported for this partition. This will indicate to the client that it should retry, but will not affect the ability for other partitions to make progress. The partition remains in cooldown until the collection schema is updated where on the next fetch a new Read is created, or until a timeout (4m default, 2x the spec_ttl) where the connection will be closed with an error like before.
danielnelson
force-pushed
the
dbn/dekaf-schema-leader-not-available
branch
from
August 14, 2026 16:59
5bffb38 to
d5d4c74
Compare
mdibaiee
reviewed
Aug 17, 2026
Comment on lines
+265
to
+275
| // Content-addresses the derived key/value schemas so callers can | ||
| // cheaply detect when a binding's effective schema has changed, | ||
| // without a network round trip (unlike `registered_schema_id`'s | ||
| // `avro_schema_md5`, which addresses the same schemas against the | ||
| // control plane's schema registry table). | ||
| let schema_hash = { | ||
| let key_json = serde_json::to_value(&key_schema).unwrap().to_string(); | ||
| let value_json = serde_json::to_value(&value_schema).unwrap().to_string(); | ||
| format!("{:x}", md5::compute(format!("{key_json}{value_json}"))) | ||
| }; | ||
|
|
Member
There was a problem hiding this comment.
AFAIK the value_schema here depends on field selections, which means if two bindings of the same collection have different field selections, we will have two schema hashes for the same collection.
In one iteration of the loop, one of those bindings sets the cooldown for the journal, the other one detects a change in hash and removes it, so there is an edge case where if multiple bindings of the same journal have different field selections, then they end up in this constant loop of setting a hash, then detecting a change in the hash, so on and so forth
Contributor
Author
There was a problem hiding this comment.
This seems like an issue, I'll look into it.
Contributor
Author
There was a problem hiding this comment.
The CooldownEntry is now keyed on (topic, partition).
This avoids and issue if you have two topics using the same collection.
danielnelson
force-pushed
the
dbn/dekaf-schema-leader-not-available
branch
from
August 18, 2026 22:37
742d225 to
835cae0
Compare
mdibaiee
approved these changes
Aug 19, 2026
GregorShear
added
pending:agent-api
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
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
3 participants
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.
Description:
Collection schemas are refreshed every spec_ttl. Currently if a document does not validate against its collection schema the session would log an error and close the connection. This would interrupt all partitions on this connection and the client would need to reconnect. If the error is corrected by the inferred schema you it should be updated on the next spec_ttl refresh, but if the client reconnects before then it would result in further disconnections.
With this change, when a partition fails to validate against the schema, it is placed into "cooldown". LEADER_NOT_AVAILABLE will be reported for this partition. This will indicate to the client that it should retry, but will not affect the ability for other partitions to make progress. The partition remains in cooldown until the collection schema is updated where on the next fetch a new Read is created, or until a timeout (4m default, 2x the spec_ttl) where the connection will be closed with an error like before.
closes: #3338
Workflow steps:
No changes
Documentation links affected:
None
Notes for reviewers:
I've tested this with rdkafka/kcat and tinybird