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

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

            private TimestampedKeyValueStore<Bytes, SubscriptionWrapper<K>> store;
            private Sensor droppedRecordsSensor;

            @Override
            public void init(final ProcessorContext<CombinedKey<KO, K>, Change<ValueAndTimestamp<SubscriptionWrapper<K>>>> context) {
                super.init(context);
                final InternalProcessorContext<?, ?> internalProcessorContext = (InternalProcessorContext<?, ?>) context;

                droppedRecordsSensor = TaskMetrics.droppedRecordsSensor(
                    Thread.currentThread().getName(),
                    internalProcessorContext.taskId().toString(),
                    internalProcessorContext.metrics()
                );
                store = internalProcessorContext.getStateStore(storeName);

                keySchema.init(context);
            }

            @Override
            public void process(final Record<KO, SubscriptionWrapper<K>> record) {
                if (record.key() == null && !SubscriptionWrapper.Instruction.PROPAGATE_NULL_IF_NO_FK_VAL_AVAILABLE.equals(record.value().getInstruction())) {
                    dropRecord();
                    return;
                }
                if (record.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.");
                }
                context().forward(
                    record.withKey(new CombinedKey<>(record.key(), record.value().getPrimaryKey()))
                        .withValue(inferChange(record))
                        .withTimestamp(record.timestamp())
                );
            }

            private Change<ValueAndTimestamp<SubscriptionWrapper<K>>> inferChange(final Record<KO, SubscriptionWrapper<K>> record) {
                if (record.key() == null) {
                    return new Change<>(ValueAndTimestamp.make(record.value(), record.timestamp()), null);
                } else {
                    return inferBasedOnState(record);
                }
            }

            private Change<ValueAndTimestamp<SubscriptionWrapper<K>>> inferBasedOnState(final Record<KO, SubscriptionWrapper<K>> record) {
                final Bytes subscriptionKey = keySchema.toBytes(record.key(), record.value().getPrimaryKey());

                final ValueAndTimestamp<SubscriptionWrapper<K>> newValue = ValueAndTimestamp.make(record.value(), record.timestamp());
                final ValueAndTimestamp<SubscriptionWrapper<K>> oldValue = store.get(subscriptionKey);

                //This store is used by the prefix scanner in ForeignTableJoinProcessorSupplier
                if (record.value().getInstruction().equals(SubscriptionWrapper.Instruction.DELETE_KEY_AND_PROPAGATE) ||
                    record.value().getInstruction().equals(SubscriptionWrapper.Instruction.DELETE_KEY_NO_PROPAGATE)) {
                    store.delete(subscriptionKey);
                } else {
                    store.put(subscriptionKey, newValue);
                }
                return new Change<>(newValue, oldValue);
            }

            private void dropRecord() {
                if (context().recordMetadata().isPresent()) {
                    final RecordMetadata recordMetadata = context().recordMetadata().get();
                    LOG.warn(
                        "Skipping record due to null foreign key. "
                            + "topic=[{}] partition=[{}] offset=[{}]",
                        recordMetadata.topic(), recordMetadata.partition(), recordMetadata.offset()
                    );
                } else {
                    LOG.warn(
                        "Skipping record due to null foreign key. Topic, partition, and offset not known."
                    );
                }
                droppedRecordsSensor.record();
            }
        };
    }
@Test
    public void shouldDeleteKeyNoPropagateV1() {
        final StoreBuilder<TimestampedKeyValueStore<Bytes, SubscriptionWrapper<String>>> storeBuilder = storeBuilder();
        final SubscriptionReceiveProcessorSupplier<String, String> supplier = supplier(storeBuilder);
        final Processor<String,
            SubscriptionWrapper<String>,
            CombinedKey<String, String>,
            Change<ValueAndTimestamp<SubscriptionWrapper<String>>>> processor = supplier.get();
        stateStore = storeBuilder.build();
        context.addStateStore(stateStore);
        stateStore.init((StateStoreContext) context, stateStore);

        final SubscriptionWrapper<String> oldWrapper = new SubscriptionWrapper<>(
            new long[]{1L, 2L},
            Instruction.DELETE_KEY_NO_PROPAGATE,
            PK2,
            SubscriptionWrapper.VERSION_1,
            1
        );
        final ValueAndTimestamp<SubscriptionWrapper<String>> oldValue = ValueAndTimestamp.make(oldWrapper, 0);

        final Bytes key = COMBINED_KEY_SCHEMA.toBytes(FK, PK1);
        stateStore.put(key, oldValue);
        processor.init(context);

        final SubscriptionWrapper<String> newWrapper = new SubscriptionWrapper<>(
            new long[]{1L, 2L},
            Instruction.DELETE_KEY_NO_PROPAGATE,
            PK1,
            SubscriptionWrapper.VERSION_1,
            1
        );
        final ValueAndTimestamp<SubscriptionWrapper<String>> newValue = ValueAndTimestamp.make(
            newWrapper, 1L);
        final Record<String, SubscriptionWrapper<String>> record = new Record<>(
            FK,
            newWrapper,
            1L
        );
        processor.process(record);

        final List<CapturedForward<? extends CombinedKey<String, String>,
            ? extends Change<ValueAndTimestamp<SubscriptionWrapper<String>>>>> forwarded = context.forwarded();
        assertNull(stateStore.get(key));
        assertEquals(1, forwarded.size());
        assertEquals(
            record.withKey(new CombinedKey<>(FK, PK1))
                .withValue(new Change<>(newValue, oldValue)),
            forwarded.get(0).record()
        );
    }