chia7712 commented on code in PR #24015:
URL: https://github.com/apache/kafka/pull/24015#discussion_r4453197177
##########
streams/src/main/java/org/apache/kafka/streams/kstream/internals/KTableImpl.java:
##########
@@ -1505,7 +1505,15 @@ public Optional<Set<Integer>> partitions(final String
topic,
final Headers headers,
final int numPartitions) {
final Integer partition = value.primaryPartition();
- return partition == null ? Optional.empty() :
Optional.of(Set.of(partition));
+ if (partition == null) {
+ return Optional.empty();
+ }
+ if (partition < 0 || partition >= numPartitions) {
Review Comment:
> Punctuation: the partition is -1 there, so a processValues() punctuator
forwarding into toTable() + FK join fails with "Invalid partition: -1". With
the guard it falls back to the default partitioner.
I was wondering about this case before, but the production code forward uses
`context().taskId().partition()`, which should never return -1 I guess ... ?
```java
private void forward(final Record<KLeft, Change<VLeft>> record,
final KRight foreignKey, final Instruction deleteKeyNoPropagate) {
final SubscriptionWrapper<KLeft> wrapper = new
SubscriptionWrapper<>(
hash(record),
deleteKeyNoPropagate,
record.key(),
context().taskId().partition()
);
context().forward(record.withKey(foreignKey).withValue(wrapper));
}
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
