kafka streams with multiple threads and state store

Hi All, I want to use one state store in all my kafka stream threads in my application, how can i do it. 1. i created one topic (name: test2) with 3 partitions . 2. wrote kafka stream with num.stream.threads = 3 in java code 3. using state store (name: count2) in my application. But state store

Kafka 0.9.0.1 partitions shrink and expand frequently after restart the broker

Hi, we have a kafka cluster which is made of 6 brokers, with 8 cpu and 16G memory on each broker’s machine, and we have about 1600 topics in the cluster，about 1700 partitions’ leader and 1600 partitions' replica on each broker. when we restart a normal broke, we find that there are 500

Re: Kafka 0.9.0.1 partitions shrink and expand frequently after restart the broker

I've seen this before and it was due to long GC pauses due in large part to a memory heap > 8 GB. --John On Thu, Nov 9, 2017 at 8:17 AM, Json Tu wrote: > Hi, > we have a kafka cluster which is made of 6 brokers, with 8 cpu and > 16G memory on each broker’s machine, and we have about 1600 t

Re: Kafka 0.9.0.1 partitions shrink and expand frequently after restart the broker

Hi Json. John might have a point. It is not reasonable to have more than 6-8GB of heap provided for the JVM that's running Kafka. One of the reason is GC time and the other is that Kafka relies heavily on the OS' disk read/write in-memory caching. Also there were a few synchronization bugs in 0.9

Re: Kafka 0.9.0.1 partitions shrink and expand frequently after restart the broker

Excellent points Viktor! Also, the reason I mistakenly went > 8 GB memory heap was due to OOM errors that were being thrown when I upgraded from 0.9.0.1 to 0.10.0.0 and forgot to explicitly set the message format to 0.9.0.1 because we needed to support the older clients and the corresponding format

Re: Kafka 0.9.0.1 partitions shrink and expand frequently after restart the broker

I'm happy that it's solved :) On Thu, Nov 9, 2017 at 3:32 PM, John Yost wrote: > Excellent points Viktor! Also, the reason I mistakenly went > 8 GB memory > heap was due to OOM errors that were being thrown when I upgraded from > 0.9.0.1 to 0.10.0.0 and forgot to explicitly set the message forma

Re: Kafka 0.9.0.1 partitions shrink and expand frequently after restart the broker

Yep, the team here, including Ismael, pointed me in the right direction, which was much appreciated. :) On Thu, Nov 9, 2017 at 10:02 AM, Viktor Somogyi wrote: > I'm happy that it's solved :) > > On Thu, Nov 9, 2017 at 3:32 PM, John Yost wrote: > > > Excellent points Viktor! Also, the reason I m

Re: Kafka Monitoring..

JMX was pretty easy to setup for us. Look up the various jmx beans locally for your particular version of kafka (i.e. with jconsole..etc) https://github.com/jmxtrans/jmxtrans On 11/8/17, 7:10 PM, "chidigam ." wrote: Hi All, What is the simplest way of monitoring the metrics in kaka br

Documentation or reference on listener names/types?

I've been working with Kafka broker listeners and I'm curious is there any documentation that explains what all of them apply to? Such as CLIENT, PLAINTEXT, SASL/SSL, etc. I see the encryption part of the documentation, but is it just inferred what these listeners apply to? Thank you in advanc

Re: Kafka Monitoring..

We’ve recently started using Prometheus, and use Prometheus JMX Exporter to get Kafka metrics into prometheus. Here’s our JMX Exporter config: https://github.com/wikimedia/puppet/blob/production/modules/profile/files/kafka/broker_prometheus_jmx_exporter

Re: Kafka Monitoring..

Yeah…jmx is really the cheap/easy way to monitor kafka. You should also monitor OS metrics like pageScanD and pageScanK. Those were very helpful for us. -David On 11/9/17, 11:40 AM, "Andrew Otto" wrote: We’ve recently started using Prometheus, and use Prometheus JMX Exporter

Kafka Streams - Custom processor "init" method called before state store has data restored into it

I have a custom processor that implements AbstractProcessor and a custom store that implements StateStore. Before Kafka 1.0.0 the processors "init" method gets called after the state store is restored from changelog and that is good. With Kafka 1.0.0 the processors "init" method is called BEFORE

Re: Kafka Streams - Custom processor "init" method called before state store has data restored into it

Hi Rainer, Thanks for reporting this issue. Do you have any log data you can share? In the meantime, I'll look into the issue. Thanks, Bill On Thu, Nov 9, 2017 at 1:23 PM, Rainer Guessner wrote: > I have a custom processor that implements AbstractProcessor and a custom > store that implement

Re: Kafka Streams - Custom processor "init" method called before state store has data restored into it

I have a few logs below. Thank you. Rainer 14:21:04,304 INFO [StreamThread] stream-thread [t10_nr3_metadatarecovery-17935bc8-6ec2-4fcc-b62f-15a63c9a051c-StreamThread-1] Creating restore consumer client 14:21:04,393 INFO [StreamThread] stream-thread [t10_nr3_metadatarecovery-17935bc8-6ec2-4fc

Re: Manual offset control and topic compaction

I should probably have been a little more clear what I'm asking, so here's an example. Let's say that I have a consumer on a topic partition, and I'm doing manual commits. Because I'm doing manual commits and the messages are processed asynchronously from the consumer poll loop, I keep track of wh

Migrating from log4j 1 to log4j 2

Hi All We have a requirement to migrate log4J 1.x to log4j 2 for our kafka brokers using log4j bridge utility. According to Apache Docs the code must not call DOMConfigurator or PropertyConfigurator class, But when I dig into the code I see on Tools package and other packages they have used Prop

Re: Migrating from log4j 1 to log4j 2

PropertyConfigurator is used here: https://github.com/apache/kafka/blob/trunk/tools/src/main/java/org/apache/kafka/tools/VerifiableLog4jAppender.java#L233 On Thu, Nov 9, 2017 at 2:17 PM, Arunkumar wrote: > Hi All > We have a requirement to migrate log4J 1.x to log4j 2 for our kafka > brokers us

Re: Migrating from log4j 1 to log4j 2

Hi Ted Thanks for quick response. Yes, I know it is used. But my question is if I migrate without changing the code will it break anything. We are using Confluent Kafka and we cannot modify kafka code. All I can do is change log4j 1.2.7 jar to latest,  create log4j.xml  and add log4j bridge util

How do I gracefully handle stream joins where the other side never appears?

Hi all, So let's say I have 2 topics coming that I want to join using KStream#join. I set them up like so: KStreamBuilder builder = new KStreamBuilder(); KStream a = builder.stream(TOPIC_A); KStream b = builder.stream(TOPIC_B); a.join(b, (msgA, msgB) -> msgA + msgB, JoinWindows.of(TimeUnit.HOU

Re: kafka streams with multiple threads and state store

Ranjit, Note that the "testStore" instance you are passing is a StateStoreSupplier which will generate a new StateStore instance for each thread's task. If you really want to have all the thread's share the same state store you should implement your own StateStoreSupplier that only return the sam

How to handle messages that don't find a join partner in Streams?

Hi all, So let's say I have 2 topics coming that I want to join using KStream#join. I set them up like so: KStreamBuilder builder = new KStreamBuilder(); KStream a = builder.stream(TOPIC_A); KStream b = builder.stream(TOPIC_B); a.join(b, (msgA, msgB) -> msgA + msgB, JoinWindows.of(TimeUnit.HOU

Kafka Streams - Custom processor "init" method called before state store has data restored into it

I have a custom processor that implements AbstractProcessor and a custom store that implements StateStore. Before Kafka 1.0.0 the processors "init" method gets called after the state store is restored from changelog and that is good. With Kafka 1.0.0 the processors "init" method is called BEFOR

Re: Kafka 0.9.0.1 partitions shrink and expand frequently after restart the broker

I‘m so sorry for my poor english. what I really means is my broker machine is configured as 8 core 16G. but my jvm configure is as below. java -Xmx1G -Xms1G -server -XX:+UseG1GC -XX:MaxGCPauseMillis=20 -XX:InitiatingHeapOccupancyPercent=35 -XX:+DisableExplicitGC -Djava.awt.headless=true -Xloggc

答复: Kafka 0.9.0.1 partitions shrink and expand frequently after restart the broker

Seems broker `4759750` was always removed for partition [Yelp, 5] every round of ISR shrinking. Did you check if everything works alright for this broker? 发件人: Json Tu 发送时间: 2017年11月10日 11:08 收件人: users@kafka.apache.org 抄送: d...@kafka.apache.org; Guozhang Wang 主

Re: Kafka 0.9.0.1 partitions shrink and expand frequently after restart the broker

The broker with broker id 4759750 is just restart，and there are 500+ replica partitions shrink and expand frequently，and there leader partition is distributed in the other 5 brokers. the log is pulled from one broker，and extract logs related to 1 partition. > 在 2017年11月10日，下午12:06，Hu Xi 写道： >

Questions about kafka-consumer-groups output

Hello all: I’m using Kafka version 0.11.0.1, with the new Java consumer API (same version), and commit offsets to Kafka. I want to get the consumer lags, so I use the following operation command: $ bin/kafka-consumer-groups.sh --bootstrap-server localhost:9093 --describe --group foo.test.cons