@Override
            public void process(final Record<CombinedKey<KO, K>, Change<ValueAndTimestamp<SubscriptionWrapper<K>>>> record) {
                Objects.requireNonNull(record.key(), "This processor should never see a null key.");
                Objects.requireNonNull(record.value(), "This processor should never see a null value.");
                final ValueAndTimestamp<SubscriptionWrapper<K>> valueAndTimestamp = record.value().newValue;
                Objects.requireNonNull(valueAndTimestamp, "This processor should never see a null newValue.");
                final SubscriptionWrapper<K> value = valueAndTimestamp.value();

                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.
                    throw new UnsupportedVersionException("SubscriptionWrapper is of an incompatible version.");
                }

                final ValueAndTimestamp<VO> foreignValueAndTime =
                    record.key().getForeignKey() == null ?
                        null :
                        foreignValues.get(record.key().getForeignKey());

                final long resultTimestamp =
                    foreignValueAndTime == null ?
                        valueAndTimestamp.timestamp() :
                        Math.max(valueAndTimestamp.timestamp(), foreignValueAndTime.timestamp());

                switch (value.getInstruction()) {
                    case DELETE_KEY_AND_PROPAGATE:
                        context().forward(
                            record.withKey(record.key().getPrimaryKey())
                                .withValue(new SubscriptionResponseWrapper<VO>(
                                    value.getHash(),
                                    null,
                                    value.getPrimaryPartition()
                                ))
                                .withTimestamp(resultTimestamp)
                        );
                        break;
                    case PROPAGATE_NULL_IF_NO_FK_VAL_AVAILABLE:
                        //This one needs to go through regardless of LEFT or INNER join, since the extracted FK was
                        //changed and there is no match for it. We must propagate the (key, null) to ensure that the
                        //downstream consumers are alerted to this fact.
                        final VO valueToSend = foreignValueAndTime == null ? null : foreignValueAndTime.value();

                        context().forward(
                            record.withKey(record.key().getPrimaryKey())
                                .withValue(new SubscriptionResponseWrapper<>(
                                        value.getHash(),
                                        valueToSend,
                                        value.getPrimaryPartition()
                                ))
                                .withTimestamp(resultTimestamp)
                        );
                        break;
                    case PROPAGATE_ONLY_IF_FK_VAL_AVAILABLE:
                        if (foreignValueAndTime != null) {
                            context().forward(
                                record.withKey(record.key().getPrimaryKey())
                                   .withValue(new SubscriptionResponseWrapper<>(
                                       value.getHash(),
                                       foreignValueAndTime.value(),
                                       value.getPrimaryPartition()
                                   ))
                                   .withTimestamp(resultTimestamp)
                            );
                        }
                        break;
                    case DELETE_KEY_NO_PROPAGATE:
                        break;
                    default:
                        throw new IllegalStateException("Unhandled instruction: " + value.getInstruction());
                }
            }
Receives {@code SubscriptionWrapper<K>} events and processes them according to their Instruction.
Depending on the results, {@code SubscriptionResponseWrapper}s are created, which will be propagated to
the {@code ResponseJoinProcessorSupplier} instance.

@param <K> Type of primary keys
@param <KO> Type of foreign key
@param <VO> Type of foreign value
streams/src/main/java/org/apache/kafka/streams/kstream/internals/foreignkeyjoin/SubscriptionJoinProcessorSupplier.java