Repository: apache/kafka
File: streams/src/main/java/org/apache/kafka/streams/kstream/internals/foreignkeyjoin/SubscriptionSendProcessorSupplier.java (Java)

Code before:
            final VLeft oldValue = record.value().oldValue;
            final VLeft newValue = record.value().newValue;

            final KRight oldForeignKey = oldValue == null ? null : foreignKeyExtractor.extract(record.key(), oldValue);
            final boolean needToUnsubscribe = oldForeignKey != null;

            // if left row is inserted or updated, subscribe to new FK (if new FK is valid)
            if (newValue != null) {
                final KRight newForeignKey = foreignKeyExtractor.extract(record.key(), newValue);

                if (newForeignKey == null) { // invalid FK
                    logSkippedRecordDueToNullForeignKey();
                    if (needToUnsubscribe) {
                        // delete old subscription
                        //
                        // this may lead to unnecessary tombstones if the old FK did not join;
                        // however, we cannot avoid it as we have no means to know if the old FK joined or not
                        forward(record, oldForeignKey, DELETE_KEY_AND_PROPAGATE);
                    }
                } else { // valid FK
                    // regular insert/update

                    if (needToUnsubscribe) {
                        // update case

                        final boolean foreignKeyChanged = !Arrays.equals(serialize(newForeignKey), serialize(oldForeignKey));

                        if (foreignKeyChanged) {
                            // if FK did change, we need to explicitly delete the old subscription,
                            // because the new subscription goes to a different partition
                            //
                            // we don't need any response, as we only want a response from the new subscription
                            forward(record, oldForeignKey, DELETE_KEY_NO_PROPAGATE);

                            // subscribe for new FK (note, could be on a different task/node than the old FK)
                            // additionally, propagate null if no FK is found so we can delete the previous result (if any)
                            //
                            // this may lead to unnecessary tombstones if the old FK did not join and the new FK key does not join either;
                            // however, we cannot avoid it as we have no means to know if the old FK joined or not
                            forward(record, newForeignKey, PROPAGATE_NULL_IF_NO_FK_VAL_AVAILABLE);
                        } else {
                            // if FK did not change, we only need a response from the new FK subscription, if there is a join
                            // if there is no join, we know that the old row did not join either (as it used the same FK)
                            // and thus we don't need to propagate an idempotent null result
                            forward(record, newForeignKey, PROPAGATE_ONLY_IF_FK_VAL_AVAILABLE);
                        }
                    } else {
                        // insert case

                        // subscribe to new key
                        // don't propagate null if no FK is found:

Review comment (suggestion):
Could consider
```
   private boolean foreignKeyChanged(KRight oldFK, KRight newFK) {
       return !Arrays.equals(serialize(newFK), serialize(oldFK));
   }
```
Since this appears several times here and then we can also inline this boolean

Code after:

            final KRight oldForeignKey = oldValue == null ? null : foreignKeyExtractor.extract(record.key(), oldValue);
            final boolean needToUnsubscribe = oldForeignKey != null;

            // if left row is inserted or updated, subscribe to new FK (if new FK is valid)
            if (newValue != null) {
                final KRight newForeignKey = foreignKeyExtractor.extract(record.key(), newValue);

                if (newForeignKey == null) { // invalid FK
                    logSkippedRecordDueToNullForeignKey();
                    if (needToUnsubscribe) {
                        // delete old subscription
                        //
                        // this may lead to unnecessary tombstones if the old FK did not join;
                        // however, we cannot avoid it as we have no means to know if the old FK joined or not
                        forward(record, oldForeignKey, DELETE_KEY_AND_PROPAGATE);
                    }
                } else { // valid FK
                    // regular insert/update

                    if (needToUnsubscribe) {
                        // update case

                        if (foreignKeyChanged(newForeignKey, oldForeignKey)) {
                            // if FK did change, we need to explicitly delete the old subscription,
                            // because the new subscription goes to a different partition
                            //
                            // we don't need any response, as we only want a response from the new subscription
                            forward(record, oldForeignKey, DELETE_KEY_NO_PROPAGATE);

                            // subscribe for new FK (note, could be on a different task/node than the old FK)
                            // additionally, propagate null if no FK is found so we can delete the previous result (if any)
                            //
                            // this may lead to unnecessary tombstones if the old FK did not join and the new FK key does not join either;
                            // however, we cannot avoid it as we have no means to know if the old FK joined or not
                            forward(record, newForeignKey, PROPAGATE_NULL_IF_NO_FK_VAL_AVAILABLE);
                        } else {
                            // if FK did not change, we only need a response from the new FK subscription, if there is a join
                            // if there is no join, we know that the old row did not join either (as it used the same FK)
                            // and thus we don't need to propagate an idempotent null result
                            forward(record, newForeignKey, PROPAGATE_ONLY_IF_FK_VAL_AVAILABLE);
                        }
                    } else {
                        // insert case

                        // subscribe to new key
                        // don't propagate null if no FK is found:
                        // for inserts, we know that there is no need to delete any previous result
                        forward(record, newForeignKey, PROPAGATE_ONLY_IF_FK_VAL_AVAILABLE);
                    }
                }