This is an automated email from the ASF dual-hosted git repository.
mjsax pushed a commit to branch 3.3
in repository https://gitbox.apache.org/repos/asf/kafka.git
The following commit(s) were added to refs/heads/3.3 by this push:
new 9dd25ecd9c KAFKA-13769 Fix version check in
SubscriptionJoinForeignProcessorSupplier (#12420)
9dd25ecd9c is described below
commit 9dd25ecd9ce17e608c6aba98e0422b26ed133c12
Author: Alex Sorokoumov <[email protected]>
AuthorDate: Mon Jul 18 23:10:02 2022 +0200
KAFKA-13769 Fix version check in SubscriptionJoinForeignProcessorSupplier
(#12420)
This commit changes the version check from != to > as the process method
works correctly on both version 1 and 2. != incorrectly throws on v1
records.
Reviewers: Matthias J. Sax <[email protected]>
---
.../foreignkeyjoin/SubscriptionJoinForeignProcessorSupplier.java | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
diff --git
a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/foreignkeyjoin/SubscriptionJoinForeignProcessorSupplier.java
b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/foreignkeyjoin/SubscriptionJoinForeignProcessorSupplier.java
index fea8e73867..56d6a13321 100644
---
a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/foreignkeyjoin/SubscriptionJoinForeignProcessorSupplier.java
+++
b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/foreignkeyjoin/SubscriptionJoinForeignProcessorSupplier.java
@@ -70,7 +70,7 @@ public class SubscriptionJoinForeignProcessorSupplier<K, KO,
VO>
Objects.requireNonNull(valueAndTimestamp, "This processor
should never see a null newValue.");
final SubscriptionWrapper<K> value = valueAndTimestamp.value();
- if (value.getVersion() != SubscriptionWrapper.CURRENT_VERSION)
{
+ if (value.getVersion() > SubscriptionWrapper.CURRENT_VERSION) {
//Guard against modifications to SubscriptionWrapper. Need
to ensure that there is compatibility
//with previous versions to enable rolling upgrades. Must
develop a strategy for upgrading
//from older SubscriptionWrapper versions to newer
versions.
