I have an input message with an identifier, with that identifier I perform a query against a database and retrieve a list of data. I send each of these elements to the next stream via context.forward; but there comes a time when it stops processing and an error appears:
org.apache.kafka.streams.errors.StreamsException: task [0_1] Abort sending since an error caught with a previous record to topic "group.outputs" due to org.apache.kafka.common.errors.RecordTooLargeException: The request included a message larger than the max message size the server will accept.
Simplifying the code it would be similar to:
// ProcessContext:
private ProcessorContext context;
...
private void handler(List<String> parameters) {
for(String param : parameters) {
byte[] key = ....; // calculate key
byte[] value = ...; // calculate value
To to = .kafkaMessage..outputstream..;
context.forward(key , value , to);
context.commit();
}
}
- I don't understand the problem, can't multiple messages be sent one after the other?
- Should the commit do it after processing the entire list?
- Why doesn't it show me which error or which message is oversized? I have put a log trace to see how much each (key - value) occupies, and it does not exceed 20 bytes. The limit that is configured is 1Mb
Configuration is similar to: https://gist.github.com/itzg/e3ebfd7aec220bf0522e23a65b1296c8 And like this: https://kafka.apache.org/20/documentation/streams/developer-guide/testing#