Questions tagged [apache-kafka-streams]
Related to Apache Kafka's built-in stream processing engine called Kafka Streams, which is a Java library for building distributed stream processing apps using Apache Kafka.
3,601
questions
0
votes
0
answers
15
views
Data filtering with kafka and mysql
I receive data in following format:
msg1:
{
"A" : null,
"B" : 123,
"C" : {...}
}
msg2:
{
"A" : 456,
"B" : null,
"C" : {...}
}...
0
votes
2
answers
25
views
Multiple kafka Consumer Not receiving messages
I am using embeddedKafkaBroker and Kafka Binder Streams.
@Configuration
@Profile({"dev", "test"})
@Slf4j
public class EmbeddedKafkaBrokerConfig {
private static final String ...
0
votes
0
answers
6
views
Kafka stream - when join is triggered
Can someone please explain me when join is triggered in kafka stream?
For example join is triggered between 2 streams regrades who is the left/right? or only when message from right/left arrives?
the ...
0
votes
0
answers
11
views
Spring Cloud Stream StreamsBuilderFactoryBeanCustomizer not replacing the ERROR thread
I am using Spring Cloud Stream with Kafka Stream binder. I want to alter the way deserilization exception and transient error are handled.
To handle deserilization exceptions I added
spring.cloud....
0
votes
0
answers
13
views
Kafka Streams Windowing reduce() not producing output
When trying to execute different operations over a Kafka Stream, windowing doesn't work. The first map() and filter() functions work, but after that the program just doesn't do what it's supposed to ...
0
votes
0
answers
11
views
How to block kstream-ktable join until ktable gets updated in Kafka streams
I want to perform a left join on an updated kTable.
I have a kstream with input Topic A and a materialized kTable with input Topic B. What I'm doing is I perform a kstream-ktable left join and writing ...
0
votes
0
answers
28
views
Spring Cloud Stream KStream DLQ Issue
Any exception occur in processing should got DLQ, currently this not happening.
I am getting org.apache.kafka.common.errors.SerializationException: Error serializing Avro message
it is not going to ...
0
votes
0
answers
22
views
Kafka Streams: Exception not caught by try-catch block
I am using Spring with kafka Streams. I am trying to simulate an error scenario. As you can see from the code, if I pass the input value as 0 to one of the input topics, I will get an divide by 0 ...
1
vote
1
answer
20
views
Wait for store to load when using KStream.transform
The problem I want to solve starts as a basic one: I want to enrich some stream message with data available in some other stream. When I start the application I want to make sure the secondary stream ...
0
votes
0
answers
15
views
Kafka streams app not streaming/consuming data after idle time
I have a Kafka streams app with Processor API
enter co @Bean
@Primary
public KafkaStreams kafkaStreams() {
log.info("Create Kafka Stream Bean with defined topology");
Topology ...
-1
votes
0
answers
13
views
How to aggregate last 12 months user count using kafka windows?
Hi I am recently started working of kafka streams java api. Can you please hlep on the below use case how we can solve in kafka streams.
We are looking for distinct users count who registered in last ...
0
votes
0
answers
23
views
KTable vs KSqlDb
i'd like to understand difference between KTable and KsqlDb. I need two data flows from my "states" topic:
Actual snapshot of a state as key-value store
Subscription to events of state data ...
0
votes
1
answer
23
views
Is it possible to create a multi-binder binding with Spring-Cloud-Streams Kafka-Streams to stream from a cluster A and produce to cluster B
I want to create a Kafka-Streams application with Spring-Cloud-Streams which integrates 2 different Kafka Clusters / setups. I tried to implement it using multi-binder configurations as mentioned in ...
0
votes
0
answers
30
views
KStream left join doesn't produce record on left side null
I have the following left join to sink code, which should have created a record when healthStream got a message, but it's only producing a record when both streams have a message.
val cncStream = ...
0
votes
1
answer
18
views
Kafka Stream multiple punctuator execution per iteration
I have a transformer with a schedule
context.schedule(scanFrequency, PunctuationType.WALL_CLOCK_TIME, new MyPunctuator(stateStore));
Then my Punctuator class
public class MyPunctuator implements ...
0
votes
1
answer
22
views
Improve performance by using Flink instead of Kafka Streams when Source and Sink are in Kafka?
Assuming I have input data coming in via Kafka topics, and output data to be sent to Kafka topics as well, under what circumstances would Flink be able to process data faster than Kafka Streams? At ...
0
votes
0
answers
18
views
Kafka Streams SpringBoot binder perform actions after branching
I'm using kafka streams in springboot with bindings.
I have to read from one topic and depending of the data I have to write in one of three topics.
for do that I used this code.
@Bean
public Function&...
1
vote
1
answer
47
views
A Kafka Stream contains List<Employee>, I want to create a unpacked output stream with Employee objects only
The Kafka stream contains a List<Employee> and I want to process the stream as the new Kafka Stream will contain the all Employee from the List<Employee>.
In nutshell I want to unpack the ...
0
votes
1
answer
26
views
Kafka streams application reset globalTable and input topics
I am building a kafka-streams application with the intent of horizontal scaling and data reprocessing in case of business logic failure.
The application consumes data from two topics that have the ...
0
votes
0
answers
15
views
Do valueTransformer in kafka stream can have access to tombstone?
is it possible to have access to the tombstone that an upstream Ktable process ?
Let say I would like to apply a logic to every deletion that happen to an upstream KTable, that is, do ValueTransformer ...
0
votes
0
answers
22
views
Kamon throws java.lang.ClassNotFoundException on Kafka instrumentation class
I'm working on a Kafka streams application written in scala 2.13 project with the following Kamon packages:
libraryDependencies += "io.kamon" %% "kamon-datadog" % "2.5.4"
...
-1
votes
1
answer
20
views
How to properly create KsqlDB stream based on Kafka topic?
I'm attempting to create a KsqlDB stream (using KSQL CLI) in the following way:
CREATE STREAM orders_stream (
OrderId BIGINT,
Description VARCHAR
) WITH (
KAFKA_TOPIC = ‘orders’,
...
0
votes
0
answers
12
views
Can a kafka Stream join have more that two topics that can be joined together
I have the below example which will consume these two topics -> "test-topic-A" and "test-topic-B" perform a join and then output to a new streamed topic -> "test-topic-...
0
votes
0
answers
44
views
Does Spring-kafka has inbuilt mechanism to restart the kafka stream when the stream transitions into an ERROR state?
We are using spring-kafka for our Kafka Streams application. We currently have a kafkaStreamCustomizer written that listens to stream state transition changes and when it goes into ERROR state, we go ...
0
votes
0
answers
17
views
KTable data flush after some time
I have one KTable and using that I'm doing a join. My join works well with the KTable and KStream. But when the data gets older, like 2-3 weeks, then the data is not found from KTable, which is why my ...
0
votes
0
answers
24
views
ProcessorContext schedule only executed once
I'm using Apache Kafka Stream where I added a transform in my stream
final StreamsBuilder streamsBuilder = new StreamsBuilder();
final StoreBuilder<KeyValueStore<String, byte[]>> ...
0
votes
0
answers
19
views
Kafka Streams flatMapValues returns only one event in output
I am trying to split an incoming message to outgoing messages 1:N using flatMapValues() per docs here-> https://kafka.apache.org/20/javadoc/org/apache/kafka/streams/kstream/KStream.html#...
0
votes
0
answers
24
views
Is there a way to exclude jackson library from kafkaStreams?
When i am trying to exclude jackson library from a Maven dependency in my pom.xml, and trying to run the streaming application, I am getting an error on stateDirectory.java from kafkaStreams as it ...
0
votes
0
answers
24
views
Accessing Kafka Streams Metrics Programmatically
I'm using the following code to export metrics from Kafka Streams application, this works however I don't fully understand how it works. This page from confluent docs explains how to collect metrics
...
0
votes
2
answers
30
views
How can Kafka Streams be used for Event sourcing?
I read about how event sourcing can be achieved by using Apache Kafka as the event broker. (Link to the confluent article)
If we take a look at this picture, it shows how event is written into Kafka, ...
0
votes
0
answers
25
views
Stream data from Heroku Postgresql to AWS Redshift
I have this task of streaming changes made to a table in Heroku Postgresql to Redshift. I tried using examples in this article https://aws.amazon.com/blogs/database/stream-changes-from-amazon-rds-for-...
1
vote
0
answers
21
views
Kafka interactive queries returns store that doesn't contain key
I have a Kafka Streams application that utilizes Interactive Queries.
The application stores key-value pairs in a state store and exposes a REST endpoint that allows querying the local and remote ...
0
votes
0
answers
73
views
Odd behaviour on repartitioning existing data in Kafka
We have the following situation:
Existing topic with 9 partitions in Kafka contains multiple record types. These are partitioned according to a custom header (key = null) which is basically a string ...
0
votes
0
answers
16
views
Is there any way to identify whether the windowedBy time is expired or not in kafka streams
outerjoinResult.toStream()
.groupByKey(Grouped.with(Serdes.String(), Serdes.String()))
.windowedBy(SessionWindows.with(Duration.ofMillis(2000)).grace(Duration.ofMillis(...
0
votes
1
answer
27
views
Enrich kafka stream from different topics using different keys
How can I join a stream with another topic with different keys?
All topics for all tables/streams below are being sourced by DB tables using Kafka connect.
My app is a Spring Cloud Stream with Kafka ...
-1
votes
0
answers
17
views
Kafka stream "revert" message
I use Spring cloud stream with kafka stream to do some processing and somewhere in the middle of the processing I need to save the intermediate result to DB via Hibernate (Spring data jpa).
1.Should ...
0
votes
0
answers
25
views
Quarkus Kafka Streams uses db queries
I want to use Kafka Streams in my quarkus microservice, to read messages from one topic, filter them and redirect message to another topic.
@Inject
Service service;
@Produces
public ...
0
votes
1
answer
22
views
Kafka Streams Monitoring
Is it any solution to monitor Kafka Streams topology? How can we determine state store or global state store is down or when applications has a problem with topology?
0
votes
0
answers
15
views
kafka stream Tumbling Window can't close when no producer sending message
when I use kafka stream Tumbling Window, i found that if the producer send messages to source topic continuously, the stream TimeWindow could work as expected. but once i stopped the producer, the ...
0
votes
0
answers
18
views
Handler method for collecting skipped records for expired window in kafka streams
I have a java application (spring-kafka) with Kafka streams. It streams on a kafka topic using a tumbling window. I have implemented a custom timestamp extractor to keep windowing based on timing ...
0
votes
0
answers
16
views
Kafka kstream left join with Ktable, but stream not always joinASAP
I have below simulation
Source topic A
from source topic A --> use same applicationID, generate two new stream Topic B, C
Create another applicationID
Use Topic B -> generate KTableBuild from B
...
0
votes
1
answer
34
views
Kafka Streams State Store range vs prefixScan
State stores in Kafka Streams have a range(from, to) method, where we can do a range query. Since version 2.8.0 there is also a prefixScan(prefix, serializer) method.
What is the difference between ...
0
votes
0
answers
7
views
KTable with KStreams as base for rule engine
I plan to have microservice with KTable that stores user's object. Objects in it will have a method String rule() that returns for example Groovy shell script in plane text. KTable content will be ...
0
votes
1
answer
21
views
MessageAndMetadata cannot be resolved to a type
The current import that I used for MessageAndMetadata cannot be resolved
this is the import that I have used import kafka.message.MessageAndMetadata; which cannot be resolved to a type.
I want to ...
0
votes
0
answers
37
views
Kafka Streams aggregate write amplification issue
I need to group user notification configs by stock ticker.
e.g. AAPL - {"users": ["a", "b", ...] }
I'm currently using below topology
KTable<> ...
0
votes
0
answers
33
views
Kafka Streams aggregate over time per key- windows not closing if no more events seen for key
I am attempting to build a Kafka Streams application to reduce the volume of events by simply producing the latest event for each key over a period of time (e.g. 1 minute). I cannot guarantee that ...
0
votes
0
answers
29
views
Handle KStream when record missing in GlobalKTable
I'm implementing a KStreme & GlobalKTable Join using Spring. I found a case which needs to be handled differently than what Join does by default.
This is the explanation on the case:
I have a ...
0
votes
0
answers
24
views
Can Kafka streams group by and add a sequence field based on a non-key field
I am new to Kafka streams and finding this problem a bit difficult to navigate.
I have this following code where I am doing a lookup of a GlobalKtable from a kafka stream by doing a leftjoin on key. ...
0
votes
1
answer
57
views
KSQL: How to cast JSON string to raw JSON
I need to copy messages from one Kafka topic to another based on a specific JSON property. That is, if property value is "A" - copy the message, otherwise do not copy. I'm trying to figure ...
0
votes
0
answers
33
views
Kafka streams keep logging 'Discovered transaction coordinator' after a node crash (with config StreamsConfig.EXACTLY_ONCE_V2)
I have a kafka(kafka_2.13-2.8.0) cluster with 3 partitions and 3 replications distributed in 3 nodes.
A producer cluster is sending messages to the topic.
I also have a consumer cluster using Kafka ...