it doesn't matter. In my production code, I'm attaching to MessageChannel - that's the main stream object I use to invoke send method. Then I use for example MessageCollector to check if the proper message was send. InputDestination and OutputDestination are useless, as I cannot link them with my production code.
For the most part, they all use the new test binder.
It's not the whole codebase, but let's assume that in my production code I have something like this: https://github.com/jkubrynski/cloud-stream-sample/blob/master/src/main/java/com/kubrynski/streamdemo/permission/PermissionService.java
then in the test I want to check if after invoking revokeAccess method, the proper message was sent to the proper destination. I don't mean testing the business logic - just the integration. If the proper destination, message headers and a valid payload were set.
with the old test-support I use
messageCollector.forChannel(myChannel.myDestination()).poll(1, TimeUnit.SECONDS);
then on the retrieved message I can verify the whole structure. The question is how similar approach is possible with the new test-binder
if the above is not clear let me know -> I'll code the whole approach in the separate project
@sobychacko basic question on consuming messages from Kafka. How do I choose between KafkaListener, StreamListerner, Function etc. I may optionally write to a retry topic but I can also live with same topic by not committing the offset for failed messages. We use Kerberos Auth and Kafka Broker version 0.10.2 and have no plans on upgrading to 0.11 atleast for the next 1 year.
@sakella1986 That is a very old Kafka version and it won't work with the latest versions of the Kafka binder in Spring Cloud Stream (You will likely see some Kafka wire protocol compatibility issues if you combine 0.10.2 and latest SCSt versions using later Kafka clients). We used to provide a separate 0.11 binder when the default version supported was 0.10.2/0.10.1. Those binders are no longer supported.
With that said,
With that said,
KafkaListener is part of the Spring Kafka library and you don't need Spring Cloud Stream to use that. You use StreamListener or functional support from Spring Cloud Stream when you need the higher-level abstractions and programming model provided. In addition, SCSt comes with a lot of opinionated choices you can rely upon. SCSt Kafka binder is built upon the foundations of Spring Kafka although it doesn't directly use KafkaListener internally.
1 reply
But you can focus on just the tests you've shared with me. Normally the selected lines would be in the production code. How can you move them from this test to
main sourceset: https://github.com/spring-cloud/spring-cloud-stream/blob/master/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/RoutingFunctionTests.java#L72;L77
@jkubrynski you can also try to use
StreamBridge to send outbound messages. It is relatively a new feature. But available on the recent Horsham SR releases. The idea is that you don't directly deal with channels, but the framework internally handles that for you. You interact with this in terms of a binding destination. OutputDestinaion can be used to test that scenario. See these tests: https://github.com/spring-cloud/spring-cloud-stream/blob/master/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/StreamBridgeTests.java
@sakella1986 I spoke with @garyrussell and you should be fine using SCSt or Spring Kafka there. Based on this compatibility matrix, it should work with brokers 0.10.2. https://cwiki.apache.org/confluence/display/KAFKA/Compatibility+Matrix.
Sorry for the confusion. The modern clients can negotiate with the older wire protocols and therefore you may use the later Kafka binders in Spring Cloud Stream (or Spring Kafka's
Sorry for the confusion. The modern clients can negotiate with the older wire protocols and therefore you may use the later Kafka binders in Spring Cloud Stream (or Spring Kafka's
KafkaListener).
2 replies
Hi, I'm trying to use this implementation https://github.com/spring-cloud/stream-applications/blob/master/functions/supplier/rabbit-supplier/src/main/java/org/springframework/cloud/fn/supplier/rabbit/RabbitSupplierConfiguration.java, in a kafka environment, I'm using dataflow and it is setting
--spring.cloud.stream.bindings.output.destination=someTopic yet the messages are published to the topic rabbitSupplier-out-0, did I miss something in the implementation?, is there a way of saying that the FluxMessageChannel is the default output and topic bindings apply to it?
Hi @olegz ,
I have created a simple Spring cloud stream apps(source, processor and sink) where the processor and sink are transferring reactive data.
https://github.com/sumit1912/springcloud.stream.pipeline
I deploy these apps in Spring Cloud Data Flow with RabbitMQ as broker. My apps work fine and transfer data when I use, springBootVersion = "2.1.9.RELEASE" and org.springframework.cloud:spring-cloud-stream-reactive:2.2.0.RELEASE dependency.
When I update Spring boot to v2.2.6.RELEASE and replace spring-cloud-stream-reactive by "org.springframework.cloud:spring-cloud-dependencies:Hoxton.SR3" and "org.springframework.cloud:spring-cloud-stream-dependencies:Horsham.SR3" I got below error:
ERROR 5206 — [ wer.http.wer-1] o.s.integration.handler.LoggingHandler : org.springframework.messaging.MessageHandlingException: error occurred in message handler [org.springframework.cloud.stream.binding.StreamListenerMessageHandler@39ce27f2]; nested exception is org.springframework.messaging.converter.MessageConversionException: Cannot parse payload ; nested exception is com.fasterxml.jackson.databind.exc.InvalidDefinitionException: Cannot construct instance of `reactor.core.publisher.Flux` (no Creators, like default construct, exist): abstract types either need to be mapped to concrete types, have custom deserializer, or contain additional type information
at [Source: (byte[])...The error appears only after I replace spring-cloud-stream-reactive by spring-cloud-dependencies and spring-cloud-stream-dependencies and update Spring boot to 2.2.x
Is it because of the fact that spring-cloud-stream-reactive module is deprecated in spring-cloud-dependencies:Hoxton.SR3 and we need to use spring-cloud-function (Supplier/Function/Consumer model) for handling reactive messages?
Hello Team. I am facing a weird issue. We are using using @StreamListener with condition. We have the condition as
Currently i do not have any more logs to provide more information.Sorry. Does someone have any idea why this could have happened? Is there any sort of setting i need to check in kafka or is it some problem with serializers etc?
new String(payload).event.type matches 'MyEvent-Type'. This works fine in local Kafka, even in embedded kafka(integration-test). But when deployed to other environment for the same message i get Cannot find a @StreamListener matching for message".Currently i do not have any more logs to provide more information.Sorry. Does someone have any idea why this could have happened? Is there any sort of setting i need to check in kafka or is it some problem with serializers etc?
when consumers contentiously revoking from a group since timeout, the group will count less and less consumers until it left 0 consumers, it obvious we need to increase the timeout and improve performance... but still, is there a possibility to join back the consumers to the group without restart the microservice?
Hello, I have the following setup for consumer, it is working for the first message, and then I get an exception:
INFO 47834 --- [container-0-C-1] o.s.c.s.m.DirectWithAttributesChannel : Channel 'application-1.myConsumer-in-0' has 0 subscriber(s).
ERROR 47834 --- [container-0-C-1] o.s.integration.handler.LoggingHandler : org.springframework.messaging.MessageDeliveryException: Dispatcher has no subscribers for channel 'application-1.myConsumer-in-0'.; nested exception is org.springframework.integration.MessageDispatchingException: Dispatcher has no subscribers, failedMessage=GenericMessage [payload={...}]public interface Sink<I> extends Consumer<Flux<I>> {
@Override
default void accept(Flux<I> input) {
consume(input);
}
void consume(Flux<I> input);
}@Component("myConsumer")
public class MyConsumer implements Sink<MyMessage> {
public MyConsumer() {
}
@Override
public void consume(Flux<MyMessage> input) {
input
.transform(...)
.transform(...)
.subscribe();
}
}spring:
cloud:
stream:
function:
definition: myConsumer
bindings:
myConsumer-in-0:
destination: ...
group: myConsumerGroup
consumer:
autoRebalanceEnabled: true
partitioned: true
maxAttempts: 3
kafka:
bindings:
myConsumer-in-0:
consumer:
configuration:
value:
deserializer: ...Not sure what am I doing wrong, any help is appreciated
Hi Team , I am reading through the docs here https://cloud.spring.io/spring-cloud-static/spring-cloud-stream/current/reference/html/spring-cloud-stream.html#spring-cloud-stream-overview-error-handling , how does re-queue of the message work in case of kafka. Does it require any special configuration? I have
maxAttempts: 3 . Here's my sample project https://github.com/natraj09/spring-cloud-stream-playground/blob/master/reactive-processor-kafka/src/main/java/reactive/kafka/ReactiveProcessorApplication.java#L52, if the rest payload has "error:true" it prints an error and throws an Exception. I only see the info log once instead of 3 times.
2 replies
By default, this will create 2 bindings (listener containers). If you set the
multiplex consumer property to true, you'll get a single binding consuming from both destinations.
Hi all. I'm trying to ensure that applications always use round robin partitioning with sending to the DLQ. In the docs it says I should implement a
DlqPartitionFunction bean, but looking at the parameters, it's not clear how I could implement round robin, since I must always return the actual partition number and there isn't any info on the number of partitions for the DLQ topic. Any advise? Reference: https://cloud.spring.io/spring-cloud-static/spring-cloud-stream-binder-kafka/3.0.0.RELEASE/reference/html/spring-cloud-stream-binder-kafka.html#kafka-dlq-processing
1 reply
Hey, I have a question for some of the motivation for the new test style for the test binders.
Unlike all other spring tests, these are no longer leveraging @Autowired and instead using this context builder, like
context = new SpringApplicationBuilder(
TestChannelBinderConfiguration.getCompleteConfiguration(Application.class)
.run("--server.port=0",
"--spring.cloud.function.definition=listener");This is a huge departure from the standard use of @SpringBootTest and @RunWith(SpringRunner.class). does anyone know why this is happening? Is it out of necessity, is the direction that Spring is moving in? I haven't been able to find any documentation or blog posts about the motivation behind this.
It's making rewriting existing tests really challenging, and it feels like a code smell to rewrite the existing tests is a far more verbose way.
1 reply
@garyrussell I tried creating 2 bindings to the same function . The original topic and DLQ topic are both binded to the same function (sample configuration below). I found the message is not being sent to dlq on error https://github.com/natraj09/spring-cloud-stream-playground/blob/master/reactive-processor-kafka/src/main/java/reactive/kafka/ReactiveProcessorApplication.java#L54 . I came across this open issue spring-cloud/spring-cloud-stream#1922 , looks like DLQ is not supported for reactive function. Instead of throwing an exception can I use sendto destination to route it to error channel Mono.just(MessageBuilder.fromMessage(inbound).setHeader("spring.cloud.stream.sendto.destination", "rest-input-error").build()) . What is the recommended pattern to handle errors in case of reactive function?
spring:
cloud:
stream:
function.definition: testProcess;testSource;testSink
bindings:
testProcess-in-0:
destination: rest-input,rest-input-error
group: input-group
maxAttempts: 1
testProcess-out-0:
destination: rest-output
testSink-in-0:
destination: rest-output
group: sink-group
testSource-out-0:
destination: rest-input
kafka:
bindings:
testProcess-in-0:
consumer:
enableDlq: true
dlqName: rest-input-error
feel free to raise the issue there
@sobychacko and all, quick question on improving kafka message consumer performance. we have 12 partitions in topic, and publishing at a rate of 200 TPS. after receiving the message (we use 12 sprintboot containers for reading from all partitions), we need to make an external vendor call which takes 600 ms which means I can only support 20 TPS. What options I have to achieve 200 TPS message consumption. Appreciate your help!
20 replies
