@Override
    public Processor<K, SubscriptionResponseWrapper<VO>, K, VR> get() {
        return new ContextualProcessor<K, SubscriptionResponseWrapper<VO>, K, VR>() {
            private String valueHashSerdePseudoTopic;
            private Serializer<V> runtimeValueSerializer = constructionTimeValueSerializer;

            private KTableValueGetter<K, V> valueGetter;
            private Sensor droppedRecordsSensor;


            @SuppressWarnings("unchecked")
            @Override
            public void init(final ProcessorContext<K, VR> context) {
                super.init(context);
                valueHashSerdePseudoTopic = valueHashSerdePseudoTopicSupplier.get();
                valueGetter = valueGetterSupplier.get();
                valueGetter.init(context);
                if (runtimeValueSerializer == null) {
                    runtimeValueSerializer = (Serializer<V>) context.valueSerde().serializer();
                }

                final InternalProcessorContext<?, ?> internalProcessorContext = (InternalProcessorContext<?, ?>) context;
                droppedRecordsSensor = TaskMetrics.droppedRecordsSensor(
                        Thread.currentThread().getName(),
                        internalProcessorContext.taskId().toString(),
                        internalProcessorContext.metrics()
                );
            }

            @Override
            public void process(final Record<K, SubscriptionResponseWrapper<VO>> record) {
                if (record.value().getVersion() != SubscriptionResponseWrapper.CURRENT_VERSION) {
                    //Guard against modifications to SubscriptionResponseWrapper. 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("SubscriptionResponseWrapper is of an incompatible version.");
                }
                final ValueAndTimestamp<V> currentValueWithTimestamp = valueGetter.get(record.key());

                final long[] currentHash = currentValueWithTimestamp == null ?
                    null :
                    Murmur3.hash128(runtimeValueSerializer.serialize(valueHashSerdePseudoTopic, currentValueWithTimestamp.value()));

                final long[] messageHash = record.value().getOriginalValueHash();

                //If this value doesn't match the current value from the original table, it is stale and should be discarded.
                if (java.util.Arrays.equals(messageHash, currentHash)) {
                    final VR result;

                    if (record.value().getForeignValue() == null && (!leftJoin || currentValueWithTimestamp == null)) {
                        result = null; //Emit tombstone
                    } else {
                        result = joiner.apply(currentValueWithTimestamp == null ? null : currentValueWithTimestamp.value(), record.value().getForeignValue());
                    }
                    context().forward(record.withValue(result));
                } else {
                    LOG.trace("Dropping FK-join response due to hash mismatch. Expected {}. Actual {}", messageHash, currentHash);
                    droppedRecordsSensor.record();
                }
            }
        };
    }
@Test
    public void shouldEmitTombstoneForLeftJoinWhenRightIsNullAndLeftIsNull() {
        final TestKTableValueGetterSupplier<String, String> valueGetterSupplier =
            new TestKTableValueGetterSupplier<>();
        final boolean leftJoin = true;
        final ResponseJoinProcessorSupplier<String, String, String, String> processorSupplier =
            new ResponseJoinProcessorSupplier<>(
                valueGetterSupplier,
                STRING_SERIALIZER,
                () -> "value-hash-dummy-topic",
                JOINER,
                leftJoin
            );
        final Processor<String, SubscriptionResponseWrapper<String>, String, String> processor = processorSupplier.get();
        final MockInternalNewProcessorContext<String, String> context = new MockInternalNewProcessorContext<>();
        processor.init(context);
        context.setRecordMetadata("topic", 0, 0);

        valueGetterSupplier.put("lhs1", null);
        final long[] hash = null;
        processor.process(new Record<>("lhs1", new SubscriptionResponseWrapper<>(hash, null, 0), 0));
        final List<MockProcessorContext.CapturedForward<? extends String, ? extends String>> forwarded = context.forwarded();
        assertThat(forwarded.size(), is(1));
        assertThat(forwarded.get(0).record(), is(new Record<>("lhs1", null, 0)));
    }