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

Code before:
            }
            droppedRecordsSensor = TaskMetrics.droppedRecordsSensor(
                Thread.currentThread().getName(),
                context.taskId().toString(),
                (StreamsMetricsImpl) context.metrics()
            );
        }

        @Override
        public void process(final Record<KLeft, Change<VLeft>> record) {
            // clear cashed hash from previous record
            recordHash = null;
            // drop out-of-order records from versioned tables (cf. KIP-914)
            if (useVersionedSemantics && !record.value().isLatest) {
                LOG.info("Skipping out-of-order record from versioned table while performing table-table join.");
                droppedRecordsSensor.record();
                return;
            }
            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 unsubscribe = oldForeignKey != null;
            if (unsubscribe) {
                // delete old subscription 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 (!Arrays.equals(serialize(newForeignKey), serialize(oldForeignKey))) {
                    // this may lead to unnecessary tombstones, if we delete an existing key,
                    // which did not join previously;
                    // 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);
                }
            }


Review comment (suggestion):
The diff is unfortunately a little bit messy, but if you compare new and old code, also because I added comments, it should become clear the the new code is much cleaner, especially for left-join case.

Code after:
            }
            droppedRecordsSensor = TaskMetrics.droppedRecordsSensor(
                Thread.currentThread().getName(),
                context.taskId().toString(),
                (StreamsMetricsImpl) context.metrics()
            );
        }

        @Override
        public void process(final Record<KLeft, Change<VLeft>> record) {
            // clear cashed hash from previous record
            recordHash = null;
            // drop out-of-order records from versioned tables (cf. KIP-914)
            if (useVersionedSemantics && !record.value().isLatest) {
                LOG.info("Skipping out-of-order record from versioned table while performing table-table join.");
                droppedRecordsSensor.record();
                return;
            }
            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;