I like the separation of layers a lot. I left some minor comments. At a high-level I think I am still stuck by the end user interface being a little awkward. Currently I think the interface is to define a class that builds a topology and then pass it in as a class. I think this is a bit weird.

I expect to do something like this:

    Topology myTopology = new TopologyBuilder(defaultDeser)
                                                 .addProcessor("my-processor", MyProcessor.class, new Source("my-source"))
                                                .addProcessor("my-other-processor", MyOtherProcessor.class, new Source("other-source", customDeser, customDeser), "my-processor");
    KafkaStreaming streaming = new KafkaStreaming(config, myTopology);
   streaming.run();

I think this is tied back to the question of whether I pass processor instances or classes. I think I may understand part of the confusion on this issue. One thing I am assuming is that the user will pass fully instantiated Objects to the processors that would be available in the init method.

So the implementation of KStream.filter currently looks like this:

    public KStream<K, V> filter(Predicate<K, V> predicate) {
        KStreamFilter<K, V> filter = new KStreamFilter<>(predicate);
        topology.addProcessor(filter, processor);
        return new KStreamImpl<>(topology, filter);
    }

What I am proposing is that it look instead like this:

    public KStream<K, V> filter(Predicate<K, V> predicate) {
        KStreamFilter<K, V> filter = new KStreamFilter<>(predicate);
        topology.addProcessor(KStreamFilter.class, new Configs("predicate", predicate));
        return this;
    }

The advantage is that the user code can now get rid of the whole Topology class with the builder. I think the order of execution for that API is quite unintuitive.

I am not convinced that we need this complexity.