Anyway I investigate this parte because my aggregation process is very slow for 12.000 message took 25 min!
```
@StreamListener(GlobalReportProcessor.GLOBAL_REPORT_TO_PROCESS)
@SendTo(GlobalReportProcessor.GLOBAL_REPORT_PROCESSED)
public KStream<String, GlobalReport> receiveGlobalReportToProcess(KStream<?, Record> input) {
return input
.filter(
(key, value) -> {
boolean isValidRecord = false;
if (value.getGlobalMetadata() != null && value.getGlobalMetadata().getOutputFormatID() != null) {
isValidRecord = true;
} else {
log.error("ERROR receiveGlobalReportToProcess(). Skip the corrupted message: {}", value);
}
return isValidRecord;
}
)
.groupBy(
(key, record) -> record.getGlobalMetadata().getOutputFormatID(),
Serialized.with(null, new JsonSerde<>(Record.class)))
.windowedBy(SessionWindows.with(TimeUnit.SECONDS.toMillis(this.sessionWindowWithSeconds)).until(TimeUnit.SECONDS.toMillis(this.sessionWindowUntilSeconds)))
.aggregate(
GlobalReport::new,
(aggKey, newRecord, aggregateRecord) -> {
aggregateRecord.setGlobalMetadata(newRecord.getGlobalMetadata());
aggregateRecord.setMetadataSend(newRecord.getMetadataSend());
GlobalRecord record = new GlobalRecord();
record.setMetadata(newRecord.getMetadata());
record.setText(newRecord.getText());
aggregateRecord.addRecord(record);
return aggregateRecord;
},
(aggKey, leftPageList, rightPageList) -> {
leftPageList.getRecords().addAll(rightPageList.getRecords());
return leftPageList;
},
Materialized.<String, GlobalReport, SessionStore<Bytes, byte[]>>as(this.stateStoreName)
.withValueSerde(new JsonSerde<>(GlobalReport.class))
)
.toStream()
.filter(((key, value) -> value != null))
.map((key, value) -> new KeyValue<>(key.key(), value));
}
```
windowedBy is 30 seconds with interval.commit 20000 ms and work fine until 6.000 messages with gzip, then it goes slow
I have two kstream process with different aggregation in the same app and for this case I tried to optimize aggregation increased commit.intreval is when I discovered what my app always use config only one of them
Let's see if we can isolate the first issue, where you are experiencing the configuration issue.
@olegz sorry for the delay, and thanks for the response. You’re right, supplier-in-0 can be ignored. I was testing something else (somewhat blindly). I should mention that the message I’m trying to post in the topic is of custom type Message (for which I’ve set up a Serde). I was assuming I needed application/json in order for it to process the deserializing properly. I get no errors, I just don’t see anything show up in my target topic. Or any topic for that matter.
you can temporarily try to add web dependency to ensure the JVM does not exit at all and hopefully, the problem will manifest itself somehow
@olegz :thumbsup:
@sobychacko with the latest version (Hoxton.RELEASE), our avro message conversion is not working. Here is the xception stackTrace for more details:
org.springframework.messaging.converter.MessageConversionException: Could not write JSON: Not an array: {"type":"record","name":"Attachment","namespace":"com.avro.document","fields":[{"name":"url","type":[{"type":"string","avro.java.string":"String"},"null"]},{"name":"type","type":[{"type":"string","avro.java.string":"String"},"null"]},{"name":"description","type":[{"type":"string","avro.java.string":"String"},"null"]}]} (through reference chain: com.avro.document.Document["attachments"]->java.util.ArrayList[0]->com.avro.document.Attachment["schema"]->org.apache.avro.Schema$RecordSchema["elementType"]); nested exception is com.fasterxml.jackson.databind.JsonMappingException: Not an array: {"type":"record","name":"Attachment","namespace":"com.avro.document","fields":[{"name":"url","type":[{"type":"string","avro.java.string":"String"},"null"]},{"name":"type","type":[{"type":"string","avro.java.string":"String"},"null"]},{"name":"description","type":[{"type":"string","avro.java.string":"String"},"null"]}]} (through reference chain: com.avro.document.Document["attachments"]->java.util.ArrayList[0]->com.avro.document.Attachment["schema"]->org.apache.avro.Schema$RecordSchema["elementType"])
at org.springframework.messaging.converter.MappingJackson2MessageConverter.convertToInternal(MappingJackson2MessageConverter.java:286) ~[spring-messaging-5.2.1.RELEASE.jar:5.2.1.RELEASE]
at org.springframework.cloud.stream.converter.ApplicationJsonMessageMarshallingConverter.convertToInternal(ApplicationJsonMessageMarshallingConverter.java:69) ~[spring-cloud-stream-3.0.0.RELEASE.jar:3.0.0.RELEASE]
at org.springframework.messaging.converter.AbstractMessageConverter.toMessage(AbstractMessageConverter.java:201) ~[spring-messaging-5.2.1.RELEASE.jar:5.2.1.RELEASE]
at org.springframework.messaging.converter.AbstractMessageConverter.toMessage(AbstractMessageConverter.java:191) ~[spring-messaging-5.2.1.RELEASE.jar:5.2.1.RELEASE]
at org.springframework.messaging.converter.CompositeMessageConverter.toMessage(CompositeMessageConverter.java:83) ~[spring-messaging-5.2.1.RELEASE.jar:5.2.1.RELEASE]
at org.springframework.cloud.function.context.catalog.BeanFactoryAwareFunctionRegistry$FunctionInvocationWrapper.lambda$convertOutputValueIfNecessary$2(BeanFactoryAwareFunctionRegistry.java:622) ~[spring-cloud-function-context-3.0.0.RELEASE.jar:3.0.0.RELEASE]
at java.base/java.util.stream.ReferencePipeline$3$1.accept(ReferencePipeline.java:195) ~[na:na]
at java.base/java.util.ArrayList$ArrayListSpliterator.tryAdvance(ArrayList.java:1631) ~[na:na]
at java.base/java.util.stream.ReferencePipeline.forEachWithCancel(ReferencePipeline.java:127) ~[na:na]
at java.base/java.util.stream.AbstractPipeline.copyIntoWithCancel(AbstractPipeline.java:502) ~[na:na]
at java.base/java.util.stream.AbstractPipeline.copyInto(AbstractPipeline.java:488) ~[na:na]
at java.base/java.util.stream.AbstractPipeline.wrapAndCopyInto(AbstractPipeline.java:474) ~[na:na]
at java.base/java.util.stream.FindOps$FindOp.evaluateSequential(FindOps.java:150) ~[na:na]
at java.base/java.util.stream.AbstractPipeline.evaluate(AbstractPipeline.java:234) ~[na:na]
at java.base/java.util.stream.ReferencePipeline.findFirst(ReferencePipeline.java:543) ~[na:na]
at org.springframework.cloud.function.context.catalog.BeanFactoryAwareFunctionRegistry$FunctionInvocationWrapper.convertOutputValueIfNecessary(BeanFactoryAwareFunctionRegistry.java:628) ~[spring-cloud-function-context-3.0.0.RELEASE.jar:3.0.0.RELEASE]
at org.springframework.cloud.function.context.catalog.BeanFactoryAwareFunctionRegistry$FunctionInvocationWrapper.lambda$convertOutputPublisherIfNecessary$5(BeanFactoryAwareFunctionRegistry.java:641) ~[spring-cloud-function-context-3.0.0.RELEASE.jar:3.0.0.RELEASE]
at reactor.core.publisher.FluxMap$MapSubscriber.onNext(FluxMap.java:100) ~[reactor-core-3.3.0.RELEASE.jar:3.3.0.RELEASE]
at reactor.core.publisher.FluxPeek$PeekSubscriber.onNext(Fl
we are having these dependencies:
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-stream-schema</artifactId>
<version>2.1.1.RELEASE</version>
</dependency>
<dependency>
<groupId>org.apache.avro</groupId>
<artifactId>avro</artifactId>
<version>${avro.version}</version>
</dependency>
schema registry is migrated to new artifacts/maven coordinates. See this: https://github.com/spring-cloud/spring-cloud-stream-starters/blob/master/spring-cloud-stream-dependencies/pom.xml#L36
wonder if you can change to those.
@sobychacko looks like
Here I have pushed sample project for you. Could you please have a look into it.
https://github.com/msshitole/reactive-avro-source
StreamMessageConverter has been deprecated. Something we are missing. But yes it works fine in the old version.Here I have pushed sample project for you. Could you please have a look into it.
https://github.com/msshitole/reactive-avro-source
I tried adding these registries dependencies but no success. still the same error.
In the above repo, I kept both our current working prod version (Greenwich.SR3) and new. You can test/play by uncommenting it. Incase needed. Just to see working (success) and error cases.
@msshitole I think I found the issue. You need to use spring.cloud.stream.bindings.emit-out-0.contentType=avro/bytes. You were using output as the binding name. Also, this is the right new dependency:
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-schema-registry-client</artifactId>
</dependency>After making those 2 changes, I am able to run the app without errors and see the output.
You don't need to annotate the converter with
StreamMessageConverter any longer.
@sobychacko And one more question,
AvroSchemaMessageConverter constructor has been deprecated. Could you please share the new way of instantiating the same. Thank you!
@sobychacko Additional info,
AvroSchemaMessageConverterclass is present in two dependencies. Both work fine (any one of them). The constructor is only deprecated in spring-cloud-schema-registry-client. Hence we will use the old dependency. Just don't want to upgrade any deprecated version which will be eventually pushed into the prod. Do let us know is any recommendation from your side. Thank you! <dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-stream-schema</artifactId>
<version>2.1.1.RELEASE</version>
</dependency>
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-schema-registry-client</artifactId>
</dependency>
In order to fix the deprecation issue, use this constructor:
new AvroSchemaMessageConverter(MimeType.valueOf("avro/bytes"), new AvroSchemaServiceManagerImpl())
@msshitole You can read the new binding name changes for functional beans here.
Exactly. I was looking for the same. I will go through it. Thank you! :)
I'm running into an issue where I'm getting a ClassCastException invoking my function instead of the data being converted using ApplicationJsonMessageMarshallingConverter. The exception is for trying to cast a Message to my object type instead of converting. If I change the input type to Message<Object>, I can see that the headers include a content type of application/json. One thing I noticed in BeanFactoryAwareFunctionRegistry.doApply() is:
result = this.invokeFunction(this.composed ? input : this.convertInputValueIfNecessary( ... )));Since I have multiple beans that are considered functions functions, I'm using spring.cloud.function.definition=myFunction to specify the desired bean for the input, but this marks the function as composed in the above code which doesn't seem to go through the input value conversion. Is there a way to force the type conversion for a composed function? Or a way to specify the bean to use without marking it as composed? (I'm using Spring Cloud Release Train Hoxton.RELEASE)
but I cannot push it to github because I have corporate proxy...
I can attach here zip?
can you actually post your application somehow?
I am trying to understand what triggers your function to be
composed when it's a single function