What about KafkaStreams#toString() method? I think, we want to deprecate it as with KIP-120 and the changes of this KIP, is gets obsolete.
If we do so, please update the KIP accordingly. -Matthias On 3/28/17 7:00 PM, Matthias J. Sax wrote: > Thanks for updating the KIP! > > I think it's good as is -- I would not add anything more to TaskMetadata. > > About subtopologies and tasks. We do have the concept of subtopologies > already in KIP-120. It's only missing and ID that allow to link a > subtopology to a task. > > IMHO, adding a simple variable to `Subtopoloy` that provide the id > should be sufficient. We can simply document in the JavaDocs how > Subtopology and TaskMetadata can be linked to each other. > > I did update KIP-120 accordingly. > > > -Matthias > > On 3/28/17 3:45 PM, Florian Hussonnois wrote: >> Hi all, >> >> I've updated the KIP and the PR to reflect your suggestions. >> https://cwiki.apache.org/confluence/display/KAFKA/KIP+130%3A+Expose+states+of+active+tasks+to+KafkaStreams+public+API >> https://github.com/apache/kafka/pull/2612 >> >> Also, I've exposed property StreamThread#state as a string through the >> new class ThreadMetadata. >> >> Thanks, >> >> 2017-03-27 23:40 GMT+02:00 Florian Hussonnois <fhussonn...@gmail.com >> <mailto:fhussonn...@gmail.com>>: >> >> Hi Guozhang, Matthias, >> >> It's a great idea to add sub topologies descriptions. This would >> help developers to better understand topology concept. >> >> I agree that is not really user-friendly to check if >> `StreamsMetadata#streamThreads` is not returning null. >> >> The method name localThreadsMetadata looks good. In addition, it's >> more simple to build ThreadMetadata instances from the `StreamTask` >> class than from `StreamPartitionAssignor` class. >> >> I will work on modifications. As I understand, I have to add the >> property subTopologyId property to the TaskMetadata class - Am I right ? >> >> Thanks, >> >> 2017-03-26 0:25 GMT+01:00 Guozhang Wang <wangg...@gmail.com >> <mailto:wangg...@gmail.com>>: >> >> Re 1): this is a good point. May be we can move >> `StreamsMetadata#streamThreads` as >> `KafkaStreams#localThreadsMetadata`? >> >> 3): this is a minor suggestion about function name of >> `assignedPartitions`, to `topicPartitions` to be consistent with >> `StreamsMetadata`? >> >> >> Guozhang >> >> On Thu, Mar 23, 2017 at 4:30 PM, Matthias J. Sax >> <matth...@confluent.io <mailto:matth...@confluent.io>> wrote: >> >> Thanks for the progress on this KIP. I think we are on the >> right path! >> >> Couple of comments/questions: >> >> (1) Why do we not consider the "rejected alternative" to add >> the method >> to KafkaStreams? The comment on #streamThreads() says: >> >> "Note this method will return <code>null</code> if called on >> {@link >> StreamsMetadata} which represent a remote application." >> >> Thus, if we cannot get any remote metadata, it seems not >> straight >> forward to not add it to KafkaStreams directly -- this would >> avoid >> invalid calls and `null` return value in the first place. >> >> I like the idea about exposing sub-topologies.: >> >> (2a) I would recommend to rename `topicsGroupId` to >> `subTopologyId` :) >> >> (2b) We could add this to KIP-120 already. However, I would >> not just >> link both via name, but leverage KIP-120 directly, and add a >> "Subtopology" member to the TaskMetadata class. >> >> >> Overall, I like the distinction of KIP-120 only exposing >> "static" >> information that can be determined before the topology get's >> started, >> while this KIP allow to access runtime information. >> >> >> >> -Matthias >> >> >> On 3/22/17 12:42 PM, Guozhang Wang wrote: >> > Thanks for the updated KIP, and sorry for the late replies! >> > >> > I think a little bit more about KIP-130, and I feel that >> if we are going >> > to deprecate the `toString` function (it is not explicitly >> said in the >> > KIP, so I'm not sure if you plan to still keep the >> > `KafkaStreams#toString` as is or are going to replace it >> with the >> > proposed APIs) with the proposed ones, it may be okay. More >> > specifically, after both KIP-120 and KIP-130: >> > >> > 1. users can use `#describe` function to check the >> generated topology >> > before calling `KafkaStreams#start`, which is static >> information. >> > 2. users can use the `StreamsMetadata -> ThreadMetadata -> >> TaskMetadata` >> > programmatically after called `KafkaStreams#start` to get the >> > dynamically changeable information. >> > >> > One thing I'm still not sure though, is that in >> `TaskMetadata` we only >> > have the TaskId and assigned partitions, whereas in >> > "TopologyDescription" introduced in KIP-120, it will >> simply describe the >> > whole topology possibly composed of multiple >> sub-topologies. So it is >> > hard for users to tell which sub-topology is executed >> under which task >> > on-the-fly. >> > >> > Hence I'm thinking if we can expose the "sub-topology-id" >> (named as >> > topicsGroupId internally) in >> TopologyDescription#Subtopology, and then >> > from the task id which is essentially "sub-topology-id DASH >> > partition-group-id" users can make the link, though it is >> still not that >> > straight-forward. >> > >> > Thoughts? >> > >> > Guozhang >> > >> > >> > >> > On Wed, Mar 15, 2017 at 3:16 PM, Florian Hussonnois >> > <fhussonn...@gmail.com <mailto:fhussonn...@gmail.com> >> <mailto:fhussonn...@gmail.com >> <mailto:fhussonn...@gmail.com>>> wrote: >> > >> > Thanks Guozhang for pointing me to the KIP-120. >> > >> > I've made some modifications to the KIP. I also proposed a >> new PR >> > (there is >> > still some tests to make). >> > >> https://cwiki.apache.org/confluence/display/KAFKA/KIP+130%3A+Expose+states+of+active+tasks+to+KafkaStreams+public+API >> >> <https://cwiki.apache.org/confluence/display/KAFKA/KIP+130%3A+Expose+states+of+active+tasks+to+KafkaStreams+public+API> >> > >> <https://cwiki.apache.org/confluence/display/KAFKA/KIP+130%3A+Expose+states+of+active+tasks+to+KafkaStreams+public+API >> >> <https://cwiki.apache.org/confluence/display/KAFKA/KIP+130%3A+Expose+states+of+active+tasks+to+KafkaStreams+public+API>> >> > >> > Exposing consumed offsets through JMX is sufficient for >> debugging >> > purpose. >> > But I think this could be part to another JIRA as there is >> no impact to >> > public API. >> > >> > Thanks >> > >> > 2017-03-10 22:35 GMT+01:00 Guozhang Wang >> <wangg...@gmail.com <mailto:wangg...@gmail.com> >> > <mailto:wangg...@gmail.com <mailto:wangg...@gmail.com>>>: >> >> > >> > > Hello Florian, >> > > >> > > As for programmatically discover monitoring data by >> piping metrics >> > into a >> > > dedicated topic. I think you can actually use a >> > KafkaMetricsReporter which >> > > pipes the polled metric values into a pre-defined >> topic (note that >> > in Kafka >> > > the MetricsReporter is simply an interface and users >> can build >> > their own >> > > impl in addition to the JMXReporter), for example : >> > > >> > > https://github.com/krux/kafka-metrics-reporter >> <https://github.com/krux/kafka-metrics-reporter> >> > <https://github.com/krux/kafka-metrics-reporter >> <https://github.com/krux/kafka-metrics-reporter>> >> > > >> > > As for the "static task-level assignment", what I >> meant is that >> > the mapping >> > > from source-topic-partitions -> tasks are static, >> via the >> > > "PartitionGrouper", and a task won't switch from an >> active task to a >> > > standby task, it is actually that an active task >> could be >> > migrated, as a >> > > whole along with all its assigned partitions, to >> another thread / >> > process >> > > and a new standby task will be created on the host >> that this >> > active task is >> > > migrating from. So for the SAME task, its taskMetadata. >> > > assignedPartitions() >> > > will always return you the same partitions. >> > > >> > > As for the `toString` function that what we have >> today, I feel it >> > has some >> > > correlations with KIP-120 so I'm trying to >> coordinate some >> > discussions here >> > > (cc'ing Matthias as the owner of KIP-120). My >> understand is that: >> > > >> > > 1. In KIP-120, the `toString` function of >> `KafkaStreams` will be >> > removed >> > > and instead the `Topology#describe` function will be >> introduced >> > for users >> > > to debug the topology BEFORE start running their >> instance with the >> > > topology. And hence the description won't contain >> any task >> > information as >> > > they are not formed yet. >> > > 2. In KIP-130, we want to add the task-level >> information for >> > monitoring >> > > purposes, which is not static and can only be >> captured AFTER the >> > instance >> > > has started running. Again I'm wondering for KIP-130 >> alone if >> > adding those >> > > metrics mentioned in my previous email would suffice >> even for the >> > use case >> > > that you have mentioned. >> > > >> > > >> > > Guozhang >> > > >> > > On Wed, Mar 8, 2017 at 3:18 PM, Florian Hussonnois >> > <fhussonn...@gmail.com <mailto:fhussonn...@gmail.com> >> <mailto:fhussonn...@gmail.com <mailto:fhussonn...@gmail.com>>> >> >> > > wrote: >> > > >> > > > Hi Guozhang >> > > > >> > > > Thank you for your feedback. I've started to look >> more deeply >> > into the >> > > > code. As you mention, it would be more clever to >> use the current >> > > > StreamMetadata API to expose these information. >> > > > >> > > > I think exposing metrics through JMX is great for >> building >> > monitoring >> > > > dashboards using some tools like jmxtrans and grafana. >> > > > But for our use case we would like to expose the >> states >> > directely from >> > > the >> > > > application embedding the kstreams topologies. >> > > > So we expect to be able to retrieve states in a >> programmatic way. >> > > > >> > > > For instance, we could imagin to produce those >> states into a >> > dedicated >> > > > topic. In that way a third application could >> automatically >> > discover all >> > > > kafka-streams applications which could be monitored. >> > > > In production environment, that can be clearly a >> solution to have a >> > > > complete overview of a microservices architecture >> based on Kafka >> > Streams. >> > > > >> > > > The toString() method give a lots of information >> it can only be >> > used for >> > > > debugging purpose but not to build a topologies >> visualization >> > tool. We >> > > > could actually expose same details about the >> stream topology >> > from the >> > > > StreamMetadata API ? So the TaskMetadata class you >> have >> > suggested could >> > > > contains similar information that ones return by >> the toString >> > method from >> > > > AbstractTask class ? >> > > > >> > > > I can update the KIP in that way. >> > > > >> > > > Finally, I'm not sure to understand your last >> point :* "Note >> > that the >> > > > task-level assignment information is static, i.e. >> it will not change >> > > during >> > > > the runtime" * >> > > > >> > > > Does that mean when a rebalance occurs new tasks >> are created for >> > the new >> > > > assignments and old ones just switch to a standby >> state ? >> > > > >> > > > Thanks, >> > > > >> > > > 2017-03-05 7:04 GMT+01:00 Guozhang Wang >> <wangg...@gmail.com <mailto:wangg...@gmail.com> >> > <mailto:wangg...@gmail.com <mailto:wangg...@gmail.com>>>: >> >> > > > >> > > > > Hello Florian, >> > > > > >> > > > > Thanks for the KIP and your detailed explanation >> of your use >> > case. I >> > > > think >> > > > > there are two dimensions to discuss on how to >> improve Streams' >> > > > > debuggability (or more specifically state >> exposure for >> > visualization). >> > > > > >> > > > > First question is "what information should we >> expose to the >> > user". From >> > > > > your KIP I saw generally three categories: >> > > > > >> > > > > 1. The state of the thread within a process, as >> you mentioned >> > currently >> > > > we >> > > > > only expose the state of the process but not the >> finer grained >> > > per-thread >> > > > > state. >> > > > > 2. The state of the task. Currently the most >> close API to this is >> > > > > StreamsMetadata, >> > > > > however it aggregates the tasks across all >> threads and only >> > present the >> > > > > aggregated set of the assigned partitions / >> state stores etc. >> > We can >> > > > > consider extending this method to have a new >> > StreamsMetadata#tasks() >> > > > which >> > > > > returns a TaskMetadata with the similar fields, >> and the >> > > > > StreamsMetadata.stateStoreNames / etc would >> still be returning the >> > > > > aggregated results but users can still "drill >> down" if they want. >> > > > > >> > > > > The second question is "how should we expose >> them to the >> > user". For >> > > > > example, you mentioned about >> consumedOffsetsByPartition in the >> > > > activeTasks. >> > > > > We could add this as a JMX metric based on fetch >> positions >> > inside the >> > > > > consumer layer (note that Streams is just >> embedding consumers) >> > or we >> > > > could >> > > > > consider adding it into TaskMetadata. Either >> case it can be >> > visualized >> > > > for >> > > > > monitoring. The reason we expose StreamsMetadata >> as well as >> > State was >> > > > that >> > > > > it is expected to be "polled" in a programmatic >> way for >> > interactive >> > > > queries >> > > > > and also for control flows (e.g. I would like to >> ONLY start >> > running my >> > > > > other topology until the first topology has been >> up and >> > running) while >> > > > for >> > > > > your case it seems the main purpose is to >> continuously query >> > them for >> > > > > monitoring etc. Personally I'd prefer to expose >> them as JMX >> > only for >> > > such >> > > > > purposes only to have a simpler API. >> > > > > >> > > > > So given your current motivations I'd suggest >> expose the following >> > > > > information as newly added metrics in Streams: >> > > > > >> > > > > 1. Thread-level state metric. >> > > > > 2. Task-level hosted client identifier metric >> (e.g. host:port). >> > > > > 3. Consumer-level per-topic/partition position >> metric ( >> > > > > >> https://kafka.apache.org/documentation/#topic_fetch_monitoring >> <https://kafka.apache.org/documentation/#topic_fetch_monitoring> >> > >> <https://kafka.apache.org/documentation/#topic_fetch_monitoring >> >> <https://kafka.apache.org/documentation/#topic_fetch_monitoring>>). >> > > > > >> > > > > Note that the task-level assignment information is >> static, >> > i.e. it will >> > > > not >> > > > > change during the runtime at all and can be accessed >> from the >> > > > `toString()` >> > > > > function already even before the instance start >> running, so I >> > think >> > > this >> > > > > piece of information do not need to be exposed >> through JMX >> > anymore. >> > > > > >> > > > > WDYT? >> > > > > >> > > > > Guozhang >> > > > > >> > > > > >> > > > > On Thu, Mar 2, 2017 at 3:11 AM, Damian Guy >> > <damian....@gmail.com <mailto:damian....@gmail.com> >> <mailto:damian....@gmail.com <mailto:damian....@gmail.com>>> >> > > wrote: >> > > > > >> > > > > > Hi Florian, >> > > > > > >> > > > > > Thanks for the KIP. >> > > > > > >> > > > > > It seems there is some overlap here with what we >> already have in >> > > > > > KafkaStreams.allMetadata(). This currently returns >> a >> > > > > > Collection<StreamsMetadata> where each >> StreamsMetadata >> > instance holds >> > > > the >> > > > > > state stores and partition assignment for every >> instance of the >> > > > > > KafkaStreams application. I'm wondering if that is >> good >> > enough for >> > > what >> > > > > you >> > > > > > are trying to achieve? If not could it be modified >> to >> > include the per >> > > > > > Thread assignment? >> > > > > > >> > > > > > Thanks, >> > > > > > Damian >> > > > > > >> > > > > > >> > > > > > >> > > > > > >> > > > > > >> > > > > > >> > > > > > On Wed, 1 Mar 2017 at 22:49 Florian Hussonnois < >> > > fhussonn...@gmail.com <mailto:fhussonn...@gmail.com> >> <mailto:fhussonn...@gmail.com <mailto:fhussonn...@gmail.com>>> >> >> > > > > > wrote: >> > > > > > >> > > > > > > Hi Matthias, >> > > > > > > >> > > > > > > First, I will answer to your last question. >> > > > > > > >> > > > > > > The main reason to have both >> TaskState#assignment and >> > > > > > > TaskState#consumedOffsetsByPartition is that >> tasks have no >> > > consumed >> > > > > > offsets >> > > > > > > until at least one message is consumed for >> each partition >> > even if >> > > > > > previous >> > > > > > > offsets exist for the consumer group. >> > > > > > > So yes this methods are redundant as it only >> diverge at >> > application >> > > > > > > startup. >> > > > > > > >> > > > > > > About the use case, currently we are >> developping for a >> > customer a >> > > > > little >> > > > > > > framework based on KafkaStreams which >> > transform/denormalize data >> > > > before >> > > > > > > ingesting into hadoop. >> > > > > > > >> > > > > > > We have a cluster of workers (SpringBoot) >> which instantiate >> > > KStreams >> > > > > > > topologies dynamicaly based on dataflow >> configurations. >> > > > > > > Each configuration describes a topic to >> consume and how to >> > process >> > > > > > messages >> > > > > > > (this looks like NiFi processors API). >> > > > > > > >> > > > > > > Our architecture is inspired from >> KafkaConnect. We have >> > topics for >> > > > > > configs >> > > > > > > and states which are consumed by each >> workers (actually we >> > have >> > > > reused >> > > > > > some >> > > > > > > internals classes to the connect API). >> > > > > > > >> > > > > > > Now, we would like to develop UIs to >> visualize topics and >> > > partitions >> > > > > > > consumed by our worker applications. >> > > > > > > >> > > > > > > Also, I think it would be nice to be able, >> in the futur, to >> > > develop >> > > > > web >> > > > > > > UIs similar to Spark but for KafkaStreams to >> visualize >> > DAGs...so >> > > > maybe >> > > > > > this >> > > > > > > KIP is just a first step. >> > > > > > > >> > > > > > > Thanks, >> > > > > > > >> > > > > > > 2017-03-01 22:52 GMT+01:00 Matthias J. Sax >> > <matth...@confluent.io <mailto:matth...@confluent.io> >> <mailto:matth...@confluent.io <mailto:matth...@confluent.io>> >> >> > > >: >> > > > > > > >> > > > > > > > Thanks for the KIP. >> > > > > > > > >> > > > > > > > I am wondering a little bit, why you need >> to expose this >> > > > information. >> > > > > > > > Can you describe some use cases? >> > > > > > > > >> > > > > > > > Would it be worth to unify this new API with >> > KafkaStreams#state() >> > > > to >> > > > > > get >> > > > > > > > the overall state of an application >> without the need to >> > call two >> > > > > > > > different methods? Not sure how this >> unified API might >> > look like >> > > > > > though. >> > > > > > > > >> > > > > > > > >> > > > > > > > One minor comment about the API: >> TaskState#assignment >> > seems to be >> > > > > > > > redundant. It should be the same as >> > > > > > > > TaskState#consumedOffsetsByPartition.keySet() >> > > > > > > > >> > > > > > > > Or do I miss something? >> > > > > > > > >> > > > > > > > >> > > > > > > > -Matthias >> > > > > > > > >> > > > > > > > On 3/1/17 5:19 AM, Florian Hussonnois wrote: >> > > > > > > > > Hi Eno, >> > > > > > > > > >> > > > > > > > > Yes, but the state() method only returns >> the global >> > state of >> > > the >> > > > > > > > > KafkaStream application (ie: CREATED, >> RUNNING, >> > REBALANCING, >> > > > > > > > > PENDING_SHUTDOWN, NOT_RUNNING). >> > > > > > > > > >> > > > > > > > > An alternative to this KIP would be to >> change this >> > method to >> > > > return >> > > > > > > more >> > > > > > > > > information instead of adding a new method. >> > > > > > > > > >> > > > > > > > > 2017-03-01 13:46 GMT+01:00 Eno Thereska < >> > > eno.there...@gmail.com >> <mailto:eno.there...@gmail.com> >> <mailto:eno.there...@gmail.com <mailto:eno.there...@gmail.com>> >> > > > >: >> > > > > > > > > >> > > > > > > > >> Thanks Florian, >> > > > > > > > >> >> > > > > > > > >> Have you had a chance to look at the new >> state methods in >> > > > 0.10.2, >> > > > > > > e.g., >> > > > > > > > >> KafkaStreams.state()? >> > > > > > > > >> >> > > > > > > > >> Thanks >> > > > > > > > >> Eno >> > > > > > > > >>> On 1 Mar 2017, at 11:54, Florian >> Hussonnois < >> > > > > fhussonn...@gmail.com >> <mailto:fhussonn...@gmail.com> <mailto:fhussonn...@gmail.com >> <mailto:fhussonn...@gmail.com>> >> >> > > > > > > >> > > > > > > > >> wrote: >> > > > > > > > >>> >> > > > > > > > >>> Hi all, >> > > > > > > > >>> >> > > > > > > > >>> I have just created KIP-130 to add a >> new method to the >> > > > > KafkaStreams >> > > > > > > API >> > > > > > > > >> in >> > > > > > > > >>> order to expose the states of threads >> and active tasks. >> > > > > > > > >>> >> > > > > > > > >>> >> > https://cwiki.apache.org/confluence/display/KAFKA/KIP+ >> <https://cwiki.apache.org/confluence/display/KAFKA/KIP+> >> > >> <https://cwiki.apache.org/confluence/display/KAFKA/KIP+ >> <https://cwiki.apache.org/confluence/display/KAFKA/KIP+>> >> > > > > > > > >> >> 130%3A+Expose+states+of+active+tasks+to+KafkaStreams+ >> > > public+API >> > > > > > > > >>> >> > > > > > > > >>> >> > > > > > > > >>> Thanks, >> > > > > > > > >>> >> > > > > > > > >>> -- >> > > > > > > > >>> Florian HUSSONNOIS >> > > > > > > > >> >> > > > > > > > >> >> > > > > > > > > >> > > > > > > > > >> > > > > > > > >> > > > > > > > >> > > > > > > >> > > > > > > >> > > > > > > -- >> > > > > > > Florian HUSSONNOIS >> > > > > > > >> > > > > > >> > > > > >> > > > > >> > > > > >> > > > > -- >> > > > > -- Guozhang >> > > > > >> > > > >> > > > >> > > > >> > > > -- >> > > > Florian HUSSONNOIS >> > > > >> > > >> > > >> > > >> > > -- >> > > -- Guozhang >> > > >> > >> > >> > >> > -- >> > Florian HUSSONNOIS >> > >> > >> > >> > >> > -- >> > -- Guozhang >> >> >> >> >> -- >> -- Guozhang >> >> >> >> >> -- >> Florian HUSSONNOIS >> >> >> >> >> -- >> Florian HUSSONNOIS >
signature.asc
Description: OpenPGP digital signature
