Currently, KTable API offers some methods that confuse uses, because users think in terms of table instead of a changelog stream. Also, not all KTables are materialized and the users might want to control (ie, force) when a KTable should be materialized (for example, to allow for querying the store using interactive queries).

Importance: high

medium to high

• public API changes for KTable

• leak internal methods
• no clean separation of abstractions

Might be hard for users to understand concept.

User might be confused by verbose API (and leaking methods) they should never see.

Importance: high

medium

• need to use different imports
• change pattern to create topology with KStreamBuilder

high

• need to rewrite large parts of their code

Some APIs have non-consistent overloaded methods that might be confusing to the user (why do I need to specify this for overload A, but not for overload B? – why does overload X allow me to do this, but not overload Y)

Example:

• for KGroupedStream#aggregate required to provide argument aggValueSerde
• for KGroupedTable#aggregate has an overload without aggValueSerde

Importance: medium 

medium

• user might need to rewrite parts of the code if we deprecate some confusing overloads
• user code might get cleaner

Some interfaces like ValueJoiner only provide the values of both records to be joined, but user might want to read the key, too. For adding the key, we loose the guarantee, that the key is not modified though. (There are more similar examples, where the key is not accessible.)

Record metadata (like offset, timestamp, partition, topic) is not accessible in DSL interfaces.

Importance: low 

low

• this is more about improving the API and/or adding new features

Some very helpful classes, that are currently in package internal could get added to public API. For example, windows and some serde classes.

Importance: low

low

• we only add new stuff

• get rid of minimum retention time (that is a performance improvement that confuses many users).
• remove some leaking internal APIs

Importance: low

API is verbose and with intermixed consumer and producer configs hard to use correctly.

Importance: low

medium

• users need to rewrite the config code

ProcessorContext give access to method that cannot be called. This is hard to reason about for users.

Importance: low

low

• most user are expected to use mainly DSL

Currently, low-level API is integrated into DSL via process()/transform() and transformValues(). Those abstraction are not perfectly defined and confusing to users.

Importance: medium

medium

• most user are expected to use mainly DSL

Currently, low-level API is used to empower the user to do anything within DSL. This approach is questionable to some extents. For example, if a user wants to do a stateful 1:1 transformation of records, she must implement Transformer interface, thus has a lot of boiler plate code to access the actual state via the context and needs to implement non related methods like punctuate(). A DSL method like statefulMap with interface #map(K key, V Value, S state) might be easier to use. The question is, if DSL can provide more DSL like methods to allow more advance computations without forcing the user to too low-level.

Importance: medium

medium

• it's about adding new method so existing code should not be affected

process(), transform(), and transformValue() all accept a stat. In order to allow for scaling, state is usually partitioned by key. However, Streams does not enforce a correct partitioning (via a call to groupByKey) and thus, data might not be partitioned correctly for those three operators. The use need to be aware of this, and do a manual call to through() right now to ensure correct partitioning.

Importance: medium 

medium

• most user are expected to use mainly DSL

Simplify "message callback" use case

From mailing list:

2. For the processor api, I think this api is mostly not for end users. However this are a couple cases where it might make sense to expose it. I think users coming from Samza, or JMS's MessageListener ( https://docs.oracle.com/javaee/7/api/javax/jms/MessageListener.html) understand a simple callback interface for message processing. In fact, people often ask why Kafka's consumer doesn't provide such an interface. I'd argue we do, it's KafkaStreams. The only issue is that the processor API documentation is a bit scary for a person implementing this type of api. My observation is that people using this style of API don't do a lot of cross-message operations, then just do single message operations and use a database for anything that spans messages. They also don't factor their code into many MessageListeners and compose them, they just have one listener that has the complete handling logic. Say I am a user who wants to implement a single Processor in this style. Do we have an easy way to do that today (either with the .transform/.process methods in kstreams or with the topology apis)? Is there anything we can do in the way of trivial helper code to make this better? Also, how can we explain that pattern to people? I think currently we have pretty in-depth docs on our apis but I suspect a person trying to figure out how to implement a simple callback might get a bit lost trying to figure out how to wire it up. A simple five line example in the docs would probably help a lot. Not sure if this is best addressed in this KIP or is a side comment.

low

• would add new API and not affect current users

In Processor API, sources, processors, and sinks are solely connected to each other by name (ie, by using String). Each time, a processor or sink should be downstream to a source/processor user need to specify the corresponding name. It might be easier to allow to use the the actual "Processor Object" that should be used.

Current:

builder.addSource("soureNode", "sourceTopic").addProcessor("processor", ..., "sourceNode"); // builder returns TopologyBuilder to allow chaining

Basic Idea:

Source s = builder.addSource("sourceNode", "sourceTopic");
Processor p = builder.addProcessor("processor", ..., s); // the source object s replaces the name "sourceTopic");

The main questions we need to consider is, if we don't limit the user, and how intuitive the API will get. It's should be a low level API and there is no need to get too close to patterns as offered in the DSL.

Importance: low

We would need to expose the concept on a "node" in TopologyBuilder.

If we allow for this, we could actually get rid of all names (and only have them as optional parameters; if not specified, the name is generated and can be retrieved via Source#name()):

Source s = builder.addSource("sourceTopic");
Processor p = builder.addProcessor(..., s); // the source object s replaces the name "sourceTopic");

We could even allow chaining (that would implicitly connect nodes):

builder.addSource("sourceTopic).addProcessor(...); // no name for neither Source no Processor and Processor consumes from Source

 low

• this would add "syntactic sugar"

Currently, user can call store.close() within their user code. However, Streams will handle the stores including closing store automatically. Thus, user should not be able to call this method. For custom store, users only need to implement this method. Currently, we have a JavaDoc hint for this, but it would be better to get it into the API directly.

Importance: low

It's clumsy to use branch() as handling the returned array requires to do "index mapping" and breaks the flow.