deemoliu commented on code in PR #14496:
URL: https://github.com/apache/pinot/pull/14496#discussion_r1850979779
##########
pinot-plugins/pinot-stream-ingestion/pinot-kinesis/src/main/java/org/apache/pinot/plugin/stream/kinesis/KinesisConsumer.java:
##########
@@ -150,8 +150,8 @@ private BytesStreamMessage extractStreamMessage(Record
record, String shardId) {
String sequenceNumber = record.sequenceNumber();
KinesisPartitionGroupOffset offset = new
KinesisPartitionGroupOffset(shardId, sequenceNumber);
// NOTE: Use the same offset as next offset because the consumer starts
consuming AFTER the start sequence number.
- StreamMessageMetadata.Builder builder =
- new
StreamMessageMetadata.Builder().setRecordIngestionTimeMs(timestamp).setOffset(offset,
offset);
+ StreamMessageMetadata.Builder builder = new
StreamMessageMetadata.Builder().setRecordIngestionTimeMs(timestamp)
+
.setSerializedValueSize(record.data().asByteArray().length).setOffset(offset,
offset);
Review Comment:
sounds good, let's change it to value.length.
--
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...@pinot.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscr...@pinot.apache.org
For additional commands, e-mail: commits-h...@pinot.apache.org
