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.