@Override
    public Processor<K, Change<V>, K, Change<V>> get() {
        return new KTableReduceProcessor();
    }
@Test
    public void shouldAddAndSubtract() {
        final InternalMockProcessorContext<String, Change<Set<String>>> context = new InternalMockProcessorContext<>();

        final Processor<String, Change<Set<String>>, String, Change<Set<String>>> reduceProcessor =
            new KTableReduce<String, Set<String>>(
                "myStore",
                this::unionNotNullArgs,
                this::differenceNotNullArgs
            ).get();

        final TimestampedKeyValueStore<String, Set<String>> myStore =
            new GenericInMemoryTimestampedKeyValueStore<>("myStore");

        context.register(myStore, null);
        reduceProcessor.init(context);
        context.setCurrentNode(new ProcessorNode<>("reduce", reduceProcessor, singleton("myStore")));

        reduceProcessor.process(new Record<>("A", new Change<>(singleton("a"), null), 10L));
        assertEquals(ValueAndTimestamp.make(singleton("a"), 10L), myStore.get("A"));
        reduceProcessor.process(new Record<>("A", new Change<>(singleton("b"), singleton("a")), 15L));
        assertEquals(ValueAndTimestamp.make(singleton("b"), 15L), myStore.get("A"));
        reduceProcessor.process(new Record<>("A", new Change<>(null, singleton("b")), 12L));
        assertEquals(ValueAndTimestamp.make(emptySet(), 15L), myStore.get("A"));
    }