oscerd commented on code in PR #14241:
URL: https://github.com/apache/camel/pull/14241#discussion_r1620483018
##########
components/camel-aws/camel-aws2-kinesis/src/main/java/org/apache/camel/component/aws2/kinesis/Kinesis2Producer.java:
##########
@@ -69,8 +96,37 @@ private PutRecordRequest createRequest(Exchange exchange) {
return putRecordRequest.build();
}
- public static Message getMessageForResponse(final Exchange exchange) {
- return exchange.getMessage();
+ private PutRecordsRequestEntry createRequestEntry(Exchange exchange) {
Review Comment:
There is no security about the presence of the header.
##########
components/camel-aws/camel-aws2-kinesis/src/main/java/org/apache/camel/component/aws2/kinesis/Kinesis2Producer.java:
##########
@@ -47,9 +59,24 @@ public Kinesis2Endpoint getEndpoint() {
@Override
public void process(Exchange exchange) throws Exception {
+ Boolean batchComplete = exchange.getProperty(Exchange.BATCH_COMPLETE,
Boolean.class);
+ if (batchComplete == null) {
+ flushRequestBatch();
+ sendSingleRecord(exchange);
+ return;
+ }
+
+ this.requestBatch.add(createRequestEntry(exchange));
+ if (batchComplete) {
+ flushRequestBatch();
+ }
+ }
+
+ private void sendSingleRecord(Exchange exchange) {
PutRecordRequest request = createRequest(exchange);
PutRecordResponse putRecordResult =
connection.getClient(getEndpoint()).putRecord(request);
- Message message = getMessageForResponse(exchange);
+ LOG.trace("Sent 1 record.");
Review Comment:
I think this is useless without details.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscr...@camel.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
