Questions tagged [apache-kafka]
Apache Kafka is a distributed streaming platform that is used for building real-time data pipelines and streaming applications.
32,329
questions
0
votes
0
answers
8
views
Enabling TLSv1.1 with AES128 Cipher in Kafka 2.8 with Confluence Docker Container (Java-11) for Backward Compatibility
We are currently in the process of migrating from Kafka 1.0 to Kafka 2.8, utilizing the Confluence Kafka container image version 6.2.0. Our challenge lies in maintaining backward compatibility with ...
0
votes
1
answer
9
views
Can enableObjectReuse() cause data discrepency In Flink-SQL based job
Can env.getConfig.enableObjectReuse() cause data discrepancy In Flink-SQL based job where we don't have any custom type, serializer/deserializer ?
create table T1 (...) WITH ( 'connector' = 'upsert-...
0
votes
0
answers
10
views
Configure ports using KafkaClusterTestKit
How can we configure a custom port when using KafkaClusterTestKit to create an embedded Kafka instance in kraft mode? KafkaClusterTestKit appears to assign a random port when creating a Kafka instance ...
0
votes
1
answer
15
views
Does KafkaJS auto-pause during rebalancing?
Does KafkaJS automatically pause during rebalancing? It's not mentioned in the KafkaJS docs what's going to happen during a rebalance. Or how does KafkaJS handle rebalance?
0
votes
0
answers
20
views
deduplicate (different file contents found in the following) of netty and jackson after changing to Kafka 2.4
I want to benchmark different versions of Kafka. I've already done some tests for Kafka 2.0 to 2.3. After changing to Kafka 2.4 (also tried for 2.5), I'm flooded with errors concerning deduplicates: ...
0
votes
1
answer
12
views
When does compaction in kafka happens if min.cleanable.dirty.ratio, max.compaction.lag.ms and max.compaction.lag.ms parameters are set?
I've set for my topic these three parameters:
max.compaction.lag.ms: 43200000
min.compaction.lag.ms
min.cleanable.dirty.ratio: "0.5
Did I get it right that compaction on this topic will happen ...
0
votes
0
answers
4
views
spark-streaming (DStream): Does Spark stream Support pause/resume consumption of message from Kafka on runtime
We are using Spark 3.4.0 DStream in our application for Message consumption and processing via Kafka.
I have a use case, where I need to pause message consumption on some condition and resume back ...
-1
votes
1
answer
17
views
Consume MSK topic
I am working on a AWS MSK consumer project, where I am looking to develop/test my application before deploying to EMR. However I am not aware how to configure my local to consume from my MSK kafka ...
0
votes
1
answer
20
views
I need to poll only unprocessed message from kafka
Scenario:
I've a kafka consumer which polls message from kafka every second from the beginning. Message contains timestamp field so that I can filter only those messages which has time is equal to ...
0
votes
1
answer
20
views
Caused by: java.lang.ClassNotFoundException: org.apache.kafka.common.serialization.ByteArraySerializer
i am doing spark structured streaming with kafka and cassandra while i am running below command i got an error
spark-submit --class StreamHandler --master local[*] --packages "org.apache.spark:...
0
votes
0
answers
11
views
I have a flink program,but it can not write some data into redis
I have a Flink program that receives information from Kafka, performs some word frequency statistics, and then writes the results to the Redis server. After running it, it can see the information it ...
0
votes
0
answers
14
views
Failed to find data source: kafka. Please deploy the application as per the deployment section of "Structured Streaming + Kafka Integration Guide [closed]
I was working on a kafka pyspark structured streaming workflow usind docker. All the dependencies are in docker: kafka,pyspark,postgres.
this pyspark script sshould receive the streams data from kafka ...
0
votes
1
answer
14
views
How to get the broker address of Kafka in the app?
I use Java, Spring-Boot and Kafka in a project.
@Autowired
private KafkaTemplate <String, MyRecord> kt;
private void produce ()
{
MyRecord mr = new MyRecord ("abc");
...
0
votes
1
answer
11
views
kafka http sink connector: request latency
I have a http sink connector running in kubernetes. Its making requests to a rest service. The amount of messages consumed and sent to the topic seems low. Is there any metric which the connector ...
0
votes
0
answers
22
views
twisted python 2.7 integrate with confluent-kafka issue for a legacy system
First of all, yes it's a legacy system which is not yet completely ported to Python 3 yet.
I have this piece of code where there is a Twisted program which serves some HTTP request, for simplicity I'm ...
0
votes
1
answer
31
views
Apache Kafka + how to defected Kafka clusters with Kraft mode or Kafka with zookeeper mode
We have dozens different Kafka clusters labs
Some of them are with Kraft mode and most of them are ordinary Kafka cluster with zookeeper
Since we run smoke tests scripts based on python/perl/bash on ...
0
votes
0
answers
10
views
When does a kafka connect SourceTask fail?
I am looking into emitting metrics when a SourceTask fails for specific reasons only, my current understanding is that whenever start or poll encounters an unhandled exception source task enters ...
0
votes
1
answer
20
views
LAMBDA not achieving expected concurrency with MSK Triggers
I've configured two services on AWS Lambda to trigger off an MSK (Amazon Managed Streaming for Apache Kafka) event source. Both services are supposed to handle messages from 100 partitions. However, ...
1
vote
1
answer
62
views
Giving completion instructions ahead of calling send with MockProducer
I have the following code (Obviously made as an example):
class RandomClass(
val producer: Producer<String, String>
) {
fun randomFunction(): Boolean {
// Using .get() because I want to ...
0
votes
0
answers
29
views
Invoke Kotlin suspended function inside reactive stream
I use Spring Boot app and Kotlin lang. I want to receive messages though the reactive Kafka consumer, and then pass these messages to the suspended Kotlin function and use coroutines there, then get ...
0
votes
1
answer
20
views
MSK Trigger process duplicate message
We have below setup
MSK Topic(10 partitions) -> Trigger -> Lambda(Single Group - Single consumer)
Lambda Details:
Batch Size : 1
Reserved Concurrency : 1
Lambda Retry : 0
Lambda Timeout : 2mins
...
0
votes
0
answers
7
views
Debezium tries to ALTER tables even if I'm manually handling them
I'm currently testing Debezium's behaviour, running inside Kafka Connect, on handling a schema modification with two identical MySQL dbs.
Firstly, I create a table in both dbs:
CREATE TABLE testTable (...
0
votes
0
answers
27
views
High load: real-time get SQL message and send it to the Kafka broker. What architectural pattern is suitable here?
I need to make a simple Spring Boot stream Java application that selects messages from a table in the database and sends them to a Kafka topic. And on success - deleting messages from PostgreSQL:
The ...
0
votes
0
answers
22
views
Pausing offset and commit in Kafka
I am new to Kafka and trying to handle a case where I want the offset and commit to pause and want it to get increase at some specific point(when the message is read successfully).
This is my consumer:...
0
votes
0
answers
17
views
Apache Kafka Microservice in docker [duplicate]
I'm currently facing challenges managing the connection between two Docker-based microservices using Apache Kafka. When running the microservices without Docker, everything works smoothly. However, ...
0
votes
0
answers
37
views
Unable to get kafka_connect_connector_status on Promethues
I am using Confluent for Kubernetes to deploy Kafka Connect. For metrics monitoring I am using promethues operator and have added a job in Secrets name additional-scrape-configs. The configurations ...
0
votes
0
answers
30
views
What is the consumer group used to consume from a kafka changelog topic?
I have this kafka-streams application which is returning Prometheus metrics. If I were to believe these metrics, the application has a lag on most partitions of the changelog topic.
I wanted to verify ...
0
votes
0
answers
9
views
Write topic data from Redis-sink-connector to redis container
We are looking for sending topic data from Kafka Redis sink connector to a docker redis container. The connector state is showing running, and it connected with topic as well , still I am not getting ...
0
votes
0
answers
14
views
How to configure confluentinc/cp-kafka with SASL_PLAINTEXT authentication in docker-compose?
I have working configuration for confluentinc/cp-kafka that works fine in docker-compose as following:
User
kafka-kraft:
image: confluentinc/cp-kafka:7.5.1
container_name: kafka-kraft
...
1
vote
0
answers
15
views
RejectedExecutionException when running Kafka + Netty in OpenShift
We're developing a stateful Kafka application in Kotlin designed to consume records from two topics and produce records to another topic. The application runs as a statefulset in OpenShift with three ...
0
votes
1
answer
24
views
Force clickhouse to reset offset of kafka engine table
I have a kafka topic and a Created a MergeTree Table and a Kafka Table and a Materialized view to get data from topic .
I need to consume message from earliest offset. My new consumer group offset set ...
0
votes
0
answers
11
views
Apache StreamPipes - Kafka topic data reading time out
i have the problem with loading the data from Kafka topic to Apache StreamPipes adapter.
I'm using Kafka 3.5.0 and Apache StreamPipes 0.93 (but the same problem occurs on 0.92), i have the default ...
0
votes
0
answers
50
views
Need to read data from Kafka and Hive in one query via Flink(FlinkSQL)
I want to run a heterogeneous query in Flink to read several tables from Kafka (the data is written in Kafka so that you can join them via FlinkSQL), and several tables from Hive, in one query.
I can ...
-1
votes
0
answers
20
views
logstash used with ms sql server to ingest data to elastic seach server [closed]
What is the most efficient solution for synchronizing data between Elasticsearch and SQL Server databases with a large number of tables. and how can this real time.
My stack is python, django, kibana ...
0
votes
0
answers
16
views
How to use C#/C++/librdkafka with OIDC and Azure AD as token provider for OAUTHBEARER in apache kafka?
I try to use C# to produce message to Apache Kafka. But I notice there are a config called sasl.login.callback.handler.class which should be configured in kafka producer client. But I can't find the ...
0
votes
0
answers
19
views
Can we add interceptors in Kafka Consumers?
I'm planning to fetch some photon metrics from a spring boot application. This application acts as a Kafka Consumer. Similarly I have multiple spring boot applications.
I want to make this activity ...
0
votes
1
answer
45
views
The Kafka Consumer cannot find any available Kafka brokers to connect to
The Kafka Consumer cannot find any available Kafka brokers to connect to.
I was hoping to be able to observe the logs, because when I use docker-compose up -d, the topics created after starting kafka ...
0
votes
0
answers
18
views
Deserialize Kafka-Avro-Message received from AWS Lambda Trigger
We have set up a Kafka Lambda Trigger in AWS to consume from a Kafka read API. The connection was successful and I can receive Json Objects in my C# Lambda function. However, the Json Objects contain ...
1
vote
0
answers
29
views
Spark job stuck at running process
I use docker-compose to make a cluster of Spark and Kafka. When I tried to consume a topic by submitting it to this cluster, the job was always stuck and not executed.
Here is my docker-compose file:
...
0
votes
0
answers
18
views
kafkaProducer parameter not in JSON format
I am getting below error after upgread the java verison from 1.8 to 11
java.lang.IllegalArgumentException: kafkaProducer parameter not in JSON format:$propertySource.spring.cloud.stream.kafka.bindings....
0
votes
0
answers
55
views
Clickhouse import of "tombstone" Kafka messages with Kafka Engine as RawBLOB
We are using Clickhouse to read messages from Kafka as RawBLOB, which used to work for our use case.
However since we updated Clickhouse from 21.2.3.15 to 23.3.13.6 Kafka Engine is not reading "...
0
votes
1
answer
20
views
Unable to get Kafka working due to address issue
I'm trying to follow a cloud guru course on kafka but when I setup the server and try and run it I get this error in the kafka server.log file:
WARN Cannot open channel to 2 at election address zoo2/...
0
votes
0
answers
12
views
Intermittent "Node X disconnected" issue for Kafka despite batch.size , linger.ms change
I am trying to consume/read/receive and publish/write/send data from/to Kafka without intermittent "Node X disconnected" issue.
With this reactive Kafka consumer and producer config:
public ...
0
votes
0
answers
13
views
Kafka transaction dirty reads
I am trying to se if it is simple and possible to implement a ledger using Kafka.
I am basically trying to implement something that would match this example :
https://developer.confluent.io/learn/...
0
votes
0
answers
27
views
Parallel consumer with spring boot
I'm currently testing how the parallel-consumer library (https://github.com/confluentinc/parallel-consumer) behaves. However, I've encountered an issue while setting up the factory container ...
0
votes
2
answers
23
views
How to unit test the intercept method of a RecordIntercept interface from Kafka using Mockito
I have the class KafkaRecordInterceptor which implements the interface RecordInterceptor<String, Message>
as follow
@Component
public class KafkaRecordInterceptor implements RecordInterceptor<...
0
votes
0
answers
34
views
Apache Camel Kafka consumer consuming message even after pausing
I have a Kafka consumer in Camel route which consumes the message as below:
from("kafka:MykafkaTopic?brokers=localhost:9092&autoCommitEnable=false&allowManualCommit=true&...
0
votes
0
answers
6
views
Kafka Consumer Offsets expiration
I have this scenario.
A kafka broker with offsets.retention.minutes set to 1440 (1 day)
1 consumer C, consuming from topic T
Topic T has 1 week retention.
Oldest data on topic T is from 1 day ago.
...
0
votes
0
answers
11
views
Is there any modern nodejs client for kafka in Kraft mode?
I want to use kafka in my nodejs application and only proper client library kafkajs .
i have hosted kafka:3.6.0 in Raft mode using docker . The problem i am facing is kafkajs throws this error
...
0
votes
0
answers
26
views
Can Flink only process part of datastream? [closed]
For example, I want 80% of data from a Kafka topic to be processed by operators, and the rest 20% remain not processed. Any way to achieve this?