Is there any circuit breaker pattern supported with latest version of spring cloud stream. For e.g If we want to stop listening to messages and write to a database . I saw a solution here https://stackoverflow.com/questions/39598943/stop-spring-cloud-stream-streamlistener-from-listening-when-target-system-is-do , this will stop the entire application.
1 reply
Hi, I have te following error while consuming a Message from a Kafka Queue using stream functions:
2020-05-21 14:22:39.660 ERROR 19052 --- [container-0-C-1] onfiguration$FunctionToDestinationBinder : Failed to process the following content which will be dropped: Context1{reactor.onNextError.localStrategy=reactor.core.publisher.OnNextFailureStrategy$ResumeStrategy@3da3dc99}
org.springframework.transaction.reactive.TransactionContextManager$NoTransactionInContextException: No transaction in contextThe problem is presented when I use Spring cloud stream functions with mongo. The code to reproduce is the following:
Repository
@Repository
public interface UsersRepo extends ReactiveMongoRepository<User, String> {
}Consumer
@Bean
public Function<Flux<Message<String>>, Mono<Void>> toUpperCase() {
return messageFlux -> { return messageFlux
.map( msg -> {
String msgStr = msg.getPayload();
return new User().setId("1231").setName(msgStr);
})
.flatMap(usersRepo::save)
.then();
};
}If I replace usersRepo::save by a Mono.just(something) the error is not presented. Please, could you help me, what I'm doing wrong? Thanks
I would like to use Spring cloud stream for our upcoming project. We plan to use AWS SQS and SNS. However, it seems there are no binders available (https://spring.io/projects/spring-cloud-stream). Perhaps there is some alternative that I am not aware of at the moment. On Github there is some traction but that is not production ready(e.g. https://github.com/maciejwalkowiak/spring-cloud-stream-binder-sqs) Please advise.
What could be the cause for Dispatcher has no subscribers for channel 'classification-group-provider-1.productMapping-in-0'.? My function looks like
@Configuration
class ProductMapper(
private val mappingService: MappingService
) {
@Bean
fun mapClassificationGroup(): Function<Flux<Product>, Flux<Message<ProductOfSource>>> = Function { products ->
products
.concatMap { handleProduct(it) }
.onErrorStop()
}
}and my configuration like
spring:
cloud:
function:
definition: mapClassificationGroup
stream:
bindings:
mapClassificationGroup-in-0:
destination: product-updates
group: ${spring.application.name}
mapClassificationGroup-out-0:
destination: products-of-source
producer:
partition-key-expression: headers['productId']
partition-count: 32
kafka:
binder:
consumer-properties:
key.deserializer: org.apache.kafka.common.serialization.StringDeserializer
producer-properties:
key.serializer: org.apache.kafka.common.serialization.StringSerializerto me it looks like all my other services and I cannot spot the error :-S
Hi! I'm trying to use the new approach for testing (with the test-binder) and I have no idea how to connect it with my production code? I have my component that sends the event and I want to test if the event was sent successfully. I can use the OutputDestination to receive the message, however, the InputDestination is not related to the MessageChannel object so it's not possible to inject it into the production component. Any hints?
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?
2 replies