Hi All, Apologies if this not the right forum
We are using spring-cloud-stream:1.2.2.RELEASE along with spring-cloud-starter-stream=kafka:1.2.1.RELEASE
The issue we are facing is sometimes our consumer simply stops receiving messages, (FYI we don't use reactor or ASYNC) our code is as simple as this
@StreamListener(ImageChannels.INPUT)
public void handle(Map<String, Object> map) {
Event event = Event.fromMap(map);
log.trace("Received Event {}", event);
if (imageFilters.test(event)) {
governorService.handleEvent(event);
}
}Upon restarting the service, it resumes receiving messages. We also don't see any exceptions when this happens. I read somewhere that if the consumer does not send back an ACK then no new messages are delivered, is there a way to check that ? or a way to always send ACK on receiving a message ?
Any suggestions on how to debug this is appreciated
The acks are sent by default so I don't think that's the problem. Many times, problems like this are caused by threads being "stuck" somewhere in user code. Next time it happens take a thread dump to see what the container threads are doing.
That said, 1.2.x is very old; it is based on spring-kafka 1.1.x which is no longer supported. I would recommend upgrading to at least spring-cloud-stream 1.3.x and override the spring-kafka version to 1.3.9. This version uses a much simpler threading model thanks to KIP-62. With older versions, we had to have 2 threads, one to keep the consumer "alive" and one to call the listener. If the listener gets behind, the consumer thread pause()s the consumer and resume()s it when the listener catches up. Pause/resume allows us to keep the consumer alive without retrieving more records on each poll().
I have seen problems in the past where the resume() did not take effect and resulted in the behavior you see. spring-kafka 1.3.x gets rid of this complexity because the kafka-clients now keeps the consumer "alive" on its own, with a background heartbeat mechanism.
Hi @sobychacko sure, here we go: using Spring Boot 2.0.0.M4/BUILD-SNAPSHOT and scs 3.0.0.M2 I have set upspring.cloud.stream.kafka.streams.binder.brokers=someURL. I end up having this
WARN log: Connection to node -1 (localhost/127.0.0.1:9092) could not be establishedI've seen that in KafkaStreamsBinderSupportAutoConfiguration.streamConfigGlobalProperties() the KafkaStreamsConfiguration kafkaStreamsConfigurationparameter contains a localhost:9092 value for bootstrap-servers that comes from the default Spring boot inicialization of org.springframework.boot.autoconfigure.kafka.KafkaProperties, so it is not overriden by the binder config (if (ObjectUtils.isEmpty(properties.get(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG))) {). To work around this I had to add spring.kafka.bootstrap-servers=someURL
Caused by: java.lang.IllegalArgumentException: 'avroSchemaServiceManager' cannot be null
at org.springframework.util.Assert.notNull(Assert.java:198) ~[spring-core-5.2.0.BUILD-SNAPSHOT.jar:5.2.0.BUILD-SNAPSHOT]
at org.springframework.cloud.stream.schema.avro.AvroSchemaRegistryClientMessageConverter.<init>(AvroSchemaRegistryClientMessageConverter.java:171) ~[spring-cloud-stream-schema-3.0.0.BUILD-SNAPSHOT.jar:3.0.0.BUILD-SNAPSHOT]
at org.springframework.cloud.stream.schema.avro.AvroMessageConverterAutoConfiguration.avroSchemaMessageConverter(AvroMessageConverterAutoConfiguration.java:63) ~[spring-cloud-stream-schema-3.0.0.BUILD-SNAPSHOT.jar:3.0.0.BUILD-SNAPSHOT]AvroMessageConverterAutoConfiguration.avroSchemaMessageConverter() the autowired of the following properties is null (this worked in 3.0.0.M2): @Autowired
private AvroMessageConverterProperties avroMessageConverterProperties;
@Autowired
private AvroSchemaServiceManager avroSchemaServiceManager;
Hi, I have a question regarding stateful message processing using the cloud function across multiple messages. To me it seems, it is impossible to do something like because every message becomes its own flux.
@Bean
public Consumer<Flux<Message<?>>> sink() {
return (input) -> input
.buffer(Duration.ofSeconds(10))
.doOnNext(System.out::println);
}Am I right with this assumption?
Yes, thanks a lot for the input.
The ability to reason about a stream of messages over time is a key feature for cloud stream in my opinion.
This would put cloud stream in a good spot, where full blown streaming solutions like flink are not needed, because spring cloud stream is capable enough to handle some complex scenarios.
And for me this was what spring-cloud-reactive promised.
So if the stream of messages in cloud function is not a modeled as a single Flux, imo we lose most of the advantages brought by the reactive API right now.
To make the binders reactive, couldn't we put a Flux Consumer above a PollableConsumer, which would act as a bridge?
I think the is a major concern, so I'm thinking about opening an issue for this topic. Do you think this will be the right approach?
@Component
@EnableBinding(PolledConsumer::class)
class ReactiveConsumer(
val service: DummyService,
val consumer: PolledConsumer
) : ApplicationRunner {
private val logger = KotlinLogging.logger { }
override fun run(args: ApplicationArguments) {
Flux.create<List<DummyElement>> { sink ->
sink.onRequest {
logger.info { "requested $it elements" }
val elements: MutableList<DummyElement> = mutableListOf()
for (i in (0..it.toInt())) {
consumer.destIn().poll({
val element: DummyElement = it.payload as DummyElement
elements.add(element)
}, object : ParameterizedTypeReference<DummyElement>() {})
}
sink.next(elements)
}
}
.subscribe(BackpressureSubscriber(service))
}
private class BackpressureSubscriber(val service: DummyService) : BaseSubscriber<List<DummyElement>>() {
override fun hookOnNext(elements: List<DummyElement>) {
elements.toFlux()
.flatMap { service.process(it) }
.blockLast()
request(10)
}
override fun hookOnSubscribe(subscription: Subscription) {
request(10)
}
}
}
interface PolledConsumer {
@Input("input")
fun destIn(): PollableMessageSource
}
Hi! I'm using Cloud Stream with Kafka Streams, and i've got a question about KTable. When my binder is a KStream type the spring cloud stream parse my JSON payload (value) to my Object automatic, but when i'm use KTable as input binder occurr an error that say "java.lang.ClassCastException: class java.lang.String cannot be cast to class br.com.wspot.accounting.streamprocessor.models.RawAccounting", bellow follow my code and my application.properties:
## This example works great!
@EnableBinding({Stream.AccountingProcessor.class })
public class Stream {
@StreamListener
public void process(@Input(AccountingProcessor.INPUT_ACCT) KStream<String, RawAccounting> rawAccountingKStream) {
rawAccountingKStream.foreach((key, value) -> System.out.println(value));
}
public interface AccountingProcessor {
String INPUT_ACCT = "inputAcct";
@Input(AccountingProcessor.INPUT_ACCT)
KStream<String, RawAccounting> rawAcct();
}
}## This example throw the follow exception: java.lang.ClassCastException: class java.lang.String cannot be cast to class br.com.wspot.accounting.streamprocessor.models.RawAccounting
@EnableBinding({Stream.AccountingProcessor.class })
public class Stream {
@StreamListener
public void process(@Input(AccountingProcessor.INPUT_ACCT) KTable<String, RawAccounting> rawAccountingKStream) {
rawAccountingKStream.toStream().foreach((key, value) -> System.out.println(value));
}
public interface AccountingProcessor {
String INPUT_ACCT = "inputAcct";
@Input(AccountingProcessor.INPUT_ACCT)
KTable<String, RawAccounting> rawAcct();
}
}## My application.properties
spring.cloud.stream.kafka.streams.binder.configuration.commit.interval.mms=1000
spring.cloud.stream.kafka.streams.binder.configuration.default.key.serde=org.apache.kafka.common.serialization.Serdes$StringSerde
spring.cloud.stream.kafka.streams.binder.configuration.default.value.serde=org.apache.kafka.common.serialization.Serdes$StringSerde
spring.cloud.stream.bindings.inputAcct.destination=wspot.accounting.raw
spring.cloud.stream.bindings.inputAcct.content-type=application/json
spring.cloud.stream.kafka.streams.bindings.inputAcct.consumer.application-id=wspot-acct-raw-processor-id
spring.cloud.stream.kafka.binder.brokers=localhost:9092maybe i missed some concept, is the first time i'm use Spring cloud Stream.