navina commented on code in PR #9224:
URL: https://github.com/apache/pinot/pull/9224#discussion_r980381894
##########
pinot-core/src/main/java/org/apache/pinot/core/data/manager/realtime/LLRealtimeSegmentDataManager.java:
##########
@@ -544,16 +547,16 @@ private boolean processStreamEvents(MessageBatch
messagesAndOffsets, long idlePi
// Index each message
reuse.clear();
- // retrieve metadata from the message batch if available
- // this can be overridden by the decoder if there is a better indicator
in the message payload
- RowMetadata msgMetadata = messagesAndOffsets.getMetadataAtIndex(index);
-
- GenericRow decodedRow = _messageDecoder
- .decode(messagesAndOffsets.getMessageAtIndex(index),
messagesAndOffsets.getMessageOffsetAtIndex(index),
- messagesAndOffsets.getMessageLengthAtIndex(index), reuse);
- if (decodedRow != null) {
+ StreamDataDecoderResult decodedRow =
_streamDataDecoder.decode(messagesAndOffsets.getStreamMessage(index));
+ RowMetadata msgMetadata =
messagesAndOffsets.getStreamMessage(index).getMetadata();
+ if (decodedRow.getException() != null) {
+ // TODO: handle exception as we do today - do we silently drop the
record or throw exception?
Review Comment:
Looks like, we increment a metric today when an exception is thrown by the
decoder itself or if `null` is returned. I am leaving this part unchanged for
now.
In a follow-up PR, I think we can elegantly handle decode errors for all
supported streams - my proposal would be to use a config to decide whether to
re-throw exception or skip the record.
--
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
