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

        leftJoinProcessor.process(new Record<>(pk, new Change<>(null, new LeftValue(fk1)), 0));

        assertThat(context.forwarded().size(), greaterThan(0));
        assertThat(
            context.forwarded().get(0).record(),
            is(new Record<>(fk1, new SubscriptionWrapper<>(null, DELETE_KEY_AND_PROPAGATE, pk, 0), 0))
        );
    }