feat: wire codec.encode into all producers, add BatchCodecProto for batch-aware encoding#2850
Open
ce1ebrimbor wants to merge 19 commits into
Open
feat: wire codec.encode into all producers, add BatchCodecProto for batch-aware encoding#2850ce1ebrimbor wants to merge 19 commits into
ce1ebrimbor wants to merge 19 commits into
Conversation
… Kafka/Confluent publish_batch
…quence handled by BatchCodecProto
…patch, and fallback tests
github-actions
Bot
added
Confluent
ce1ebrimbor
commented
May 1, 2026
ce1ebrimbor
changed the title
…er decoder resolution
… round-trip batch codec tests
ce1ebrimbor
commented
May 2, 2026
| msg = "Cannot use both 'codec' and 'decoder' — 'codec' replaces 'decoder'." | ||
| raise ValueError(msg) | ||
|
|
||
| is_batch = getattr(self._decoder, "__name__", "") == "decode_batch" |
Contributor
Author
There was a problem hiding this comment.
will be eventually removed with the decoder= config
Contributor
Author
|
@Lancetnik there are some changes in the internals, mainly the decoding and encoding are async now, (example: codecs for kafka may need to ping schema registry). Let me know if this works for you. Documentation is coming in a separate PR. |
Member
|
@ce1ebrimbor thank you! Looks good, but I need to dig into - there are some changes I want to look close |
Member
|
Give me one more day, please |
Contributor
Author
No pressure, take your time. |
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
2 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.
Production-Side Codec Encode Wiring
PR3 in the parser/codec unification series (#2841 )
What
Wires
codec.encode()into all 6 Producers, replacing directencode_message()calls. AddsBatchCodecProtofor batch-aware codecs. Custom codecs now work symmetrically on both consume and publish sides.Changes
Codec on producers
codec: CodecProtoattribute onProducerProtoand all 6 producer implementationscodecpassed from broker configconnect()toproducer.connect()DefaultCodec()used when no custom codec is set — zero behavior changeEncode wiring
encode_message()calls replaced withawait self.codec.encode()in Kafka, Confluent, NATS, MQTT producersAioPikaParser.encode_messageconverted from sync staticmethod to async withcodecparameterMessageFormat.build/encodeandBinaryMessageFormatV1.encodeconverted to async withcodecparameterBreaking Changes
build_message()in all broker testing modules is nowasync— callers mustawaititAioPikaParser.encode_message()is nowasync— callers mustawaititMessageFormat.build()andMessageFormat.encode()are nowasyncWhat Did NOT Change
encode_message()free function infaststream/message/utils.pydecode_message()and consumption-side codec wiring (PR2)CodecProto.decode/DefaultCodec.decode_get_parser_and_decoder()subscriber wiring