[jira] [Resolved] (CAMEL-15885) AWS KINESIS component not de-aggregating producer side aggregated msgs

["Claus Ibsen (Jira)" <ji...@apache.org>]

     [ https://issues.apache.org/jira/browse/CAMEL-15885?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ]

Claus Ibsen resolved CAMEL-15885.
---------------------------------
    Resolution: Won't Fix

we use aws2 now

> AWS KINESIS component not de-aggregating producer side aggregated msgs
> ----------------------------------------------------------------------
>
>                 Key: CAMEL-15885
>                 URL: https://issues.apache.org/jira/browse/CAMEL-15885
>             Project: Camel
>          Issue Type: New Feature
>          Components: camel-aws
>            Reporter: m
>            Priority: Minor
>
> {code:java}
> <quarkus-plugin.version>1.8.1.Final</quarkus-plugin.version>
> <dependency>
>    <groupId>org.apache.camel.quarkus</groupId>
>    <artifactId>camel-quarkus-aws-kinesis</artifactId>
> </dependency>{code}
>  
>  Our producer sends zipped messages. When reading those about 1/4 of all msgs cannot be unzipped. When digging into the message blob it became clear that one message contains multiple data blobs. That's probably due to the producer side message aggregation ( [https://docs.aws.amazon.com/streams/latest/dev/kinesis-kpl-consumer-deaggregation.html]).
>   
>  Instead of a single message the data from the Record entity contains obviously several merged data blobs that I'm unable to read (zip error when trying to decompress). I'd expect to receive several msgs in my processor instead of several msgs merged into one msg blob.
>  
> {code:java}
>  from("aws-kinesis://"  lsc.getStreamName()  "?amazonKinesisClient=client&bridgeErrorHandler=true&maxResultsPerRequest=500&greedy=true&delay=0&runLoggingLevel=TRACE&iteratorType=LATEST")
>                  .routeId("vssKinesisStream")
>                  .log("LSC message received.")
>                  .onException(java.util.zip.ZipException.class).process(exchange -> 
> {                     metrics.countInvalidPerSec();                 }
> )
>                  .stop()
>                  .end()
>                  .process(exchange -> 
> {                         metrics.countReceivedPerSec();                         Record record = exchange.getIn().getBody(Record.class);                         String message = extractMessageFromRecord(record);                         exchange.getIn().setBody(message);                         metrics.countValidPerSec();                         LOG.info("Extracted message: " + message);                         lastReadMsgAge = Duration.between(record.getApproximateArrivalTimestamp().toInstant(), Instant.now());                 }
> {code}
>  
>   



--
This message was sent by Atlassian Jira
(v8.20.10#820010)