@Override
    public Processor<K, Change<V>, KO, SubscriptionWrapper<K>> get() {
        return new UnbindChangeProcessor();
    }
@Test
    public void leftJoinShouldPropagateChangeFromNullFKToNullFKValue() {
        final MockInternalNewProcessorContext<String, SubscriptionWrapper<String>> context = new MockInternalNewProcessorContext<>();
        leftJoinProcessor.init(context);
        context.setRecordMetadata("topic", 0, 0);

        final LeftValue leftRecordValue = new LeftValue(null);

        leftJoinProcessor.process(new Record<>(pk, new Change<>(leftRecordValue, leftRecordValue), 0));

        assertThat(context.forwarded().size(), is(1));
        assertThat(
            context.forwarded().get(0).record(),
            is(new Record<>(null, new SubscriptionWrapper<>(hash(leftRecordValue), PROPAGATE_NULL_IF_NO_FK_VAL_AVAILABLE, pk, 0), 0))
        );
    }