State store support

[big-andy-coates]

Need way of getting serde for state stores (and re-partition topics)

Persisted store changelog topic names are prefixed with the application.id. This means the resource descriptor, which doesn't know the application id, can't define the topic name.

Two approaches spring to mind:

1. Provide a KafkaStateStore resource type, which takes the store name, rather than the topic name. Can build the topic name, given the application id. Will also need to know if its persisted or not (i.e. is there a changelog topic or not), and also needs to support windowing
2. Support having a place holder in the topic name, e.g. "${APPLICATION.ID}-my-first-store-changelog", where the actual topic name is determined by doing a search and replace with the actual app id.

Of these, the first is more work, but is a more fully formed solution. The latter is more of a hack. Maybe.

Random thoughts on the KafkaStateStore solution:

• would be a internal only resource type - meaning they can't be exposed by aggregates.
• would provide a toOutput method that builds an owned, but not auto-created, output topic. The idea being that if a user does want to use this weirdly named changelog topic, then in their service descriptor they define the internal store and build an output topic descriptor for it too. Question: how to handle the application id requirement in the output topic? Maybe normal topics DO need so support application ids....
• Alternatively the user can dump the changelog to an output topic, so they can control the name.
• the store should know it's retention. This should flow into the topic descriptor retention config.
• ideally, provide a way to ensureMaterialized is correctly configured with retention...

related Kafka issue for removing the need for duplicate changelog topics: https://issues.apache.org/jira/browse/KAFKA-6035

Tutorial should cover different store types. For example, could extend filter service to read whitelist from topic into global store (with global store impl provided by creek?), add scoreboard service with hourly, daily windowed stores and unwindowed version using retention settings only.

[big-andy-coates]

Thoughts...

Should be enough to:

1. Define a new Store internal resource meta type.
2. Add new topic method to the streams extension that takes the stores, works out the topic name, and returns a KafkaTopic.

Boom!

Makes sense to have a common KafkaResource base type for topic and store, used to determine if Kafka broker should be started in tests.

Store type wouldn't be auto created. Can't be used as input/output, as topic make isn't fix.