Questions tagged [apache-kafka]
Apache Kafka is a distributed streaming platform designed to store and process high-throughput data streams.
30,205
questions
0
votes
1
answer
19
views
Understanding relationship between ConcurrentKafkaListenerContainerFactory and KafkaMessageListenerContainer with RangeAssignor in Spring Boot
In the docs it is stated that after you set concurrency on ConcurrentKafkaListenerContainerFactory you will get that many instances of KafkaMessageListenerContainer. Also, if you use RangeAssignor ...
1
vote
1
answer
37
views
Kafka & Python app on Kubernetes in separate pods - NoBrokersAvailable()
to start with - I am a sort of newbie to Kubernetes and I might omit some fundamentals.
I have a working containerized app that is orchestrated with docker-compose (and works alright) and I am ...
0
votes
0
answers
11
views
"min.fetch.bytes" not Working Properly when used in Conjunction with "fetch.max.wait.ms" in Kafka Consumer
I have an use case where I need to wait for some specific time interval before fetching records from Kafka . But if a min data is present in the topic , I need to get the records immediately and not ...
0
votes
1
answer
14
views
Kafka IHostedService consumer in Kubernetes multiple instances
I have an ASP.NET Core WebAPI service that runs in Kubernetes. That service also hosts a IHostedService that is a Kafka consumer.
Would there be an issue with the Kafka consumer if I configure ...
0
votes
2
answers
23
views
Is there any way I can maintain ordering (based on an attribute not by the message time) in a single partition of a kafka topic?
Let's say, this is a one-partition topic, and while consuming the message I want to read it in a sequence based on one of the attributes (Let's assume attr1) in the message.
Message 1 ('attr1'="...
0
votes
1
answer
20
views
Kafka consumer group not rebalancing when increasing partitions
I have a situation where in my dev environment, my Kafka consumer groups will rebalance and distribute partitions to consumer instances just fine after increasing the partition count of a subscribed ...
0
votes
0
answers
18
views
C# Kafka Producer using AWS MSK
I am currently trying to implement a C# dotnet Kafka producer which connects to my AWS MSK brokers.
The template code I am using is, with utilizes the Confluent.Kafka library:
var producerConfig = ...
0
votes
1
answer
15
views
Kafka Connect sink connector configured for both insert and update
I have a MySQL database and I use the Confluent Kafka Connect JDBC Sink Connector to insert elements into the table (I am not able to make any changes to the database schemas... other systems rely ...
0
votes
1
answer
14
views
Kafka - Long running job decreases through put
I have three partitions and three consumers in my consumer group.
Job A, B assigned to partition1
Job C, D assigned to partition2
Job E, F assigned to partition3
Job C,D,E,F took less than 20 seconds ...
0
votes
1
answer
21
views
Add partitions for Kafka topic dynamically using Spring Boot?
I was able to inspect particular topic for its partitions:
public void addPartitionIfNotExists(int partitionId){
Map<String, TopicDescription> games = kafkaAdmin.describeTopics("games&...
0
votes
1
answer
25
views
Confluent Schema Registry - maximum number of schemas
I'm referring to the Confluent Schema Registry:
Is there reliable information on how many distinct schemas a single schema registry can support?
From how I understand the schema registry, it reads the ...
0
votes
0
answers
5
views
resilience4J for aggregated async results - retry on failure and/or timeout?
I have a scenario where is need to track the results of a group of Kafka messages. Each message is sent and we expect a SUCCESS or FAILURE message to return after a period of time. If any message of ...
0
votes
2
answers
68
views
How do you enable TLS 1.2 on C# for kafka producer/consumer config?
I want to set communication over TLS1.2 in C#
in java we do it by setting this property "ssl.enabled.protocols" looking for same functionality in c#
I'm using .Net core 3.1 and OS is linux
...
0
votes
0
answers
7
views
(Kafka Kstream) How to do some cleanup operations after Window Time closes?
What I'd like to do is:
Do some things in Kstream using .windowedBy(TimeWindows......)
After TimeWindow is complete, do some clean up operations like reset some variables.
0
votes
1
answer
15
views
How to connect kafka producer and consumer to a machine that is running kafka and zookeeper
I have a ubuntu machine that is having kafka and zookeepr installed in it, I am using spring boot for making the consumer and producer, locally the process works, however, when the deploy the producer ...
0
votes
1
answer
14
views
Is materialized store created per partition and key or only per key?
I have the following topology definition:
KStream<String, ObjectMessage> stream = kStreamBuilder.stream("topic1", "topic2");
stream.mapValues(new ProtobufObjectConverter())
...
0
votes
1
answer
19
views
Can there be gaps in Kafka's partition ids?
I'm looking for a specification of Kafka's partition ids. If a topic has n partitions, are its partition ids guaranteed to be 0 1 2 ... n-1 or are there ever scenarios in which gaps can appear, e. g. ...
0
votes
0
answers
11
views
Alter client quotas using kafka-python
I'm using kafka-python library to manage my kafka cluster. I want to add the quotas features (alter/describe), these features are not yet implented in kafka-python library but they are available in ...
0
votes
0
answers
31
views
So I'm new at using Docker and I'm trying to figure out how to use docker compose [duplicate]
version: "3.8"
services:
kowl-docker:
container_name: kowl-cluster
image: quay.io/cloudhut/kowl:master-edb20f8
restart: always
ports:
- 80:8080
environment:
...
0
votes
1
answer
18
views
Consume multiple topics at the same time from single consumer
There are multiple topics and these topics have no key value relation. The first payload of topic1 (has no key) example as it follows:
{ "bookList": [{"bookId": "1"}, {&...
0
votes
1
answer
20
views
NotEnoughReplicasException: Messages are rejected since there are fewer in-sync replicas than required
l have 5 kafka broker. And in my producer app, my configs is like below:
Interesting part is that, after getting this exception 2 times, in third time message is producing to topic.
spring:
kafka:
...
0
votes
1
answer
12
views
Kafka Topic Rebalance
I've a problem on kafka message consuming. For some topics it's unable to consume due to partition problem. So I want to delete all topics and recreate them with less partitions and they will have a ...
0
votes
0
answers
16
views
Vertical Scaling resources (CPU, RAM) for Kafka Production cluster
Currently we are using three node kafka Cluster in Prod with hardware specifications :
Zookeeper :
CPU 2 Core
Ram 4GB
Kafka :
CPU 4 Core
Ram 8GB
We want to vertically scale our Kafka cluster to double ...
1
vote
0
answers
25
views
confluent-kafka-go consumer batch commits
I prefer confluent-kafka-go driver. It's fast but it needs some workaround. The most tricky part is commiting messages. For the better performance we shouldn't commit every message, it's better to do ...
0
votes
0
answers
17
views
Microservices: Ensure all the required kafka events are in place before beginning of enrichment
I have a product and its complete information is obtained from various JSON messages. In a Microservices architecture, how can I ensure all the required events are in place before I begin the ...
0
votes
2
answers
24
views
Difference between num.standby.replicas vs max.warmup.replicas
I am bit confused to understand the difference between num.standby.replicas and max.warmup.replicas.
Both sound same to me as both are helping to reduce the time taken in getting a standby task and it'...
1
vote
1
answer
29
views
Advantage of "forcing" the partition
I have a topic theimportanttopic with three partitions.
Is an advantage to "forcing" the partition assignment?
For instance, I have three consumers in one group. I want consumer1 to always ...
-2
votes
1
answer
39
views
Performance issues with beam, flink & kafka pipeline
I am looking for help/guidance with the apache beam pipeline, which suffers from an unexpected slowdown. I have python apache beam pipeline running on the local flink cluster (1.14), which reads from ...
0
votes
1
answer
21
views
Java / Maven Loader Constraint Violation with Apache Kafka with org.slf4j
i'm currently struggling with a java.lang.LinkageError: loader constraint violation.
I'm developing a Jira Server App with the Atlassian Jira SDK that will include a Apache Kafka Producer. I added a ...
0
votes
3
answers
40
views
Flink using multiple structures of data in java
I am reading data from kafka in Java to perform some processing in Apache Flink and sink the results.
I have the kafka topic topic_a which has some data like {name: "abc", age: 20} and some ...
-2
votes
1
answer
20
views
Consuming data from multiple Topics, Aggregate them and process
We are consuming data from multiple Kafka Topics (Topic One- Employee Basic Details & Topic 2: Having address details) and then consuming an API (/createEmployee) to another system. In order to ...
0
votes
1
answer
18
views
No JAAS configuration section named 'Client' was found in specified JAAS configuration file
Can't deploy Kafka with SASL authentication
Here is my docker-compose.yml
version: '3.1'
services:
zookeeper:
image: confluentinc/cp-zookeeper
environment:
ZOOKEEPER_CLIENT_PORT: ...
0
votes
1
answer
19
views
JDBC sink topic with multiple structs to postgres
I am trying to sink a few topics top a postgres database. However the topic schema defines a array at the top level and within it multiple structs. Automapping does not work and I cannot find any ...
0
votes
1
answer
18
views
Http Response from Kafka Consumer
Would it be possible to create a server that publishes a client's request to Kafka's topic, and the Consumer that is subscribed to the topic will respond directly to the client?
client -> server -&...
0
votes
0
answers
10
views
How to deserialize the message if the data format is not known?
I am trying to consume messages from the Kafka topic but I don't know if the format of data is JSON or AVRO, so which KEY_DESERIALIZER_CLASS_CONFIG and VALUE_DESERIALIZER_CLASS_CONFIG to use.
Is there ...
0
votes
1
answer
18
views
Kafka Connect MySQL - No suitable driver found
I am trying to use Kafkaconnect to connect to mySQL.
As per the logs, I found the mySQL driver and JDBC connector had been loaded. However, I got an error "No suitable driver found for jdbc:mysql:...
0
votes
0
answers
12
views
Why is "kafkaProducer.send().get(timeout)" not working?
I have a simple kafka producer
public class JavaKafkaProducerExample {
public static void main(String[] args) throws ExecutionException, InterruptedException {
String server = "...
1
vote
0
answers
41
views
Using a Transaction between Two different Ports in Hexagonal Architecture
We're using a hexagonal architecture in one of our microservice. Spring Boot is the framework implementing the service.
For a use case, we need to update a database table (relational) and send a ...
0
votes
0
answers
9
views
Kafka connector exception logs not showing in Datadog
I have a kafka connector application. There I am trying to send an exception as a log into Datadog. But I could not able to send that exception to Datadog.
this is my exception which I want to send to ...
0
votes
1
answer
15
views
Flink KafkaSource with multiple topics, each topic with different avro schema of data
If I do multiple of the below - one for each topic:
KafkaSource<T> kafkaDataSource = KafkaSource<T>builder().setBootstrapServers(consumerProps.getProperty("bootstrap.servers"))....
0
votes
0
answers
13
views
Manipulate kafka stream from a topic
I have a postgres table and connected it to Kafk using debezium connector. Now I want to edit the message (adding one custom column) in the kafka topic (postgres table data) and convert it as a stream ...
0
votes
0
answers
18
views
debezium date/time field value out of range: 0000-12-30T00:00:00Z
We use debezium to sync data to
In the source table we have column timestamptz start_at, when the value is zero start_at='0001-01-01 00:00:00.000000 +00:00', but when we check the data in kafka, it's ...
0
votes
1
answer
12
views
Is it possible to specify Kafka Stream topology starting sequence
Let say I have Topology A that streams from Source A to Stream A, and I have Topology B which stream from Source B to stream B (used as Table B).
Then I have a stream/table join that joins Stream A ...
0
votes
0
answers
7
views
Change pyflink consumer from an existing repository to comply with newer version of pyflink (version > 1.15)
My task is to accept incoming kafka messages using pyflink. Then the messages will be parsed. I am using the following repository as my reference: https://github.com/ArmenShamelian/kafka-spark-flink-...
0
votes
0
answers
14
views
kafka broker restart in clean state
We are using apache kafka 1.0 .We had stopped one of the kafka brokers in our cluster for an activity and its disk got fully wiped. We added it back with the same broker id so it started syncing with ...
0
votes
0
answers
9
views
i have been trying to run my Apache kafka Node. however, I keep in geeting error message Invalid config, exiting normally
I cannot start my Node server in Apache Kafka. my id is not displayed rather it only shows [myid: ]
i tried changing the port numbers yet I still have the same problem. find the attached copy of ...
0
votes
0
answers
12
views
Clarification on topic metadata sync in MM2.0 logs
In MM2.0 logs, I often see the following:
Resetting the last seen epoch of topic-partition to 0 since the associated topicId changed from null to XXXXX (org.apache.kafka.clients.Metadata:402)
I can ...
0
votes
1
answer
22
views
KAFKA with docker : cluster offline but container are up according "to docker ps -a"
I'm new with kafka and i'm trying to do something simple.
I run kafka with docker.
here is my docker-compose:
version: "3.2"
services:
zookeeper:
image: wurstmeister/zookeeper
...
0
votes
0
answers
20
views
how to read data from a streaming kafka topic in databricks' notebooks?
I have a kafka topic that I ran on confluent, and now I want to connect to it inside my databricks notebook in order to be able to then use it to build delta live tables (or just delta tables). I seem ...
0
votes
0
answers
9
views
NX.dev execute issue
i have short and simple example of kafkaJs consumer, its works well on manual node build. But when im trying to start this on NX build i have that error:
"TypeError [ERR_UNKNOWN_FILE_EXTENSION]: ...