Questions tagged [apache-kafka]

Apache Kafka is a distributed streaming platform that is used for building real-time data pipelines and streaming applications.

Filter by
Sorted by
Tagged with
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 ...
KiranKumarSingani's user avatar
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-...
Faisal Ahmed Siddiqui's user avatar
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 ...
sobychacko's user avatar
  • 5,194
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?
Mithiridi Prasanth's user avatar
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: ...
SkogensKonung's user avatar
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 ...
Магомед Якубов's user avatar
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 ...
Saurabh Agrawal's user avatar
-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 ...
bheem singh's user avatar
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 ...
vijay's user avatar
  • 23
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:...
UTHAYAKUMAR M's user avatar
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 ...
cater rick's user avatar
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 ...
Louis_Curry's user avatar
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"); ...
chris01's user avatar
  • 11.2k
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 ...
tomsoyer's user avatar
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 ...
Jithin's user avatar
  • 1,690
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 ...
Judy's user avatar
  • 1,601
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 ...
Akash's user avatar
  • 1
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, ...
Bharat Bollampally's user avatar
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 ...
Stene's user avatar
  • 157
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 ...
user2625402's user avatar
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 ...
Sugaan's user avatar
  • 43
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 (...
grizzo's user avatar
  • 161
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 ...
Сергей Аблаев's user avatar
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:...
Thor's user avatar
  • 133
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, ...
BAra's user avatar
  • 1
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 ...
Muhammad Areeb Siddiqui's user avatar
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 ...
Bilal Ennouali's user avatar
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 ...
Swati's user avatar
  • 1
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 ...
Anatoly's user avatar
  • 5,135
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 ...
HaavardG's user avatar
  • 221
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 ...
masoud's user avatar
  • 458
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 ...
Michał's user avatar
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 ...
Vladimir Frolov's user avatar
-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 ...
Alberuni Beruni's user avatar
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 ...
user22969567's user avatar
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 ...
dev333's user avatar
  • 721
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 ...
Otakus Players's user avatar
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 ...
Mindful's user avatar
  • 15
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: ...
Nguyễn Quốc Nhật Minh's user avatar
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....
Lakshmibabu's user avatar
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 "...
John A.'s user avatar
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/...
Andrew's user avatar
  • 2,668
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 ...
PatPanda's user avatar
  • 3,894
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/...
Antonin's user avatar
  • 899
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 ...
Amber Pandey's user avatar
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<...
Bogdan Alin Muresan's user avatar
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&...
Dev2405's user avatar
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. ...
Davide Berra's user avatar
  • 6,447
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 ...
arslanmughal's user avatar
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?
yqchen's user avatar
  • 11

1
2 3 4 5
647