Ability to track offsets of filtered events #12602

tibrewalpratik17 · 2024-03-08T07:55:03Z

label:
observability

Adding ability to track offsets of filtered events. Publishing this log every 1 minute along with Consumed x events log. Added a separate log and did not clubbing with consumed events for easy debuggability. This will only happen when trackFilteredMessageOffset flag is enabled in Ingestion configs.

codecov-commenter · 2024-03-08T08:32:41Z

Codecov Report

Attention: Patch coverage is 37.50000% with 10 lines in your changes missing coverage. Please review.

Project coverage is 62.08%. Comparing base (59551e4) to head (7d3579a).
Report is 1278 commits behind head on master.

Files with missing lines	Patch %	Lines
...a/manager/realtime/RealtimeSegmentDataManager.java	16.66%	8 Missing and 2 partials ⚠️

Additional details and impacted files

@@             Coverage Diff              @@
##             master   #12602      +/-   ##
============================================
+ Coverage     61.75%   62.08%   +0.33%     
+ Complexity      207      198       -9     
============================================
  Files          2436     2502      +66     
  Lines        133233   136479    +3246     
  Branches      20636    21124     +488     
============================================
+ Hits          82274    84732    +2458     
- Misses        44911    45465     +554     
- Partials       6048     6282     +234

Flag	Coverage Δ
custom-integration1	`<0.01% <0.00%> (-0.01%)`	⬇️
integration	`<0.01% <0.00%> (-0.01%)`	⬇️
integration1	`<0.01% <0.00%> (-0.01%)`	⬇️
integration2	`0.00% <0.00%> (ø)`
java-11	`<0.01% <0.00%> (-61.71%)`	⬇️
java-21	`62.08% <37.50%> (+0.45%)`	⬆️
skip-bytebuffers-false	`62.06% <37.50%> (+0.32%)`	⬆️
skip-bytebuffers-true	`62.03% <37.50%> (+34.31%)`	⬆️
temurin	`62.08% <37.50%> (+0.33%)`	⬆️
unittests	`62.08% <37.50%> (+0.33%)`	⬆️
unittests1	`46.50% <37.50%> (-0.39%)`	⬇️
unittests2	`28.01% <0.00%> (+0.28%)`	⬆️

Flags with carried forward coverage won't be shown. Click here to find out more.

☔ View full report in Codecov by Sentry.
📢 Have feedback on the report? Share it here.

...re/src/main/java/org/apache/pinot/core/data/manager/realtime/RealtimeSegmentDataManager.java

Jackie-Jiang · 2024-03-08T09:02:15Z

...re/src/main/java/org/apache/pinot/core/data/manager/realtime/RealtimeSegmentDataManager.java

@@ -572,7 +573,7 @@ private boolean processStreamEvents(MessageBatch messagesAndOffsets, long idlePi
      // Decode message
      StreamDataDecoderResult decodedRow = _streamDataDecoder.decode(messagesAndOffsets.getStreamMessage(index));
      msgMetadata = messagesAndOffsets.getStreamMessage(index).getMetadata();
-      StreamPartitionMsgOffset messageOffset = messagesAndOffsets.getNextStreamPartitionMsgOffsetAtIndex(index);
+      int messageOffset = messagesAndOffsets.getMessageOffsetAtIndex(index);


Does this give the accurate offset? Will it cause overflow?

I don't see usage of this method, so I guess it might already be deprecated

I don't see it marked as deprecated and there is nothing similar to getNextStreamPartitionMsgOffsetAtIndex which returns StreamPartitionMsgOffset (it has long value). There can be scenarios of overflow.

Will wait for @satishd to chime in here as he might have more context on this.

MessageBatch's getMessageOffsetAtIndex(index) javadoc says it gives the offset of the message at the given index. But the respective Kafka implementation does not seem to return that. What does that method really mean and what is it used for?

We need Kafka message's record offset here. So, we need a similar implementation like getNextStreamPartitionMsgOffsetAtIndex() to get it from the metadata properties and get the message record's offset.

@satishd gave a shot to having a getStreamPartitionMsgOffsetAtIndex in MessageBatch. Let me know if this looks good.

...re/src/main/java/org/apache/pinot/core/data/manager/realtime/RealtimeSegmentDataManager.java

tibrewalpratik17 · 2024-03-13T11:11:46Z

...re/src/main/java/org/apache/pinot/core/data/manager/realtime/RealtimeSegmentDataManager.java

+        if (_trackFilteredMessageOffsets) {
+          _segmentLogger.info("Filtered events with offsets: {}", _filteredMessageOffsets);
+        }


self-review: adding a nested-if instead of combining it to ensure that _filteredMessageOffsets gets clear every time just in case in future someone changes the logic of pushing to _filteredMessageOffsets.

satishd

Thanks @tibrewalpratik17 for addressing the review comments. Overall LGTM.

...re/src/main/java/org/apache/pinot/core/data/manager/realtime/RealtimeSegmentDataManager.java

...pinot-kafka-2.0/src/main/java/org/apache/pinot/plugin/stream/kafka20/KafkaStreamMessage.java

...re/src/main/java/org/apache/pinot/core/data/manager/realtime/RealtimeSegmentDataManager.java

pinot-spi/src/main/java/org/apache/pinot/spi/config/table/ingestion/StreamIngestionConfig.java

pinot-spi/src/main/java/org/apache/pinot/spi/stream/MessageBatch.java

.../pinot-kafka-2.0/src/main/java/org/apache/pinot/plugin/stream/kafka20/KafkaMessageBatch.java

Jackie-Jiang

LGTM

Jackie-Jiang · 2024-04-18T00:59:02Z

...re/src/main/java/org/apache/pinot/core/data/manager/realtime/RealtimeSegmentDataManager.java

@@ -1758,6 +1769,12 @@ private void updateCurrentDocumentCountMetrics() {
      _segmentLogger.info(
          "Consumed {} events from (rate:{}/s), currentOffset={}, numRowsConsumedSoFar={}, numRowsIndexedSoFar={}",
          rowsConsumed, consumedRate, _currentOffset, _numRowsConsumed, _numRowsIndexed);
+      if (_filteredMessageOffsets.size() > 0) {
+        if (_trackFilteredMessageOffsets) {


This check is redundant

I left it on purpose if anyone changes the logic to pushing to _filteredMessageOffsets in the future for any other usecase we can still ensure that the config is enabled.

satishd

Thanks @tibrewalpratik17 for the updated PR. LGTM.

tibrewalpratik17 mentioned this pull request Mar 8, 2024

Improve exception logging when we fail to index / transform message #12594

Merged

Jackie-Jiang added the observability label Mar 8, 2024

tibrewalpratik17 mentioned this pull request Mar 8, 2024

Add isJson UDF #12603

Merged

Jackie-Jiang reviewed Mar 8, 2024

View reviewed changes

Jackie-Jiang reviewed Mar 11, 2024

View reviewed changes

...re/src/main/java/org/apache/pinot/core/data/manager/realtime/RealtimeSegmentDataManager.java Show resolved Hide resolved

tibrewalpratik17 commented Mar 13, 2024

View reviewed changes

satishd reviewed Mar 13, 2024

View reviewed changes

...re/src/main/java/org/apache/pinot/core/data/manager/realtime/RealtimeSegmentDataManager.java Show resolved Hide resolved

ankitsultana reviewed Mar 16, 2024

View reviewed changes

...pinot-kafka-2.0/src/main/java/org/apache/pinot/plugin/stream/kafka20/KafkaStreamMessage.java Outdated Show resolved Hide resolved

ankitsultana reviewed Mar 16, 2024

View reviewed changes

...re/src/main/java/org/apache/pinot/core/data/manager/realtime/RealtimeSegmentDataManager.java Show resolved Hide resolved

ankitsultana requested a review from Jackie-Jiang March 18, 2024 20:45

tibrewalpratik17 force-pushed the update_error_log2 branch from f8ca87f to 799b623 Compare March 18, 2024 20:46

ankitsultana reviewed Mar 18, 2024

View reviewed changes

pinot-spi/src/main/java/org/apache/pinot/spi/config/table/ingestion/StreamIngestionConfig.java Outdated Show resolved Hide resolved

ankitsultana reviewed Mar 18, 2024

View reviewed changes

pinot-spi/src/main/java/org/apache/pinot/spi/stream/MessageBatch.java Outdated Show resolved Hide resolved

.../pinot-kafka-2.0/src/main/java/org/apache/pinot/plugin/stream/kafka20/KafkaMessageBatch.java Outdated Show resolved Hide resolved

Add ability to track filtered messages offset

7d3579a

tibrewalpratik17 force-pushed the update_error_log2 branch from 5b68ba8 to 7d3579a Compare April 17, 2024 07:05

tibrewalpratik17 changed the title ~~Improve logging during filtering event consumption~~ Ability to track offsets of filtered events Apr 17, 2024

Jackie-Jiang approved these changes Apr 18, 2024

View reviewed changes

satishd approved these changes Apr 18, 2024

View reviewed changes

satishd mentioned this pull request Apr 18, 2024

Cleanup the consumer interfaces and legacy code #12697

Merged

Jackie-Jiang merged commit 022e0a0 into apache:master Apr 18, 2024
20 checks passed

tibrewalpratik17 deleted the update_error_log2 branch May 16, 2024 15:07

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Ability to track offsets of filtered events #12602

Ability to track offsets of filtered events #12602

tibrewalpratik17 commented Mar 8, 2024 •

edited

Loading

codecov-commenter commented Mar 8, 2024 •

edited

Loading

Jackie-Jiang Mar 8, 2024

Jackie-Jiang Mar 8, 2024

tibrewalpratik17 Mar 8, 2024 •

edited

Loading

satishd Mar 8, 2024

tibrewalpratik17 Mar 9, 2024

tibrewalpratik17 Mar 13, 2024

satishd left a comment

Jackie-Jiang left a comment

Jackie-Jiang Apr 18, 2024

tibrewalpratik17 Apr 18, 2024 •

edited

Loading

satishd left a comment

Ability to track offsets of filtered events #12602

Ability to track offsets of filtered events #12602

Conversation

tibrewalpratik17 commented Mar 8, 2024 • edited Loading

codecov-commenter commented Mar 8, 2024 • edited Loading

Codecov Report

Jackie-Jiang Mar 8, 2024

Choose a reason for hiding this comment

Jackie-Jiang Mar 8, 2024

Choose a reason for hiding this comment

tibrewalpratik17 Mar 8, 2024 • edited Loading

Choose a reason for hiding this comment

satishd Mar 8, 2024

Choose a reason for hiding this comment

tibrewalpratik17 Mar 9, 2024

Choose a reason for hiding this comment

tibrewalpratik17 Mar 13, 2024

Choose a reason for hiding this comment

satishd left a comment

Choose a reason for hiding this comment

Jackie-Jiang left a comment

Choose a reason for hiding this comment

Jackie-Jiang Apr 18, 2024

Choose a reason for hiding this comment

tibrewalpratik17 Apr 18, 2024 • edited Loading

Choose a reason for hiding this comment

satishd left a comment

Choose a reason for hiding this comment

tibrewalpratik17 commented Mar 8, 2024 •

edited

Loading

codecov-commenter commented Mar 8, 2024 •

edited

Loading

tibrewalpratik17 Mar 8, 2024 •

edited

Loading

tibrewalpratik17 Apr 18, 2024 •

edited

Loading