Kafka

Hi, May I ask for some help about setting up SSL for Zookeeper, please? My goal is just to add SSL for my zookeeper server then just test it using openssl s_client. So far I got this from the logs (WARN) "Empty server certificate chain".  So here's what I've done so far. #1 Install Kafka and Zookeeper in my Ubuntu machine (using multipass). I downloaded the latest version here #2 I was able to connect Kafka to Zookeeper just fine. Zookeeper: kafka@zoo:/etc/kafka$ bin/zookeeper-server-start.sh config/zookeeper.properties #3 Test the SSL connection without setting up the zookeeper.properties and .jks files kafka@zoo:/etc/kafka$ openssl s_client -showcerts -connect localhost:2181 CONNECTED(00000003) write:errno=0 --- no peer certificate available --- No client certificate CA names sent --- SSL handshake has read 0 bytes and written 283 bytes Verification: OK --- New, (NONE), Cipher is (NONE) Secure Renegotiation IS NOT supported Compression: NONE Expansion: NONE No ALPN ...

Jamie, this would depend on KIP-712 (or similar) aka "shallow mirroring". This is a work in progress, but I'm optimistic it'll happen at some point. ftr, "IdentityReplicationPolicy" has landed for the upcoming release, tho "identity" in that context just means that topics aren't renamed. Ryanne On Thu, Jul 29, 2021, 11:37 AM Jamie <jamiedd13@aol.co.uk.invalid> wrote: > Hi All, > This blog post: > https://blog.cloudera.com/a-look-inside-kafka-mirrormaker-2/ mentions > that "High Throughput Identity Mirroring" (when the compression is the same > in both the source and destination cluster) will soon be coming to MM2 > which would avoid the MM2 consumer decompressing the data only for the MM2 > producer to then re-compress it again. > Has this feature been implemented yet in MM2? > Many Thanks, > Jamie

Hi All,  This blog post:  https://blog.cloudera.com/a-look-inside-kafka-mirrormaker-2/ mentions that "High Throughput Identity Mirroring" (when the compression is the same in both the source and destination cluster) will soon be coming to MM2 which would avoid the MM2 consumer decompressing the data only for the MM2 producer to then re-compress it again. Has this feature been implemented yet in MM2? Many Thanks, Jamie 

I see. Are you able to connect to the databricks spark cluster with a scala repl? The same URL should work with the R spark client. I'm not sure how you're meant to specify the kafka credentials but perhaps this will help you connect r to spark. On Thu, Jul 29, 2021, 6:35 AM Paulo Ramos < paulo.ramos@3pillarglobal.com > wrote: > Yes > > That piece of code is not working for me, because when I run it I am not > getting it. > > I am trying to connect through Data Bricks, so I think I have to change > the "local" connection in master? > > Also, how do I add in the read options the kafka username and psswd? > > Notes: Kafka is hosted in confluent > > Best Regards > Paulo > > On Wed, Jul 28, 2021 at 8:52 PM Blake Miller < blak3mill3r@gmail.com > > wrote: > >> Paulo, >> >> Could you clarify what your question is? I wasn't able to understand it >> f...

Yes That piece of code is not working for me, because when I run it I am not getting it. I am trying to connect through Data Bricks, so I think I have to change the "local" connection in master? Also, how do I add in the read options the kafka username and psswd? Notes: Kafka is hosted in confluent Best Regards Paulo On Wed, Jul 28, 2021 at 8:52 PM Blake Miller < blak3mill3r@gmail.com > wrote: > Paulo, > > Could you clarify what your question is? I wasn't able to understand it > from your email. > > I understand that you want to read data from Kafka in R, and you shared > some R code. What is the issue that you're having with this code? > > Regards, > Blake > > On Tue, Jul 27, 2021 at 2:42 PM Paulo Ramos > <paulo.ramos@3pillarglobal.com.invalid> wrote: > > > Hi All, > > > > I have a technical question, regarding reading stream msg from kafka > using ...

Paulo, Could you clarify what your question is? I wasn't able to understand it from your email. I understand that you want to read data from Kafka in R, and you shared some R code. What is the issue that you're having with this code? Regards, Blake On Tue, Jul 27, 2021 at 2:42 PM Paulo Ramos <paulo.ramos@3pillarglobal.com.invalid> wrote: > Hi All, > > I have a technical question, regarding reading stream msg from kafka using > R. > > This is what I am currently using: > > if (FALSE) { library < https://rdrr.io/r/base/library.html >(sparklyr > < https://spark.rstudio.com/ >) sc <- spark_connect > < https://spark.rstudio.com/reference/spark-connections.html >(master = > "local", version = "2.3", packages = "kafka") read_options <- list > < https://rdrr.io/r/base/list.html >(kafka.bootstrap.servers = > "localhost:9092", subscribe = ...

Hi Peter Bukowinski, Thanks for your response. That's a great deal of observation. You have rightly nailed down the problematic instance. Please find broker #21 and zookeeper logs. I see a lot happening, but something unwell! Any advice on the problem. Thanks in advance! Broker #21 logs: ---------------- ERROR [ReplicaFetcher replicaId=21, leaderId=11, fetcherId=1] Error for partition xxx to broker 11:org.apache.kafka.common.errors.NotLeaderForPartitionException: This server is not the leader for that topic-partition. (kafka.server.ReplicaFetcherThread) ERROR [ReplicaFetcher replicaId=21, leaderId=11, fetcherId=1] Error for partition xxx-3 to broker 11:org.apache.kafka.common.errors.NotLeaderForPartitionException: This server is not the leader for that topic-partition. (kafka.server.ReplicaFetcherThread) ERROR [ReplicaFetcher replicaId=21, leaderId=11, fetcherId=1] Error for partition xxx-1 to broker 11:org.apache.kafka.common.errors.NotLeaderForPartitionEx...

Hi Sridhar, If your min.insync.replicas value is set to 3, then kafka won't be able to move replicas until there are three replicas listed in the ISR. I would look into the health of broker 21 — it's either down or unhealthy. It's the only one not showing in the ISR list. — Peter Bukowinski > On Jul 27, 2021, at 1:12 AM, Sridhar Rao < sridharnuvvula@gmail.com > wrote: > > Hi Fabio Pardi, > > Thanks for your prompt response. > Split brain was our suspicion and we are investigating other possibilities. > Perhaps our understanding of the problem might be incorrect at the moment. > The issue started when one of the broker instances went down abruptly (3 > brokers, 3 zookeepers) and the cluster was unstable. > > Later, we were able to restart the affected broker instance followed by > rolling restart of other 2 brokers. The cluster was stabilized at this > point. > However, we noticed un-repl partitions a...

Hi All, I have a technical question, regarding reading stream msg from kafka using R. This is what I am currently using: if (FALSE) { library < https://rdrr.io/r/base/library.html >(sparklyr < https://spark.rstudio.com/ >) sc <- spark_connect < https://spark.rstudio.com/reference/spark-connections.html >(master = "local", version = "2.3", packages = "kafka") read_options <- list < https://rdrr.io/r/base/list.html >(kafka.bootstrap.servers = "localhost:9092", subscribe = "topic1") write_options <- list < https://rdrr.io/r/base/list.html >(kafka.bootstrap.servers = "localhost:9092", topic = "topic2") stream <- stream_read_kafka(sc, options = read_options) %>% stream_write_kafka < https://spark.rstudio.com/reference/stream_write_kafka.html >(options = write_options) stream_stop < https://spark.rstudio.com/reference/stream_stop.html >(stream) }...

Dear all,     I am running a simple kafka consumer group reactive autoscaling experiment on kubernetes, while leveraging range stop of the world assignor in the first run, and next in the second run I used incremental cooperative assignor. My workload is shown below where x-axis is the time in seconds, and y axis is the corresponding batch of messages to be sent to the broker.     At the consumer side, I started with one consumer configured with maximum consumption rate of 100 messages per second (max.poll.records = 100, and sleep 1 second across the call to poll), so every time total arrival rate becomes greater than current number of consumers *100 I automatically add a consumer etc… no complex processing logic for records, just simple logging….   Having run the experiment in stop of world range assignor (first Figure) and in cooperative sticky assignor (second figure), it can be shown unexpectedly that stop of the...

Hi Fabio Pardi, Thanks for your prompt response. Split brain was our suspicion and we are investigating other possibilities. Perhaps our understanding of the problem might be incorrect at the moment. The issue started when one of the broker instances went down abruptly (3 brokers, 3 zookeepers) and the cluster was unstable. Later, we were able to restart the affected broker instance followed by rolling restart of other 2 brokers. The cluster was stabilized at this point. However, we noticed un-repl partitions and Preferred Replica imbalance irregularities. [xxx(user):/xxx/install/1.0.0/bin] ./kafka-topics.sh --describe --zookeeper zookeeper1:2181 --under-replicated-partitions Topic: ABC Partition: 3 Leader: 31 Replicas: 31,21,11 Isr: 31,11 Topic: __consumer_offsets Partition: 1 Leader: 31 Replicas: 31,11,21 Isr: 31,11 Topic: __consumer_offsets Partition: 3 Leader: 31 Replicas: 21,11,31 Isr: ...

On 27/07/2021 09:19, Sridhar Rao wrote: > Hi Everyone, > > Recently we noticed a high number of under-replicated-partitions after > zookeeper split brain issue. > We tried fixing the issue by executing ./kafka-reassign-partitions.sh > procedure. However Kafka refuses to re-assign the partitions in ISR and > un-repl partitions remain the same. > > Kafka version: Apache Kafka 1.0.0 > > Any advice in this regard would be highly appreciated. Please feel free to > request any missing information. Thank you! > Zookeeper needs a quorum to elect a leader therefore a split brain is unlikely. Can you describe how split brain occurred? (number of instances, what happened and how it was resolved) Given that, if your topic/partitions topology did not change, (that I know) even a split brain should not affect your data, therefore your ISR. Running the below command should tell you which partitions are under replicated kafka-topics.sh --...

Hi Everyone, Recently we noticed a high number of under-replicated-partitions after zookeeper split brain issue. We tried fixing the issue by executing ./kafka-reassign-partitions.sh procedure. However Kafka refuses to re-assign the partitions in ISR and un-repl partitions remain the same. Kafka version: Apache Kafka 1.0.0 Any advice in this regard would be highly appreciated. Please feel free to request any missing information. Thank you!

Warning messages for ssl related configuration is fixed in 2.8.0 but following warning messages still appear. WARN The configuration 'config.providers.file.class' was supplied but isn't a known config. (org.apache.kafka.clients.admin.AdminClientConfig) WARN The configuration 'config.providers' was supplied but isn't a known config. (org.apache.kafka.clients.admin.AdminClientConfig) -----Original Message----- From: Tom Bentley < tbentley@redhat.com > Sent: Wednesday, July 14, 2021 5:51 PM To: Users < users@kafka.apache.org > Subject: Re: warning messages from kafka admin client This was fixed in 2.8, see https://issues.apache.org/jira/browse/KAFKA-10090 On Wed, Jul 14, 2021 at 12:19 PM Dhirendra Singh < dhirendraks@gmail.com > wrote: > Hi All, > whenever i run any kafka CLI tool i am getting following warning > messages > > kafka-topics.sh --bootstrap-server kafka-dev:9092 --command-config > client....

hi We have a kafka mirror maker 2 setup on aws. The issue is when mirror maker 2 starting it initially copied the messages from cluster A to cluster B, but then it stops replication ( while it is up). So new messages in cluster A is not getting to cluster B. Please help me , let me know if you require more information. Thanks in advance

Hi I already subscribe to the user help mail list. I have a technical question, regarding reading stream msg from kafka using R. This is what I am currently using: if (FALSE) { library < https://rdrr.io/r/base/library.html >(sparklyr < https://spark.rstudio.com/ >) sc <- spark_connect < https://spark.rstudio.com/reference/spark-connections.html >(master = "local", version = "2.3", packages = "kafka") read_options <- list < https://rdrr.io/r/base/list.html >(kafka.bootstrap.servers = "localhost:9092", subscribe = "topic1") write_options <- list < https://rdrr.io/r/base/list.html >(kafka.bootstrap.servers = "localhost:9092", topic = "topic2") stream <- stream_read_kafka(sc, options = read_options) %>% stream_write_kafka < https://spark.rstudio.com/reference/stream_write_kafka.html >(options = write_options) stream_stop < https://spark.rstudio...

Hi, What should be the consideration when using MM2 in production. Have it inside K8S or outside on top of VMs. What are you doing in production when using MM2 ? Thanks, Tomer This email and the information contained herein is proprietary and confidential and subject to the Amdocs Email Terms of Service, which you may review at https://www.amdocs.com/about/email-terms-of-service < https://www.amdocs.com/about/email-terms-of-service >

Hi all, Following my last question on soft (no failure to cluster..) exactly once semantics during a rolling upgrade for a consumer group on kubernetes. Maybe the required behavior is already guaranteed.., because with a grace period configured, Kubernetes will send a SIGTERM signal to the consumer for graceful shutdown, Next the consumer will send a leave group request to the group coordinator, where subsequently as part of the rebalancing protocol ConsumerRebalanceListener#onPartitionRevoked (where the lastest processed offset is committed), which will gracefully revoke the current partitions out of the current consumer, and assign them to different consumer.... so already soft (no cluster failure) excatly once is possibly guaranteed? Can someone confirm this behavior and/or provide a more accurate description of the behavior of a graceful shutdown of a consumer within a consumer group ? Thank you so much. ________________________________ From: Mazen Ezzeddine < m...

Dear all, I am interested in achieving zero down time upgrade of my kafka consumer group running on Kubernetes with say soft exactly once semantics (i.e., assume no failure and/or error is going to happen to my cluster). I configured my consumer group with rolling update and terminationGracePeriodSeconds, I also added a shutdown hook during which I am committing the last processed record by the consumer to be killed… would that guarantee soft exactly once rolling upgrade (i.e., under the assumption of no cluster failure…) Thank you.

yep! On Wed, Jul 21, 2021, 3:18 AM Tomer Zeltzer <tomerze@amdocs.com.invalid> wrote: > Hi, > > Can I use MirrorMaker2.0 from Kafka 2.8.0 with Kafka version 2.4.0? > > Thanks, > Tomer Zeltzer > > This email and the information contained herein is proprietary and > confidential and subject to the Amdocs Email Terms of Service, which you > may review at https://www.amdocs.com/about/email-terms-of-service < > https://www.amdocs.com/about/email-terms-of-service > >

Hi, Can I use MirrorMaker2.0 from Kafka 2.8.0 with Kafka version 2.4.0? Thanks, Tomer Zeltzer This email and the information contained herein is proprietary and confidential and subject to the Amdocs Email Terms of Service, which you may review at https://www.amdocs.com/about/email-terms-of-service < https://www.amdocs.com/about/email-terms-of-service >

Hi there, I'm trying to use open messaging to run some benchmarks for Kafka I've done the prerequisite setup for machines using terraform machine and ansible now while running the benchmark with sudo bin/benchmark --drivers driver-kafka/kafka.yaml workloads/1-topic-16-partitions-1kb.yaml , I'm getting 17:27:49.112 [main] INFO - Workers list - producers: [ http://10.0.0.241:8080 , http://10.0.0.241:8080 ] 17:27:49.112 [main] INFO - Workers list - consumers: [ http://10.0.0.241:8080 , http://10.0.0.241:8080 ] 17:27:49.115 [main] INFO - --------------- WORKLOAD : 1 producer / 1 consumers on 1 topic --- DRIVER : Kafka--------------- 17:27:49.201 [main] ERROR - Failed to run the workload '1 producer / 1 consumers on 1 topic' for driver 'driver-kafka/kafka.yaml' java.util.concurrent.CompletionException: java.net.ConnectException: Connection refused: / 10.0.0.241:8080 I'm guessing some service needs to run on port 8080, which has not bee...

> I've tried to restart the streams application using at_least_once > processing guarantee and it worked, restarted again in > exactly_once_beta and it worked too. > Spoke too soon, after a while in exactly_once_beta the "Restoration in progress for 1 partition" loop  started again. > Is there any reason why? > > On 7/20/21 2:09 PM, Alessandro Tagliapietra wrote: >> Hello everyone, >> >> after upgrading to kafka streams 2.8 we have one streams app that's >> stuck trying to restore a store changelog topic, this is the debug >> log of the app: >> >> https://gist.github.com/alex88/f31593aaabbd282b21f89a0252a28745 >> >> I would like to avoid having to delete and recreate the topic, what >> we should do? >> >> I've tried to switch back to exaclty_once (from exactly_once_beta) >> but nothing changed. >> >> Thank you in adv...

I've tried to restart the streams application using at_least_once processing guarantee and it worked, restarted again in exactly_once_beta and it worked too. Is there any reason why? On 7/20/21 2:09 PM, Alessandro Tagliapietra wrote: > Hello everyone, > > after upgrading to kafka streams 2.8 we have one streams app that's > stuck trying to restore a store changelog topic, this is the debug log > of the app: > > https://gist.github.com/alex88/f31593aaabbd282b21f89a0252a28745 > > I would like to avoid having to delete and recreate the topic, what we > should do? > > I've tried to switch back to exaclty_once (from exactly_once_beta) but > nothing changed. > > Thank you in advance >

Hello everyone, after upgrading to kafka streams 2.8 we have one streams app that's stuck trying to restore a store changelog topic, this is the debug log of the app: https://gist.github.com/alex88/f31593aaabbd282b21f89a0252a28745 I would like to avoid having to delete and recreate the topic, what we should do? I've tried to switch back to exaclty_once (from exactly_once_beta) but nothing changed. Thank you in advance

Hello everyone, after upgrading to kafka streams 2.8 we have one streams app that's stuck trying to restore a store changelog topic, this is the debug log of the app: https://gist.github.com/alex88/f31593aaabbd282b21f89a0252a28745 I would like to avoid having to delete and recreate the topic, what we should do? I've tried to switch back to exaclty_once (from exactly_once_beta) but nothing changed. Thank you in advance

Sophie, In continuation to the discussion above, now that we are using redis for storing state in our application which breaks the EOS of kafka streams, I do have one question and whether we can do some work around here: in the scenario : 1. Event consumed from source 2. Event processed and state stored in redis 3. Before event is sent to sink topic, the node goes down 4. Partition get rebalanced and another node consumes/processes same event again Even though redis state store could be inconsistent with rest of streams infrastructure like source/sink topics, can we still get exactly-once guarantee across source/sink topic where the offset commit to source and send to sink would all happen atomically or nothing would happen. If this is still guaranteed by kafka streams then we can think of applying some workaround while reading state from redis e.g. we could try and apply some conditional logic when fetching the state from redis based on unique id of event i.e. a...