kAFKA-9651:Handle edge case when partition metadata contains zero partitions

[nileshkumar3]

Handle edge case where partition metadata contains zero partitions.
Add a defensive check in the producer partition selection path to avoid operating on an empty partition list.
This improves robustness during transient metadata states.

[nileshkumar3]

@tombentley Can you review this

[showuon]

LGTM!

[showuon]

@nileshkumar3 , thanks for the PR! Please help change the PR title to prefix with MINOR: or KAFKA-xxxx. You can check what others did: https://github.com/apache/kafka/pulls?page=2&q=is%3Apr+is%3Aopen

Also, please make the PR description format correct and clear. Thanks.

[showuon]

Note: Although the producer side already has validated the partition size is > 0, the stream side (like here) I cannot see all places have the partition size validation. Adding a validation in the partitionFor makes sense to me to avoid missing validation. When we have a fail use case reported, I think we should fix in the root to have re-fetch metadata mechanism or something.

[nileshkumar3]

@nileshkumar3 , thanks for the PR! Please help change the PR title to prefix with MINOR: or KAFKA-xxxx. You can check what others did: https://github.com/apache/kafka/pulls?page=2&q=is%3Apr+is%3Aopen

Also, please make the PR description format correct and clear. Thanks.
@showuon done

[nileshkumar3]

Note: Although the producer side already has validated the partition size is > 0, the stream side (like here) I cannot see all places have the partition size validation. Adding a validation in the partitionFor makes sense to me to avoid missing validation. When we have a fail use case reported, I think we should fix in the root to have re-fetch metadata mechanism or something.

@showuon Added validation in StreamsMetadataState.

[showuon]

Note: Although the producer side already has validated the partition size is > 0, the stream side (like here) I cannot see all places have the partition size validation. Adding a validation in the partitionFor makes sense to me to avoid missing validation. When we have a fail use case reported, I think we should fix in the root to have re-fetch metadata mechanism or something.

@showuon Added validation in StreamsMetadataState.

Is this really needed? Let's say, if you don't have the change in the commit: 86511ce , the shouldFailWhenSourceTopicHasNoPartitionsInMetadata test can still pass because the IllegalStateException will be thrown in the partitionForKey method, right? I mean, if possible, we should fix from the root, instead of adding validation here and there, but still fail the application. WDYT?

[nileshkumar3]

Note: Although the producer side already has validated the partition size is > 0, the stream side (like here) I cannot see all places have the partition size validation. Adding a validation in the partitionFor makes sense to me to avoid missing validation. When we have a fail use case reported, I think we should fix in the root to have re-fetch metadata mechanism or something.

@showuon Added validation in StreamsMetadataState.

Is this really needed? Let's say, if you don't have the change in the commit: 86511ce , the shouldFailWhenSourceTopicHasNoPartitionsInMetadata test can still pass because the IllegalStateException will be thrown in the partitionForKey method, right? I mean, if possible, we should fix from the root, instead of adding validation here and there, but still fail the application. WDYT?

yes what you think make sense. we should add validation at the root. before streamsMetadataState.onChange in StreamThread and StreamsPartitionAssginor.