@Override
    public Processor<CombinedKey<KO, K>, Change<ValueAndTimestamp<SubscriptionWrapper<K>>>, K, SubscriptionResponseWrapper<VO>> get() {

        return new ContextualProcessor<CombinedKey<KO, K>, Change<ValueAndTimestamp<SubscriptionWrapper<K>>>, K, SubscriptionResponseWrapper<VO>>() {

            private KTableValueGetter<KO, VO> foreignValues;

            @Override
            public void init(final ProcessorContext<K, SubscriptionResponseWrapper<VO>> context) {
                super.init(context);
                foreignValues = foreignValueGetterSupplier.get();
                foreignValues.init(context);
            }

            @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());
                }
            }
        };
    }
@Test
    public void shouldPropagateNullIfNoFKAvailableV1() {
        final MockProcessorContext<String, SubscriptionResponseWrapper<String>> context = new MockProcessorContext<>();
        processor.init(context);

        final SubscriptionWrapper<String> newValue = new SubscriptionWrapper<>(
            new long[]{1L},
            Instruction.PROPAGATE_NULL_IF_NO_FK_VAL_AVAILABLE,
            "pk1",
            SubscriptionWrapper.VERSION_1,
            12);
        Record<CombinedKey<String, String>, Change<ValueAndTimestamp<SubscriptionWrapper<String>>>> record =
            new Record<>(
                new CombinedKey<>("fk1", "pk1"),
                new Change<>(ValueAndTimestamp.make(newValue, 1L), null),
                1L
            );
        processor.process(record);
        List<CapturedForward<? extends String, ? extends SubscriptionResponseWrapper<String>>> forwarded = context.forwarded();
        assertEquals(1, forwarded.size());
        assertEquals(
            new Record<>(
                "pk1",
                new SubscriptionResponseWrapper<>(
                    newValue.getHash(),
                    "foo",
                    12
                ),
                1L
            ),
            forwarded.get(0).record());

        record = new Record<>(
            new CombinedKey<>("fk9000", "pk1"),
            new Change<>(ValueAndTimestamp.make(newValue, 1L), null),
            1L
        );
        processor.process(record);
        // propagate null if there is no match
        forwarded = context.forwarded();
        assertEquals(2, forwarded.size());
        assertEquals(
            new Record<>(
                "pk1",
                new SubscriptionResponseWrapper<>(
                    newValue.getHash(),
                    null,
                    12
                ),
                1L
            ),
            forwarded.get(1).record());
    }