Yes, you get compile time errors
On Wed, 13 Jul 2016 at 16:22 Damian Guy <damian....@gmail.com> wrote:
> You wont get a runtime error as you wouldn't find a store of that type.
> The API would return null
>
> On Wed, 13 Jul 2016 at 16:22 Jay Kreps <j...@confluent.io> wrote:
>
>> But if "my-store" is not of type MyStoreType don't you still get a run
>> time
>> error that in effect is the same as the class cast would be? Basically the
>> question I'm asking is whether this added complexity is actually moving
>> runtime errors to compile time errors.
>>
>> -Jay
>>
>> On Wed, Jul 13, 2016 at 4:16 PM, Damian Guy <damian....@gmail.com> wrote:
>>
>> > You create your custom Store, i.e,:
>> >
>> > /**
>> > * An interface your custom store provides
>> > * @param <K>
>> > * @param <V>
>> > */
>> > interface MyStoreType<K,V> {
>> > V get(K key);
>> > void put(K key, V value);
>> > }
>> >
>> > /**
>> > * Implement your store
>> > * @param <K>
>> > * @param <V>
>> > */
>> > public class MyStoreImpl<K,V> implements StateStore, MyStoreType<K,V> {
>> > // implementation of the store goes here
>> > }
>> >
>> >
>> > Provide an implementation of QueryableStoreType to find stores that
>> match
>> > your Custom store:
>> >
>> > /**
>> > * Implement QueryableStoreType to find stores that match your Custom
>> Store
>> > * @param <K>
>> > * @param <V>
>> > */
>> >
>> > public class MyQueryableType<K, V> implements
>> > QueryableStoreType<MyStoreType<K, V>>{
>> > @Override
>> > public boolean accepts(final StateStore stateStore) {
>> > return stateStore instanceof MyQueryableType;
>> > }
>> >
>> > @Override
>> > public MyStoreType<K,V> create(final StateStoreProvider
>> > storeProvider, final String storeName) {
>> > return new MyCompositeStore<>(storeName, storeProvider);
>> > }
>> > }
>> >
>> >
>> > Create a composite type to wrap the potentially many underlying
>> instances
>> > of the store, i.e, there will be one per partition
>> >
>> > /**
>> > * Provide a wrapper over the underlying store instances.
>> > */
>> > public class MyCompositeStore<K,V> implements MyStoreType<K,V> {
>> > private final String storeName;
>> > private final StateStoreProvider provider;
>> >
>> > public MyCompositeStore(final String storeName, final
>> > StateStoreProvider provider) {
>> > this.storeName = storeName;
>> > this.provider = provider;
>> > }
>> >
>> > @Override
>> > public V get(final K key) {
>> > final List<MyStoreType<K, V>> stores =
>> > provider.getStores(storeName, new MyQueryableType<K,V>());
>> > // iterate over stores looking for key
>> > }
>> >
>> > @Override
>> > public void put(final K key, V value) {
>> >
>> > }
>> > }
>> >
>> >
>> > Lookup your new store from KafkaStreams:
>> >
>> > final MyStoreType store = kafkaStreams.store("my-store", new
>> > MyQueryableType<>());
>> >
>> >
>> > So we get type safety and we can constrain the interfaces returned to
>> Read
>> > Only versions (which is what we are doing for KeyValue and Window
>> Stores)
>> >
>> > HTH,
>> > Damian
>> >
>> > On Wed, 13 Jul 2016 at 15:30 Jay Kreps <j...@confluent.io> wrote:
>> >
>> > > But to avoid the cast you introduce a bunch of magic that doesn't
>> really
>> > > bring type safety, right? Or possibly I'm misunderstanding, how do I
>> plug
>> > > in a new store type and get access to it? Can you give the steps for
>> > that?
>> > >
>> > > -Jay
>> > >
>> > > On Wed, Jul 13, 2016 at 10:47 AM, Guozhang Wang <wangg...@gmail.com>
>> > > wrote:
>> > >
>> > > > Personally I think the additional complexity of the introduced "
>> > > > QueryableStoreType" interface is still acceptable from a user's
>> point
>> > of
>> > > > view: this is the only interface we are exposing to users, and other
>> > > > wrappers are all internal classes.
>> > > >
>> > > > Regarding "QueryableStoreTypes", maybe we can consider declaring its
>> > > > "QueryableStoreTypeMatcher" as private instead of public, since
>> > > > "QueryableStoreTypes" is just used as a convenient manner for using
>> > > > library-provided types, like serialization/Serdes.java.
>> > > >
>> > > > With this the only additional interface the library is exposing is "
>> > > > QueryableStoreType", and users optionally can just use
>> > > > "QueryableStoreTypes"
>> > > > to conveniently create library-provided store types.
>> > > >
>> > > >
>> > > > Guozhang
>> > > >
>> > > >
>> > > > On Wed, Jul 13, 2016 at 7:58 AM, Neha Narkhede <n...@confluent.io>
>> > > wrote:
>> > > >
>> > > > > Damian -- appreciate the example code and you convinced me. Agree
>> > that
>> > > > the
>> > > > > class approach is better and renaming to KafkaStreamsMetadata
>> along
>> > > with
>> > > > > renaming the API methods will address the issues I was referring
>> to.
>> > > > >
>> > > > > One other thing I wanted to get people's thoughts on was the way
>> we
>> > are
>> > > > > proposing to handle different store types. I am sure you guys have
>> > > > thought
>> > > > > about the tradeoffs of using the store wrappers and matchers (
>> > > > > QueryableStoreType) vs just making users cast the returned store
>> to
>> > the
>> > > > > type they would expect to use. That is simple but the obvious
>> > downside
>> > > is
>> > > > > that it is likely to result in exceptions for users that don't
>> know
>> > > what
>> > > > > they are doing.
>> > > > >
>> > > > > In my experience of dealing with apps that would use queriable
>> state,
>> > > it
>> > > > > appears to me that a majority would just use the key value store.
>> > > Partly
>> > > > > because that will suffice and partly because people might just
>> follow
>> > > the
>> > > > > simpler examples we provide that use key-value store. For advanced
>> > > users,
>> > > > > they will be aware of the reason they want to use the windowed
>> store
>> > > and
>> > > > > will know how to cast it. The advantage of the current approach is
>> > that
>> > > > it
>> > > > > is likely more robust and general but involves introduces more
>> > > interfaces
>> > > > > and wrapper code.
>> > > > >
>> > > > > I tend to prefer simplicity to optimize for the general case, but
>> > > curious
>> > > > > to get people's thoughts on this as well.
>> > > > >
>> > > > > On Wed, Jul 13, 2016 at 8:13 AM, Jim Jagielski <j...@jagunet.com>
>> > > wrote:
>> > > > >
>> > > > > > IMO, that makes the most sense.
>> > > > > >
>> > > > > > > On Jul 12, 2016, at 5:11 PM, Ismael Juma <ism...@juma.me.uk>
>> > > wrote:
>> > > > > > >
>> > > > > > > Hi Damian,
>> > > > > > >
>> > > > > > > How about StreamsMetadata instead? The general naming pattern
>> > seems
>> > > > to
>> > > > > > > avoid the `Kafka` prefix for everything outside of
>> `KafkaStreams`
>> > > > > itself.
>> > > > > > >
>> > > > > > > Ismael
>> > > > > > >
>> > > > > > > On Tue, Jul 12, 2016 at 7:14 PM, Damian Guy <
>> > damian....@gmail.com>
>> > > > > > wrote:
>> > > > > > >
>> > > > > > >> Hi,
>> > > > > > >>
>> > > > > > >> I agree with point 1. application.server is a better name for
>> > the
>> > > > > config
>> > > > > > >> (we'll change this). However, on point 2 I think we should
>> stick
>> > > > > mostly
>> > > > > > >> with what we already have. I've tried both ways of doing this
>> > when
>> > > > > > working
>> > > > > > >> on the JIRA and building examples and I find the current
>> > approach
>> > > > more
>> > > > > > >> intuitive and easier to use than the Map based approach.
>> > > > > > >> However, there is probably a naming issue. We should rename
>> > > > > > >> KafkaStreamsInstance to KafkaStreamsMetadata. This Class is
>> very
>> > > > > simple,
>> > > > > > >> but provides all the information a developer needs to be
>> able to
>> > > > find
>> > > > > > the
>> > > > > > >> instance(s) of a Streams application that a particular store
>> is
>> > > > > running
>> > > > > > on,
>> > > > > > >> i.e.,
>> > > > > > >>
>> > > > > > >> public class KafkStreamsMetadata {
>> > > > > > >> private final HostInfo hostInfo;
>> > > > > > >> private final Set<String> stateStoreNames;
>> > > > > > >> private final Set<TopicPartition> topicPartitions;
>> > > > > > >>
>> > > > > > >>
>> > > > > > >> So using the API to route to a new host is fairly simple,
>> > > > particularly
>> > > > > > in
>> > > > > > >> the case when you want to find the host for a particular key,
>> > > i.e.,
>> > > > > > >>
>> > > > > > >> final KafkaStreams kafkaStreams = createKafkaStreams();
>> > > > > > >> final KafkaStreamsMetadata streamsMetadata =
>> > > > > > >> kafkaStreams.instanceWithKey("word-count", "hello",
>> > > > > > >> Serdes.String().serializer());
>> > > > > > >> http.get("http://"; + streamsMetadata.host() + ":" +
>> > > > > > >> streamsMetadata.port() + "/get/word-count/hello");
>> > > > > > >>
>> > > > > > >>
>> > > > > > >> And if you want to do a scatter gather approach:
>> > > > > > >>
>> > > > > > >> final KafkaStreams kafkaStreams = createKafkaStreams();
>> > > > > > >> final Collection<KafkaStreamsMetadata> kafkaStreamsMetadatas
>> =
>> > > > > > >> kafkaStreams.allInstancesWithStore("word-count");
>> > > > > > >> for (KafkaStreamsMetadata streamsMetadata :
>> > > kafkaStreamsMetadatas) {
>> > > > > > >> http.get("http://"; + streamsMetadata.host() + ":" +
>> > > > > > >> streamsMetadata.port() + "/get/word-count/hello");
>> > > > > > >> ...
>> > > > > > >> }
>> > > > > > >>
>> > > > > > >>
>> > > > > > >> And if you iterated over all instances:
>> > > > > > >>
>> > > > > > >> final KafkaStreams kafkaStreams = createKafkaStreams();
>> > > > > > >> final Collection<KafkaStreamsMetadata> kafkaStreamsMetadatas
>> =
>> > > > > > >> kafkaStreams.allInstances();
>> > > > > > >> for (KafkaStreamsMetadata streamsMetadata :
>> > > kafkaStreamsMetadatas) {
>> > > > > > >> if
>> > (streamsMetadata.stateStoreNames().contains("word-count")) {
>> > > > > > >> http.get("http://"; + streamsMetadata.host() + ":" +
>> > > > > > >> streamsMetadata.port() + "/get/word-count/hello");
>> > > > > > >> ...
>> > > > > > >> }
>> > > > > > >> }
>> > > > > > >>
>> > > > > > >>
>> > > > > > >> If we were to change this to use Map<HostInfo,
>> > Set<TaskMetadata>>
>> > > > for
>> > > > > > the
>> > > > > > >> most part users would need to iterate over the entry or key
>> set.
>> > > > > > Examples:
>> > > > > > >>
>> > > > > > >> The finding an instance by key is a little odd:
>> > > > > > >>
>> > > > > > >> final KafkaStreams kafkaStreams = createKafkaStreams();
>> > > > > > >> final Map<HostInfo, Set<TaskMetadata>> streamsMetadata =
>> > > > > > >> kafkaStreams.instanceWithKey("word-count","hello",
>> > > > > > >> Serdes.String().serializer());
>> > > > > > >> // this is a bit odd as i only expect one:
>> > > > > > >> for (HostInfo hostInfo : streamsMetadata.keySet()) {
>> > > > > > >> http.get("http://"; + streamsMetadata.host() + ":" +
>> > > > > > >> streamsMetadata.port() + "/get/word-count/hello");
>> > > > > > >> }
>> > > > > > >>
>> > > > > > >>
>> > > > > > >> The scatter/gather by store is fairly similar to the previous
>> > > > example:
>> > > > > > >>
>> > > > > > >> final KafkaStreams kafkaStreams = createKafkaStreams();
>> > > > > > >> final Map<HostInfo, Set<TaskMetadata>> streamsMetadata =
>> > > > > > >> kafkaStreams.allInstancesWithStore("word-count");
>> > > > > > >> for(HostInfo hostInfo : streamsMetadata.keySet()) {
>> > > > > > >> http.get("http://"; + hostInfo.host() + ":" +
>> > hostInfo.port() +
>> > > > > > >> "/get/word-count/hello");
>> > > > > > >> ...
>> > > > > > >> }
>> > > > > > >>
>> > > > > > >> And iterating over all instances:
>> > > > > > >>
>> > > > > > >> final Map<HostInfo, Set<TaskMetadata>> streamsMetadata =
>> > > > > > >> kafkaStreams.allInstances();
>> > > > > > >> for (Map.Entry<HostInfo, Set<TaskMetadata>> entry :
>> > > > > > >> streamsMetadata.entrySet()) {
>> > > > > > >> for (TaskMetadata taskMetadata : entry.getValue()) {
>> > > > > > >> if
>> > (taskMetadata.stateStoreNames().contains("word-count"))
>> > > {
>> > > > > > >> http.get("http://"; + streamsMetadata.host() +
>> ":" +
>> > > > > > >> streamsMetadata.port() + "/get/word-count/hello");
>> > > > > > >> ...
>> > > > > > >> }
>> > > > > > >> }
>> > > > > > >> }
>> > > > > > >>
>> > > > > > >>
>> > > > > > >> IMO - having a class we return is the better approach as it
>> > nicely
>> > > > > wraps
>> > > > > > >> the related things, i.e, host:port, store names, topic
>> > partitions
>> > > > into
>> > > > > > an
>> > > > > > >> Object that is easy to use. Further we could add some
>> behaviour
>> > to
>> > > > > this
>> > > > > > >> class if we felt it necessary, i.e, hasStore(storeName) etc.
>> > > > > > >>
>> > > > > > >> Anyway, i'm interested in your thoughts.
>> > > > > > >>
>> > > > > > >> Thanks,
>> > > > > > >> Damian
>> > > > > > >>
>> > > > > > >> On Mon, 11 Jul 2016 at 13:47 Guozhang Wang <
>> wangg...@gmail.com>
>> > > > > wrote:
>> > > > > > >>
>> > > > > > >>> 1. Re StreamsConfig.USER_ENDPOINT_CONFIG:
>> > > > > > >>>
>> > > > > > >>> I agree with Neha that Kafka Streams can provide the bare
>> > minimum
>> > > > > APIs
>> > > > > > >> just
>> > > > > > >>> for host/port, and user's implemented layer can provide URL
>> /
>> > > proxy
>> > > > > > >> address
>> > > > > > >>> they want to build on top of it.
>> > > > > > >>>
>> > > > > > >>>
>> > > > > > >>> 2. Re Improving KafkaStreamsInstance interface:
>> > > > > > >>>
>> > > > > > >>> Users are indeed aware of "TaskId" class which is not part
>> of
>> > > > > internal
>> > > > > > >>> packages and is exposed in PartitionGrouper interface that
>> can
>> > be
>> > > > > > >>> instantiated by the users, which is assigned with input
>> topic
>> > > > > > partitions.
>> > > > > > >>> So we can probably change the APIs as:
>> > > > > > >>>
>> > > > > > >>> Map<HostState, Set<TaskMetadata>> KafkaStreams.getAllTasks()
>> > > where
>> > > > > > >>> TaskMetadata has fields such as taskId, list of assigned
>> > > > partitions,
>> > > > > > list
>> > > > > > >>> of state store names; and HostState can include hostname /
>> > port.
>> > > > The
>> > > > > > port
>> > > > > > >>> is the listening port of a user-defined listener that users
>> > > provide
>> > > > > to
>> > > > > > >>> listen for queries (e.g., using REST APIs).
>> > > > > > >>>
>> > > > > > >>> Map<HostState, Set<TaskMetadata>>
>> > > > > KafkaStreams.getTasksWithStore(String
>> > > > > > >> /*
>> > > > > > >>> storeName */) would return only the hosts and their assigned
>> > > tasks
>> > > > if
>> > > > > > at
>> > > > > > >>> least one of the tasks include the given store name.
>> > > > > > >>>
>> > > > > > >>> Map<HostState, Set<TaskMetadata>>
>> > > > > > KafkaStreams.getTaskWithStoreAndKey(Key
>> > > > > > >>> k, String /* storeName */, StreamPartitioner partitioner)
>> would
>> > > > > return
>> > > > > > >> only
>> > > > > > >>> the host and their assigned task if the store with the store
>> > name
>> > > > > has a
>> > > > > > >>> particular key, according to the partitioner behavior.
>> > > > > > >>>
>> > > > > > >>>
>> > > > > > >>>
>> > > > > > >>> Guozhang
>> > > > > > >>>
>> > > > > > >>>
>> > > > > > >>> On Sun, Jul 10, 2016 at 11:21 AM, Neha Narkhede <
>> > > n...@confluent.io
>> > > > >
>> > > > > > >> wrote:
>> > > > > > >>>
>> > > > > > >>>> Few thoughts that became apparent after observing example
>> code
>> > > of
>> > > > > what
>> > > > > > >> an
>> > > > > > >>>> application architecture and code might look like with
>> these
>> > > > > changes.
>> > > > > > >>>> Apologize for the late realization hence.
>> > > > > > >>>>
>> > > > > > >>>> 1. "user.endpoint" will be very differently defined for
>> > > respective
>> > > > > > >>>> applications. I don't think Kafka Streams should
>> generalize to
>> > > > > accept
>> > > > > > >> any
>> > > > > > >>>> connection URL as we expect to only expose metadata
>> expressed
>> > as
>> > > > > > >> HostInfo
>> > > > > > >>>> (which is defined by host & port) and hence need to
>> interpret
>> > > the
>> > > > > > >>>> "user.endpoint" as host & port. Applications will have
>> their
>> > own
>> > > > > > >> endpoint
>> > > > > > >>>> configs that will take many forms and they will be
>> responsible
>> > > for
>> > > > > > >>> parsing
>> > > > > > >>>> out host and port and configuring Kafka Streams
>> accordingly.
>> > > > > > >>>>
>> > > > > > >>>> If we are in fact limiting to host and port, I wonder if we
>> > > should
>> > > > > > >> change
>> > > > > > >>>> the name of "user.endpoint" into something more specific.
>> We
>> > > have
>> > > > > > >> clients
>> > > > > > >>>> expose host/port pairs as "bootstrap.servers". Should this
>> be
>> > > > > > >>>> "application.server"?
>> > > > > > >>>>
>> > > > > > >>>> 2. I don't think we should expose another abstraction
>> called
>> > > > > > >>>> KafkaStreamsInstance to the user. This is related to the
>> > > > discussion
>> > > > > of
>> > > > > > >>> the
>> > > > > > >>>> right abstraction that we want to expose to an application.
>> > The
>> > > > > > >>> abstraction
>> > > > > > >>>> discussion itself should probably be part of the KIP
>> itself,
>> > let
>> > > > me
>> > > > > > >> give
>> > > > > > >>> a
>> > > > > > >>>> quick summary of my thoughts here:
>> > > > > > >>>> 1. The person implementing an application using Queryable
>> > State
>> > > > has
>> > > > > > >>> likely
>> > > > > > >>>> already made some choices for the service layer–a REST
>> > > framework,
>> > > > > > >> Thrift,
>> > > > > > >>>> or whatever. We don't really want to add another RPC
>> framework
>> > > to
>> > > > > this
>> > > > > > >>> mix,
>> > > > > > >>>> nor do we want to try to make Kafka's RPC mechanism general
>> > > > purpose.
>> > > > > > >>>> 2. Likewise, it should be clear that the API you want to
>> > expose
>> > > to
>> > > > > the
>> > > > > > >>>> front-end/client service is not necessarily the API you'd
>> need
>> > > > > > >> internally
>> > > > > > >>>> as there may be additional filtering/processing in the
>> router.
>> > > > > > >>>>
>> > > > > > >>>> Given these constraints, what we prefer to add is a fairly
>> > > > low-level
>> > > > > > >>>> "toolbox" that would let you do anything you want, but
>> > requires
>> > > to
>> > > > > > >> route
>> > > > > > >>>> and perform any aggregation or processing yourself. This
>> > pattern
>> > > > is
>> > > > > > >>>> not recommended for all kinds of services/apps, but there
>> are
>> > > > > > >> definitely
>> > > > > > >>> a
>> > > > > > >>>> category of things where it is a big win and other advanced
>> > > > > > >> applications
>> > > > > > >>>> are out-of-scope.
>> > > > > > >>>>
>> > > > > > >>>> The APIs we expose should take the following things into
>> > > > > > consideration:
>> > > > > > >>>> 1. Make it clear to the user that they will do the routing,
>> > > > > > >> aggregation,
>> > > > > > >>>> processing themselves. So the bare minimum that we want to
>> > > expose
>> > > > is
>> > > > > > >>> store
>> > > > > > >>>> and partition metadata per application server identified by
>> > the
>> > > > host
>> > > > > > >> and
>> > > > > > >>>> port.
>> > > > > > >>>> 2. Ensure that the API exposes abstractions that are known
>> to
>> > > the
>> > > > > user
>> > > > > > >> or
>> > > > > > >>>> are intuitive to the user.
>> > > > > > >>>> 3. Avoid exposing internal objects or implementation
>> details
>> > to
>> > > > the
>> > > > > > >> user.
>> > > > > > >>>>
>> > > > > > >>>> So tying all this into answering the question of what we
>> > should
>> > > > > expose
>> > > > > > >>>> through the APIs -
>> > > > > > >>>>
>> > > > > > >>>> In Kafka Streams, the user is aware of the concept of tasks
>> > and
>> > > > > > >>> partitions
>> > > > > > >>>> since the application scales with the number of partitions
>> and
>> > > > tasks
>> > > > > > >> are
>> > > > > > >>>> the construct for logical parallelism. The user is also
>> aware
>> > of
>> > > > the
>> > > > > > >>>> concept of state stores though until now they were not user
>> > > > > > accessible.
>> > > > > > >>>> With Queryable State, the bare minimum abstractions that we
>> > need
>> > > > to
>> > > > > > >>> expose
>> > > > > > >>>> are state stores and the location of state store
>> partitions.
>> > > > > > >>>>
>> > > > > > >>>> For exposing the state stores, the getStore() APIs look
>> good
>> > > but I
>> > > > > > >> think
>> > > > > > >>>> for locating the state store partitions, we should go back
>> to
>> > > the
>> > > > > > >>> original
>> > > > > > >>>> proposal of simply exposing some sort of
>> > getPartitionMetadata()
>> > > > that
>> > > > > > >>>> returns a PartitionMetadata or TaskMetadata object keyed by
>> > > > > HostInfo.
>> > > > > > >>>>
>> > > > > > >>>> The application will convert the HostInfo (host and port)
>> into
>> > > > some
>> > > > > > >>>> connection URL to talk to the other app instances via its
>> own
>> > > RPC
>> > > > > > >>> mechanism
>> > > > > > >>>> depending on whether it needs to scatter-gather or just
>> query.
>> > > The
>> > > > > > >>>> application will know how a key maps to a partition and
>> > through
>> > > > > > >>>> PartitionMetadata it will know how to locate the server
>> that
>> > > hosts
>> > > > > the
>> > > > > > >>>> store that has the partition hosting that key.
>> > > > > > >>>>
>> > > > > > >>>> On Fri, Jul 8, 2016 at 9:40 AM, Michael Noll <
>> > > > mich...@confluent.io>
>> > > > > > >>> wrote:
>> > > > > > >>>>
>> > > > > > >>>>> Addendum in case my previous email wasn't clear:
>> > > > > > >>>>>
>> > > > > > >>>>>> So for any given instance of a streams application there
>> > will
>> > > > > never
>> > > > > > >>> be
>> > > > > > >>>>> both a v1 and v2 alive at the same time
>> > > > > > >>>>>
>> > > > > > >>>>> That's right. But the current live instance will be able
>> to
>> > > tell
>> > > > > > >> other
>> > > > > > >>>>> instances, via its endpoint setting, whether it wants to
>> be
>> > > > > contacted
>> > > > > > >>> at
>> > > > > > >>>> v1
>> > > > > > >>>>> or at v2. The other instances can't guess that. Think:
>> if
>> > an
>> > > > > older
>> > > > > > >>>>> instance would manually compose the "rest" of an endpoint
>> > URI,
>> > > > > having
>> > > > > > >>>> only
>> > > > > > >>>>> the host and port from the endpoint setting, it might not
>> > know
>> > > > that
>> > > > > > >> the
>> > > > > > >>>> new
>> > > > > > >>>>> instances have a different endpoint suffix, for example).
>> > > > > > >>>>>
>> > > > > > >>>>>
>> > > > > > >>>>> On Fri, Jul 8, 2016 at 6:37 PM, Michael Noll <
>> > > > mich...@confluent.io
>> > > > > >
>> > > > > > >>>> wrote:
>> > > > > > >>>>>
>> > > > > > >>>>>> Damian,
>> > > > > > >>>>>>
>> > > > > > >>>>>> about the rolling upgrade comment: An instance A will
>> > contact
>> > > > > > >>> another
>> > > > > > >>>>>> instance B by the latter's endpoint, right? So if A has
>> no
>> > > > > further
>> > > > > > >>>>>> information available than B's host and port, then how
>> > should
>> > > > > > >>> instance
>> > > > > > >>>> A
>> > > > > > >>>>>> know whether it should call B at /v1/ or at /v2/? I
>> agree
>> > > that
>> > > > my
>> > > > > > >>>>>> suggestion isn't foolproof, but it is afaict better than
>> the
>> > > > > > >>> host:port
>> > > > > > >>>>>> approach.
>> > > > > > >>>>>>
>> > > > > > >>>>>>
>> > > > > > >>>>>>
>> > > > > > >>>>>> On Fri, Jul 8, 2016 at 5:15 PM, Damian Guy <
>> > > > damian....@gmail.com>
>> > > > > > >>>> wrote:
>> > > > > > >>>>>>
>> > > > > > >>>>>>> Michael - i'm ok with changing it to a string. Any one
>> else
>> > > > have
>> > > > > a
>> > > > > > >>>>> strong
>> > > > > > >>>>>>> opinion on this?
>> > > > > > >>>>>>>
>> > > > > > >>>>>>> FWIW - i don't think it will work fine as is during the
>> > > rolling
>> > > > > > >>>> upgrade
>> > > > > > >>>>>>> scenario as the service that is listening on the port
>> needs
>> > > to
>> > > > be
>> > > > > > >>>>> embedded
>> > > > > > >>>>>>> within each instance. So for any given instance of a
>> > streams
>> > > > > > >>>> application
>> > > > > > >>>>>>> there will never be both a v1 and v2 alive at the same
>> time
>> > > > > > >> (unless
>> > > > > > >>> of
>> > > > > > >>>>>>> course the process didn't shutdown properly, but then
>> you
>> > > have
>> > > > > > >>> another
>> > > > > > >>>>>>> problem...).
>> > > > > > >>>>>>>
>> > > > > > >>>>>>> On Fri, 8 Jul 2016 at 15:26 Michael Noll <
>> > > mich...@confluent.io
>> > > > >
>> > > > > > >>>> wrote:
>> > > > > > >>>>>>>
>> > > > > > >>>>>>>> I have one further comment about
>> > > > > > >>>> `StreamsConfig.USER_ENDPOINT_CONFIG`.
>> > > > > > >>>>>>>>
>> > > > > > >>>>>>>> I think we should consider to not restricting the
>> value of
>> > > > this
>> > > > > > >>>>> setting
>> > > > > > >>>>>>> to
>> > > > > > >>>>>>>> only `host:port` pairs. By design, this setting is
>> > > capturing
>> > > > > > >>>>>>> user-driven
>> > > > > > >>>>>>>> metadata to define an endpoint, so why restrict the
>> > > creativity
>> > > > > > >> or
>> > > > > > >>>>>>>> flexibility of our users? I can imagine, for example,
>> > that
>> > > > > > >> users
>> > > > > > >>>>> would
>> > > > > > >>>>>>>> like to set values such as `https://host:port
>> > /api/rest/v1/`
>> > > > in
>> > > > > > >>> this
>> > > > > > >>>>>>> field
>> > > > > > >>>>>>>> (e.g. being able to distinguish between `.../v1/` and
>> > > > `.../v2/`
>> > > > > > >>> may
>> > > > > > >>>>>>> help in
>> > > > > > >>>>>>>> scenarios such as rolling upgrades, where, during the
>> > > upgrade,
>> > > > > > >>> older
>> > > > > > >>>>>>>> instances may need to coexist with newer instances).
>> > > > > > >>>>>>>>
>> > > > > > >>>>>>>> That said, I don't have a strong opinion here.
>> > > > > > >>>>>>>>
>> > > > > > >>>>>>>> -Michael
>> > > > > > >>>>>>>>
>> > > > > > >>>>>>>>
>> > > > > > >>>>>>>>
>> > > > > > >>>>>>>> On Fri, Jul 8, 2016 at 2:55 PM, Matthias J. Sax <
>> > > > > > >>>>> matth...@confluent.io>
>> > > > > > >>>>>>>> wrote:
>> > > > > > >>>>>>>>
>> > > > > > >>>>>>>>> +1
>> > > > > > >>>>>>>>>
>> > > > > > >>>>>>>>> On 07/08/2016 11:03 AM, Eno Thereska wrote:
>> > > > > > >>>>>>>>>> +1 (non-binding)
>> > > > > > >>>>>>>>>>
>> > > > > > >>>>>>>>>>> On 7 Jul 2016, at 18:31, Sriram Subramanian <
>> > > > > > >>> r...@confluent.io>
>> > > > > > >>>>>>> wrote:
>> > > > > > >>>>>>>>>>>
>> > > > > > >>>>>>>>>>> +1
>> > > > > > >>>>>>>>>>>
>> > > > > > >>>>>>>>>>> On Thu, Jul 7, 2016 at 9:53 AM, Henry Cai
>> > > > > > >>>>>>> <h...@pinterest.com.invalid
>> > > > > > >>>>>>>>>
>> > > > > > >>>>>>>>>>> wrote:
>> > > > > > >>>>>>>>>>>
>> > > > > > >>>>>>>>>>>> +1
>> > > > > > >>>>>>>>>>>>
>> > > > > > >>>>>>>>>>>> On Thu, Jul 7, 2016 at 6:48 AM, Michael Noll <
>> > > > > > >>>>>>> mich...@confluent.io>
>> > > > > > >>>>>>>>> wrote:
>> > > > > > >>>>>>>>>>>>
>> > > > > > >>>>>>>>>>>>> +1 (non-binding)
>> > > > > > >>>>>>>>>>>>>
>> > > > > > >>>>>>>>>>>>> On Thu, Jul 7, 2016 at 10:24 AM, Damian Guy <
>> > > > > > >>>>>>> damian....@gmail.com>
>> > > > > > >>>>>>>>>>>> wrote:
>> > > > > > >>>>>>>>>>>>>
>> > > > > > >>>>>>>>>>>>>> Thanks Henry - we've updated the KIP with an
>> example
>> > > and
>> > > > > > >>> the
>> > > > > > >>>>> new
>> > > > > > >>>>>>>>> config
>> > > > > > >>>>>>>>>>>>>> parameter required. FWIW the user doesn't
>> register a
>> > > > > > >>>> listener,
>> > > > > > >>>>>>> they
>> > > > > > >>>>>>>>>>>>> provide
>> > > > > > >>>>>>>>>>>>>> a host:port in config. It is expected they will
>> > start
>> > > a
>> > > > > > >>>>> service
>> > > > > > >>>>>>>>> running
>> > > > > > >>>>>>>>>>>>> on
>> > > > > > >>>>>>>>>>>>>> that host:port that they can use to connect to
>> the
>> > > > > > >> running
>> > > > > > >>>>>>>>> KafkaStreams
>> > > > > > >>>>>>>>>>>>>> Instance.
>> > > > > > >>>>>>>>>>>>>>
>> > > > > > >>>>>>>>>>>>>> Thanks,
>> > > > > > >>>>>>>>>>>>>> Damian
>> > > > > > >>>>>>>>>>>>>>
>> > > > > > >>>>>>>>>>>>>> On Thu, 7 Jul 2016 at 06:06 Henry Cai
>> > > > > > >>>>>>> <h...@pinterest.com.invalid>
>> > > > > > >>>>>>>>>>>>> wrote:
>> > > > > > >>>>>>>>>>>>>>
>> > > > > > >>>>>>>>>>>>>>> It wasn't quite clear to me how the user program
>> > > > > > >>> interacts
>> > > > > > >>>>> with
>> > > > > > >>>>>>>> the
>> > > > > > >>>>>>>>>>>>>>> discovery API, especially on the user supplied
>> > > listener
>> > > > > > >>>> part,
>> > > > > > >>>>>>> how
>> > > > > > >>>>>>>>>>>> does
>> > > > > > >>>>>>>>>>>>>> the
>> > > > > > >>>>>>>>>>>>>>> user program supply that listener to
>> KafkaStreams
>> > and
>> > > > > > >> how
>> > > > > > >>>>> does
>> > > > > > >>>>>>>>>>>>>> KafkaStreams
>> > > > > > >>>>>>>>>>>>>>> know which port the user listener is running,
>> > maybe a
>> > > > > > >>> more
>> > > > > > >>>>>>>> complete
>> > > > > > >>>>>>>>>>>>>>> end-to-end example including the steps on
>> > registering
>> > > > > > >> the
>> > > > > > >>>>> user
>> > > > > > >>>>>>>>>>>> listener
>> > > > > > >>>>>>>>>>>>>> and
>> > > > > > >>>>>>>>>>>>>>> whether the user listener needs to be involved
>> with
>> > > > > > >> task
>> > > > > > >>>>>>>>>>>> reassignment.
>> > > > > > >>>>>>>>>>>>>>>
>> > > > > > >>>>>>>>>>>>>>>
>> > > > > > >>>>>>>>>>>>>>> On Wed, Jul 6, 2016 at 9:13 PM, Guozhang Wang <
>> > > > > > >>>>>>> wangg...@gmail.com
>> > > > > > >>>>>>>>>
>> > > > > > >>>>>>>>>>>>>> wrote:
>> > > > > > >>>>>>>>>>>>>>>
>> > > > > > >>>>>>>>>>>>>>>> ＋1
>> > > > > > >>>>>>>>>>>>>>>>
>> > > > > > >>>>>>>>>>>>>>>> On Wed, Jul 6, 2016 at 12:44 PM, Damian Guy <
>> > > > > > >>>>>>>> damian....@gmail.com>
>> > > > > > >>>>>>>>>>>>>>> wrote:
>> > > > > > >>>>>>>>>>>>>>>>
>> > > > > > >>>>>>>>>>>>>>>>> Hi all,
>> > > > > > >>>>>>>>>>>>>>>>>
>> > > > > > >>>>>>>>>>>>>>>>> I'd like to initiate the voting process for
>> > KIP-67
>> > > > > > >>>>>>>>>>>>>>>>> <
>> > > > > > >>>>>>>>>>>>>>>>>
>> > > > > > >>>>>>>>>>>>>>>>
>> > > > > > >>>>>>>>>>>>>>>
>> > > > > > >>>>>>>>>>>>>>
>> > > > > > >>>>>>>>>>>>>
>> > > > > > >>>>>>>>>>>>
>> > > > > > >>>>>>>>>
>> > > > > > >>>>>>>>
>> > > > > > >>>>>>>
>> > > > > > >>>>>
>> > > > > > >>>>
>> > > > > > >>>
>> > > > > > >>
>> > > > > >
>> > > > >
>> > > >
>> > >
>> >
>> https://cwiki.apache.org/confluence/display/KAFKA/KIP-67%3A+Queryable+state+for+Kafka+Streams
>> > > > > > >>>>>>>>>>>>>>>>>>
>> > > > > > >>>>>>>>>>>>>>>>>
>> > > > > > >>>>>>>>>>>>>>>>> KAFKA-3909 <
>> > > > > > >>>>> https://issues.apache.org/jira/browse/KAFKA-3909
>> > > > > > >>>>>>>>
>> > > > > > >>>>>>>> is
>> > > > > > >>>>>>>>>>>>> the
>> > > > > > >>>>>>>>>>>>>>> top
>> > > > > > >>>>>>>>>>>>>>>>> level JIRA for this effort.
>> > > > > > >>>>>>>>>>>>>>>>>
>> > > > > > >>>>>>>>>>>>>>>>> Initial PRs for Step 1 of the process are:
>> > > > > > >>>>>>>>>>>>>>>>> Expose State Store Names <
>> > > > > > >>>>>>>>>>>>> https://github.com/apache/kafka/pull/1526>
>> > > > > > >>>>>>>>>>>>>>> and
>> > > > > > >>>>>>>>>>>>>>>>> Query Local State Stores <
>> > > > > > >>>>>>>>>>>>> https://github.com/apache/kafka/pull/1565>
>> > > > > > >>>>>>>>>>>>>>>>>
>> > > > > > >>>>>>>>>>>>>>>>> Thanks,
>> > > > > > >>>>>>>>>>>>>>>>> Damian
>> > > > > > >>>>>>>>>>>>>>>>>
>> > > > > > >>>>>>>>>>>>>>>>
>> > > > > > >>>>>>>>>>>>>>>>
>> > > > > > >>>>>>>>>>>>>>>>
>> > > > > > >>>>>>>>>>>>>>>> --
>> > > > > > >>>>>>>>>>>>>>>> -- Guozhang
>> > > > > > >>>>>>>>>>>>>>>>
>> > > > > > >>>>>>>>>>>>>>>
>> > > > > > >>>>>>>>>>>>>>
>> > > > > > >>>>>>>>>>>>>
>> > > > > > >>>>>>>>>>>>>
>> > > > > > >>>>>>>>>>>>>
>> > > > > > >>>>>>>>>>>>> --
>> > > > > > >>>>>>>>>>>>> Best regards,
>> > > > > > >>>>>>>>>>>>> Michael Noll
>> > > > > > >>>>>>>>>>>>>
>> > > > > > >>>>>>>>>>>>>
>> > > > > > >>>>>>>>>>>>>
>> > > > > > >>>>>>>>>>>>> *Michael G. Noll | Product Manager | Confluent |
>> +1
>> > > > > > >>>>>>>>> 650.453.5860Download
>> > > > > > >>>>>>>>>>>>> Apache Kafka and Confluent Platform:
>> > > > > > >>>> www.confluent.io/download
>> > > > > > >>>>>>>>>>>>> <http://www.confluent.io/download>*
>> > > > > > >>>>>>>>>>>>>
>> > > > > > >>>>>>>>>>>>
>> > > > > > >>>>>>>>>>
>> > > > > > >>>>>>>>>
>> > > > > > >>>>>>>>>
>> > > > > > >>>>>>>>
>> > > > > > >>>>>>>>
>> > > > > > >>>>>>>> --
>> > > > > > >>>>>>>> Best regards,
>> > > > > > >>>>>>>> Michael Noll
>> > > > > > >>>>>>>>
>> > > > > > >>>>>>>>
>> > > > > > >>>>>>>>
>> > > > > > >>>>>>>> *Michael G. Noll | Product Manager | Confluent | +1
>> > > > > > >>>>> 650.453.5860Download
>> > > > > > >>>>>>>> Apache Kafka and Confluent Platform:
>> > > > www.confluent.io/download
>> > > > > > >>>>>>>> <http://www.confluent.io/download>*
>> > > > > > >>>>>>>>
>> > > > > > >>>>>>>
>> > > > > > >>>>>>
>> > > > > > >>>>>>
>> > > > > > >>>>>>
>> > > > > > >>>>>> --
>> > > > > > >>>>>> Best regards,
>> > > > > > >>>>>> Michael Noll
>> > > > > > >>>>>>
>> > > > > > >>>>>>
>> > > > > > >>>>>>
>> > > > > > >>>>>> *Michael G. Noll | Product Manager | Confluent | +1
>> > > > 650.453.5860
>> > > > > > >>>>>> <%2B1%20650.453.5860>Download Apache Kafka and Confluent
>> > > > Platform:
>> > > > > > >>>>>> www.confluent.io/download <
>> http://www.confluent.io/download
>> > >*
>> > > > > > >>>>>>
>> > > > > > >>>>>
>> > > > > > >>>>>
>> > > > > > >>>>>
>> > > > > > >>>>> --
>> > > > > > >>>>> Best regards,
>> > > > > > >>>>> Michael Noll
>> > > > > > >>>>>
>> > > > > > >>>>>
>> > > > > > >>>>>
>> > > > > > >>>>> *Michael G. Noll | Product Manager | Confluent | +1
>> > > > > > >>> 650.453.5860Download
>> > > > > > >>>>> Apache Kafka and Confluent Platform:
>> > www.confluent.io/download
>> > > > > > >>>>> <http://www.confluent.io/download>*
>> > > > > > >>>>>
>> > > > > > >>>>
>> > > > > > >>>>
>> > > > > > >>>>
>> > > > > > >>>> --
>> > > > > > >>>> Thanks,
>> > > > > > >>>> Neha
>> > > > > > >>>>
>> > > > > > >>>
>> > > > > > >>>
>> > > > > > >>>
>> > > > > > >>> --
>> > > > > > >>> -- Guozhang
>> > > > > > >>>
>> > > > > > >>
>> > > > > >
>> > > > > >
>> > > > >
>> > > > >
>> > > > > --
>> > > > > Thanks,
>> > > > > Neha
>> > > > >
>> > > >
>> > > >
>> > > >
>> > > > --
>> > > > -- Guozhang
>> > > >
>> > >
>> >
>>
>
