Questions tagged [spring-cloud-stream]

Spring Cloud Stream allows a user to develop and run messaging microservices using Spring Integration and run them locally, or in the cloud, or even on Spring Cloud Data Flow. Just add @EnableBinding and run your app as a Spring Boot app (single application context). You just need to connect to the physical broker for the bus, which is automatic if the relevant bus implementation is available on the classpath.

Filter by
Sorted by
Tagged with
0 votes
1 answer
28 views

@StreamListener doesn't work for secured Kafka streaming, TopologyException

My demo application has: application.yml # some values to reduce duplicates in configuration kafka-conf: server: kafka.host:9093 topic-name: employee.test.incoming.json keystore: classpath:...
Sergii's user avatar
  • 7,134
0 votes
1 answer
19 views

Bind a consumer queue name to an existing exchange

I'm new to spring cloud and integration, and I'm struggling to do what I would think ought to be pretty straightforward. I was able to do it by declaring the bindings in annoatations, however I need ...
Saieden's user avatar
  • 45
0 votes
0 answers
9 views

Is it possible to pause SCS Kafka so it doesn't consume messages when application is not ready/healthy?

I am using Spring Cloud Streams with a Kafka binder, and deploying the application to Kubernetes. This includes the default K8S readiness and liveness probes. For HTTP traffic, the K8S readiness probe ...
Pete's user avatar
  • 1
-1 votes
0 answers
9 views

How to publish to a separate topic In a spring cloud streams?

There is main funtional flow for intra day processing and which is processed via kafka-streams binder. As a peripheral, when an API is invoked within the application. A message should be pushed to a ...
Karthik Suresh's user avatar
0 votes
0 answers
7 views

Spring cloud functional style and integration-test

I have a Spring Boot 3.1.6 project using Spring Cloud 2022.04 and Spring Cloud Stream 4.0.4. It is setup so the integration-test consumes events produced by the application to verify them. Like so: ...
Christian Landbo's user avatar
-1 votes
0 answers
12 views

Process Millions of records with spring batch with remote chuncking and then publish spring cloud stream

I have a use case to process millions of records. Since the processing is done on one pod. When the pod restarts, the messages are lost Looking for best approach by using spring batch and remote ...
user1421204's user avatar
0 votes
0 answers
35 views

cloud stream v4 (reactive) - unable to use simple transform function using springboot 3

I am new to spring cloud stream. I am not able to use the transform function (consume message, transform, produce message). Consumer function works well. @Bean public Function<String, ...
bruce reed's user avatar
0 votes
1 answer
41 views

Is it possible for Spring Cloud Stream to use both old @EnableBinding-@StreamListener and new functional style in one project?

Below is the code with both types of kafka consumers which I try to use . @SpringBootApplication @EnableBinding(MixApplication.IncomingBinding.class) @Slf4j public class MixApplication { public ...
Oleksii Honcharenko's user avatar
0 votes
0 answers
27 views

RetryListener not bound to consumer's RetryTemplate in Spring Cloud Stream functional model

We have developed a RetryListener to log exceptions between every retry when we are consuming messages off of a stream and exceptions occur. I am trying to bind this listener into our consumer but am ...
Keith Bennett's user avatar
0 votes
1 answer
25 views

Spring Cloud Stream Rabbit Binder concurrency

I have one consumer microservice that handles messages from other producer microservice. I managed event driven structure with Spring Cloud Stream and RabbitMq Binder. I want to limit maximum ...
user725455's user avatar
0 votes
0 answers
21 views

Spring Cloud Stream notify Producer that message is consumed

I have several spring boot applications that uses spring cloud stream for messaging. I use rabbitmq as binder. I am trying to send a message from one microservice to another. In Producer side, before ...
user725455's user avatar
0 votes
0 answers
10 views

2 versions of amazon-kinesis-client library in one project

I have a project that use the "software.amazon.kinesis:amazon-kinesis-client:2.5.0" library for quite some time for consuming events from kinesis stream. Recently we started using "org....
Lucas1983's user avatar
0 votes
0 answers
17 views

Use confluent schema registry along with spring cloud binding

I want to use enable schema registry and it works fine if I pass all schema related configs via spring.kafka path in yaml But when i use spring.cloud path, the schema registry related configs are not ...
Rahul's user avatar
  • 7
0 votes
1 answer
30 views

Naming Convention for using source in StreamBridge SPRING CLOUD STREAM

im using streamBridge in 2 different locations like this: streamBridge.send("first", "hello1"); streamBridge.send("second", "hello2"); now the problem is that ...
alpha's user avatar
  • 23
0 votes
2 answers
50 views

Spring Data Flow streams with multiple inputs

I use Spring Cloud Data Flow 2.11.2. My goal is making two sources and one processor that accepts both messages as input and performs some operations on them, e.g.: stream view I found a guide on this ...
Roman's user avatar
  • 3
0 votes
0 answers
19 views

spring cloud stream failed with IllegalStateException: deploymentId doesn't match expected state

I have been engaging with the Spring Cloud ecosystem for the past 3 months. We have created different streams with both prebuilt and custom Spring Cloud Stream apps. During this period, we have come ...
Udhaya's user avatar
  • 117
0 votes
0 answers
45 views

Propagate traceparent header when using recordMetadataChannel

I use spring boot 3. After sending message to kafka, using spring cloud stream, I would like to log offset and partition of sent message with the traceId from the thread which sent the message. Here's ...
Fedor Konstantinov's user avatar
1 vote
0 answers
27 views

Limit number of concurrently processed messages with the Spring Cloud Stream Kafka Reactive binder

Currently we are migrating our applications to the Spring Cloud Stream Kafka Reactive binder and experience very high memory usage. An incoming message causes all kind of processing here so we'd like ...
kschlesselmann's user avatar
0 votes
1 answer
51 views

Getting java.lang.ClassCastException on consuming an event payload of generic type

I have Spring Boot 3.1.1 kotlin application which is using Spring Cloud 2022.0.4 to consume Kafka events. consumer function: @Bean fun ipConsumer(): Consumer<Message<...
Faizan's user avatar
  • 383
0 votes
1 answer
56 views

Unresolved reference: Source - org.springframework.cloud.stream.messaging.source

I am updating some project libraries because of vulnerabilities and after this updates I can't meke this import work: import org.springframework.cloud.stream.messaging.Source ("Source" is ...
Rysic Rychu's user avatar
0 votes
0 answers
38 views

Kafka producer doesn't send messages after Spring-Cloud and Spring boot upgrade

After upgrading Spring Boot to version 3.0.6 and Spring Cloud to 2022.0.4 and java 17, springBootVersion=3.0.6,springDocVersion=1.67, springCloudVersion=2022.0.4,springCloudFunctionVersion=3.2.9 the ...
sekhar's user avatar
  • 57
0 votes
0 answers
22 views

Configuring Kafka Consumer Group for Multiple Topics Assignment Issue

I need to consume data from two different topics, enrich that data, and then send it to two separate topics. I've provided the applicationId at the binder level as I need to provide same groupId for ...
Rajesh kumar's user avatar
0 votes
1 answer
142 views

What happened to test-binder in Spring Cloud Stream / Spring Boot 3?

I'm failing to find any documentation on this, but what happened to the test-binder for Spring Cloud Stream / Spring Boot 3? This use to work fine with 2021 bom imports, but now get this error: Could ...
Brandon's user avatar
  • 424
0 votes
1 answer
52 views

Refreshing Kafka Producer during runtime in spring application

I'm using spring cloud stream with Kafka (Spring Boot 3.0.11, Spring Cloud 2022.0.4), sending messages with a StreamBridge. Kafka broker requires using a client certificate. Since we already have ...
Harry Developer's user avatar
-2 votes
1 answer
54 views

RabbitMQ Super Streams | Spring Cloud Stream Consumer Groups using RabbitMQ

I have 2 services (service A and Service B) which need to consume same data. I can achieve this by using queue type as stream. But at the same time, during auto-scaling I dont want service A1 and ...
jarvis's user avatar
  • 25
0 votes
0 answers
60 views

Failed to send Message to channel - using spring cloud stream binder kafka

upgraded springBoot 2.6.6 to 3.x , springCloud 2021.0.2 to 2022.0.x ,springCloudFunction 3.2 to 4.0 now junit is getting failed having Error - org.springframework.messaging.MessageDeliveryException: ...
Saptarshi Sam's user avatar
0 votes
0 answers
78 views

Programmatically create\define Spring Cloud Stream bindings and Function

I am looking for a way programmatically to create Function and binding of function to existing Channel (Channel will be added) Ideally would be in event "BindingCreatedEvent" to create ...
barsik0's user avatar
0 votes
1 answer
46 views

How to correctly configure errorChannel in spring cloud (spring integration)?

I have a webapp in which I use Kinesis Stream with spring-cloud-stream, spring-cloud-stream-binder-kinesis and spring-cloud-starter-sleuth. What I would like to do is to configure the error handling ...
AM13's user avatar
  • 683
0 votes
1 answer
58 views

Spring Cloud Stream Producer sends message but kafka broker not running

Spring cloud stream (kafka) producer is does not throw exception when kafka broker is down. boolean sent = streamBridge.send("sometopic", deliver); log.info("-------------> message ...
Pascoal Eddy Bayonne's user avatar
0 votes
0 answers
56 views

Unable to Receive Kafka Messages in Spring Cloud Stream Consumer with Java 17

I am working on a Spring Cloud Stream application that utilizes Apache Kafka for messaging. In the application, I have set up a REST API to publish messages to a Kafka topic. However, the consumer ...
Mohamed Sid Abdalla Rgagde's user avatar
1 vote
1 answer
93 views

Using SendToDlqAndContinue in Function without using processor spring-cloud-stream-binder-kafka-streams

I'm trying to send message to DLQ when the message i'm processing in my Function doesn't respect some logic. For some context, I have this configuration in my application.yaml My function looks like ...
Romain Huber's user avatar
1 vote
2 answers
301 views

Spring cloud stream with RabbitMQ binder throws java.lang.IllegalStateException: Message body is too large

I'm using spring cloud stream 4.0.4 with rabbitmq binder 4.0.4 to proccess messages in spring boot 3.1.3 app. I'm also using rabbitMQ 3.10.0. The thing is, there is an bug in my application when the ...
Jam3si's user avatar
  • 31
0 votes
1 answer
35 views

Functional binding is disabled due to the presense of @EnableBinding annotation in your configuration

I upgraded spring boot to 2.3.0.RELEASE and the spring boot application is stuck at the last line and it never starts. Any help is appreciated, Thanks! 09:26 23:57:52.276 [main ] [WARN] [...
user923499's user avatar
0 votes
1 answer
37 views

AWS Kinesis long startup waiting

I'm using AWS Kinesis with spring cloud to read data from a data event hub. The deploy end with the following log: 2023-09-26 14:07:30.476 INFO [,,] 9331 --- [esis-consumer-1] a.i.k....
AM13's user avatar
  • 683
0 votes
0 answers
34 views

Events order for Spring Cloud Stream Reactive using Kafka binder

We are using Spring Cloud Stream reactive with the kafka binder and we would like to ask if it can be possible to consume multiple events from the same partition in parallel. See the code sample ...
Diyan Yordanov's user avatar
1 vote
1 answer
99 views

Spring Cloud Stream Kinesis Binder 4.0 doesn't support Multi Stream with KCL 2 Enabled

KCL 2 allows multi stream support with the same consumer group in the same application. With cloud stream kinesis binder 4.0, kpl-kcl-enabled: true A single stream in the destination works. bindings: ...
siddharth Jain's user avatar
0 votes
1 answer
59 views

SpringCloudStream function.definition with SpringBoot 3

After Upgrading spring from 2.x to 3.x the resolving of configuration variables somehow changed. In SpringBoot 2.x.x the variable spring.cloud.stream.function.definition could be resolved by Spring to ...
GJohannes's user avatar
  • 1,473
0 votes
0 answers
36 views

globally enable RabbitMQ quorum queues in Spring Cloud Stream

Is there any possibility to define every queue to be provisioned as a quorum queue instead of a classic queue? spring.cloud.stream.rabbit.bindings.<channelName>.consumer.quorum.enabled=true is ...
hotzen's user avatar
  • 2,820
1 vote
0 answers
41 views

KafkaStream reaches ERROR state on rolling upgrade of brokers

I am using Kafka 2.8.1 (AWS MSK). I am observing whenever some rolling upgrade is done by AWS my Kafka stream application reaches an ERROR state and I keep getting the below exceptions on trying to ...
tuk's user avatar
  • 6,067
0 votes
0 answers
163 views

Spring cloud stream kafka timeout when getting topic information

We have a kafka server in house which we've been using with spring boot applications without an issue. I'm now trying to implement a simple solution for spring cloud stream and I'm getting a timeout ...
codex70's user avatar
0 votes
0 answers
95 views

Migrating RabbitMQ spring-cloud-stream from springboot 2 to springboot 3

I am updating a project that sends rabbitMQ messages from spring boot to to spring boot 3 I THINK the current springboot 2 project uses spring-cloud-stream v3.2.3 to send the messages Here is a ...
javaguy's user avatar
  • 56
0 votes
1 answer
45 views

Spring Cloud Stream Application not connecting to EmbeddedKafka when testing

Attempting to test a basic Spring Cloud Stream application, the EmbeddedKafka broker never get the message published when running the tests. Here is a github repo for my current code : https://github....
Guillaume Jobin's user avatar
0 votes
0 answers
17 views

Spring cloud kafka stream Retry with cercit breaker

I have a 3 partition kafka topic to handle events and push the messages to exexternal service. This is done by spring cloud kafka binding with functional approach in a kubernetes cluster. In happy ...
PRITAM MOHAPATRA's user avatar
0 votes
1 answer
38 views

Stream deployment properties are being locked after a deployment with errors

We're using spring cloud dataflow 2.10.3 deployed using helm in a kubernetes cluster. I'm not sure if this is a bug or something were doing wrong but we've seen it happen across all types of streams, ...
oldvirus_231's user avatar
0 votes
1 answer
30 views

Redis serializer properties in SCDF

I'm using a splitter and an aggregator in SCDF 2.10.3 and I want to cache messages in Redis. Here is a simple example stream : http | splitter | aggregator | log This stream works fine without Redis. ...
CEDDM's user avatar
  • 21
0 votes
1 answer
161 views

Spring Cloud Stream 4.0.4 | Kafka Producer | Functional Style

I have come across below example of writing Kafka Producer with Spring Cloud Stream in functional style. Dependency <dependency> <groupId>org.springframework.cloud</groupId> &...
Half Blood Prince's user avatar
0 votes
1 answer
30 views

Consumer and Producer Concurrency configuration in spring cloud stream kafka binder?

spring.cloud.stream.binding.myInputProcess.consumer.concurrency: 3 I have this configuration in my application, which takes input from kafka and process it and puts back to another kafka. So, do I ...
Siddhant Srivastava's user avatar
0 votes
1 answer
76 views

Spring-cloud-stream Solace connectivity issue

In one of my Java-Spring boot based micro-service, i have the Solace integration setup for messaging. I am using the spring-cloud-stream and Solace binder to connect the Solace. Overall the ...
learner4life's user avatar
0 votes
0 answers
23 views

spring-cloud-stream-binder-kinesis - kinesis-hashcode added as a prefix to the error stream [Java]

I am trying to implement an error handling. To test it out I throw an exception from mu Function method like this: @Configuration @EnableIntegration class MessageProcessorConfig { @Bean ...
lucek091's user avatar
0 votes
0 answers
63 views

ClassCastException testing Spring Cloud Stream producer only when another test is also executed

New to Spring Cloud Stream, using the latest 3.1.3, I am attempting to successfully write one of each type (Supplier, Function, Consumer) and matching tests, just so that I get a good grasp of the ...
Guillaume Jobin's user avatar

1
2 3 4 5
56