KAFKA-20792: Fix invalid partition -1 in FK join triggered by punctuator - #22980
Open
siddharthaDevineni wants to merge 1 commit into
Open
KAFKA-20792: Fix invalid partition -1 in FK join triggered by punctuator#22980siddharthaDevineni wants to merge 1 commit into
siddharthaDevineni wants to merge 1 commit into
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
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.
StreamTask#punctuatesets a dummyProcessorRecordContext(null topic, -1 partition and -1 offset), so records forwarded from a punctuator keep the punctuation timestamp.ProcessorContext#recordMetadatatherefore returns a present-but-dummy Òtptional` rather than an empty one.When a punctuator mutates the materialized store backing the left side of a foreign-key join and caching is enabled, the cache captures that dummy context and replays it when the change is flushed downstream.
SubscriptionSendProcessorSupplierreads partition -1 intoSubscriptionWrapper#primaryPartition, which was serialized to the subscription registration topic and later handed to the response sink partitioner, failing with ÌllegalArgumentException: Invalid partition: -1`.This change uses null for
primaryPartitionwhen no valid source partition is available.nullalready means "unknown partition, use default partitioning": the deserializer sets it for pre-VERSION_1wrappers, which is how FK joins routed responses before the field existed.Since the V1 field is a fixed-width int with no way to express absence, null is encoded as -1 on the wire and mapped back on read. The byte layout is unchanged. So, it's like we encode null as a NULL_PRIMARY_PARTITION (-1), we decode this constant as null. So, this only assigns meaning to a previously impossible value (-1 was never a valid partition).
Testing
SubscriptionSendProcessorSupplierTest: two unit tests covering dummy record metadata (partition -1) and absent record metadata.SubscriptionWrapperSerdeTest: round-trip of a V1 wrapper with a nullprimaryPartition.KTableKTableForeignKeyJoinScenarioTest#shouldNotFailWhenPunctuatorDeletesFromLeftTableStoreend-to-end testing using `TopologyTestDriver. It asserts both that no exception is thrown and that the leftJoin emits the expected tombstone. The test also fails without the production change, with the same stack trace reported in the Jira ticket.shouldThrowExceptionOnNullPrimaryPartitionV1Testis removed. It asserted an auto-unboxingNPEfrombuf.putInt(data.primaryPartition())rather than a deliberate validation:SubscriptionWrappers constructor guards other fields but notprimaryPartitionand null is already the meaningful V0 for that field.