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

Code before:
            if (leftJoin) {
                leftJoinInstructions(record);
            } else {
                defaultJoinInstructions(record);
            }
        }

        private void leftJoinInstructions(final Record<KLeft, Change<VLeft>> record) {
            final VLeft oldValue = record.value().oldValue;
            final VLeft newValue = record.value().newValue;

            if (oldValue == null && newValue == null) {
                // no output for idempotent left hand side deletes
                return;
            }

            final KRight oldForeignKey = oldValue == null ? null : foreignKeyExtractor.extract(record.key(), oldValue);
            final KRight newForeignKey = newValue == null ? null : foreignKeyExtractor.extract(record.key(), newValue);

            final boolean maybeUnsubscribe = oldForeignKey != null;
            if (maybeUnsubscribe) {
                // delete old subscription only if FK changed
                //
                // if FK did change, we need to explicitly delete the old subscription,
                // because the new subscription goes to a different partition
                final boolean foreignKeyChanged = !Arrays.equals(serialize(newForeignKey), serialize(oldForeignKey));

                if (foreignKeyChanged) {
                    // 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_NO_PROPAGATE);
                }
            }

            // for all cases (insert, update, and delete), we send a new subscription;
            // we need to get a response back for all cases to always produce a left-join result
            //
            // note: for delete, `newForeignKey` is null, what is a "hack"
            // no actual subscription will be added for null-FK on the right hand sice, but we still get the response back we need
            //
            // 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, newForeignKey, PROPAGATE_NULL_IF_NO_FK_VAL_AVAILABLE);
        }

        private void defaultJoinInstructions(final Record<KLeft, Change<VLeft>> record) {
            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;

Review comment (suggestion):
we can check for null here to avoid the unnecessary serialization and comparison

Code after:
            if (leftJoin) {
                leftJoinInstructions(record);
            } else {
                defaultJoinInstructions(record);
            }
        }

        private void leftJoinInstructions(final Record<KLeft, Change<VLeft>> record) {
            final VLeft oldValue = record.value().oldValue;
            final VLeft newValue = record.value().newValue;

            if (oldValue == null && newValue == null) {
                // no output for idempotent left hand side deletes
                return;
            }

            final KRight oldForeignKey = oldValue == null ? null : foreignKeyExtractor.extract(record.key(), oldValue);
            final KRight newForeignKey = newValue == null ? null : foreignKeyExtractor.extract(record.key(), newValue);

            final boolean maybeUnsubscribe = oldForeignKey != null;
            if (maybeUnsubscribe) {
                // delete old subscription only if FK changed
                //
                // if FK did change, we need to explicitly delete the old subscription,
                // because the new subscription goes to a different partition
                if (foreignKeyChanged(newForeignKey, oldForeignKey)) {
                    // 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_NO_PROPAGATE);
                }
            }

            // for all cases (insert, update, and delete), we send a new subscription;
            // we need to get a response back for all cases to always produce a left-join result
            //
            // note: for delete, `newForeignKey` is null, what is a "hack"
            // no actual subscription will be added for null-FK on the right hand side, but we still get the response back we need
            //
            // 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, newForeignKey, PROPAGATE_NULL_IF_NO_FK_VAL_AVAILABLE);
        }

        private void defaultJoinInstructions(final Record<KLeft, Change<VLeft>> record) {
            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)