Cleanup the consumer interfaces and legacy code #12697

Jackie-Jiang · 2024-03-22T09:03:08Z

Deprecated:

PartitionLevelConsumer: Use PartitionGroupConsumer instead
In PartitionGroupConsumer:
- Remove endOffset from fetchMessages() because offset might not be monotonically increasing for some consumers (e.g. Kinesis), and we should not use end offset to filter messages
In MessageBatch:
- getMessageAtIndex(): Use getStreamMessage() instead
- getMessageBytesAtIndex(): Use getStreamMessage() instead
- getMessageLengthAtIndex(): Use getStreamMessage() instead
- getMessageOffsetAtIndex(): Use getStreamMessage() instead, where offset info is embedded in the StreamMessageMetadata
- getMetadataAtIndex(): Use getStreamMessage() instead
- getNextStreamMessageOffsetAtIndex(): Use getOffsetOfNextBatch() instead
- getNextStreamPartitionMsgOffsetAtIndex(): Use getOffsetOfNextBatch() instead
In StreamPartitionMsgOffset:
- fromString(): Should be a static method

Removed Metadata Extractor interfaces because they are not well designed and not really pluggable (plug at consumer level should be good enough)

Add offset info into RowMetadata:

StreamPartitionMsgOffset getOffset()
StreamPartitionMsgOffset getNextOffset()

Bugfix:

With offset info properly attached to message metadata, we can correctly track the offset for each message. Currently we are using the next offset for each message

Clean up the code to adapt the above changes.

codecov-commenter · 2024-03-22T09:40:17Z

Codecov Report

Attention: Patch coverage is 54.02844% with 194 lines in your changes are missing coverage. Please review.

Project coverage is 62.00%. Comparing base (59551e4) to head (082aada).
Report is 204 commits behind head on master.

Files	Patch %	Lines
...apache/pinot/plugin/stream/pulsar/PulsarUtils.java	53.26%	38 Missing and 5 partials ⚠️
...lugin/stream/kinesis/KinesisConnectionHandler.java	5.88%	32 Missing ⚠️
...java/org/apache/pinot/spi/stream/MessageBatch.java	25.00%	12 Missing and 3 partials ⚠️
...e/pinot/plugin/stream/kinesis/KinesisConsumer.java	65.85%	8 Missing and 6 partials ⚠️
...a/manager/realtime/RealtimeSegmentDataManager.java	54.16%	8 Missing and 3 partials ⚠️
...in/stream/kafka20/KafkaPartitionLevelConsumer.java	68.57%	10 Missing and 1 partial ⚠️
...in/stream/kinesis/KinesisPartitionGroupOffset.java	33.33%	10 Missing ⚠️
...apache/pinot/spi/stream/StreamMessageMetadata.java	58.33%	10 Missing ⚠️
.../pulsar/PulsarPartitionLevelConnectionHandler.java	52.94%	5 Missing and 3 partials ⚠️
...ava/org/apache/pinot/spi/stream/StreamMessage.java	36.36%	7 Missing ⚠️
... and 13 more

Additional details and impacted files

@@             Coverage Diff              @@
##             master   #12697      +/-   ##
============================================
+ Coverage     61.75%   62.00%   +0.25%     
+ Complexity      207      198       -9     
============================================
  Files          2436     2458      +22     
  Lines        133233   134667    +1434     
  Branches      20636    20812     +176     
============================================
+ Hits          82274    83500    +1226     
- Misses        44911    45021     +110     
- Partials       6048     6146      +98

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	`61.94% <54.02%> (+0.23%)`	⬆️
java-21	`61.87% <54.02%> (+0.24%)`	⬆️
skip-bytebuffers-false	`61.97% <54.02%> (+0.22%)`	⬆️
skip-bytebuffers-true	`61.84% <54.02%> (+34.11%)`	⬆️
temurin	`62.00% <54.02%> (+0.25%)`	⬆️
unittests	`62.00% <54.02%> (+0.25%)`	⬆️
unittests1	`46.73% <45.36%> (-0.16%)`	⬇️
unittests2	`27.95% <43.60%> (+0.21%)`	⬆️

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.

...tion/pinot-kinesis/src/main/java/org/apache/pinot/plugin/stream/kinesis/KinesisConsumer.java

Jackie-Jiang · 2024-04-02T21:00:24Z

I've reverted the change of removing separate executor for Kinesis and Pulsar to make this PR only focus on the interface change

swaminathanmanish

Overall this cleanup makes sense. Thanks for taking this up and explaining the rationale.
Completed core apis & kafka review. Looking through kinesis, pulsar.

swaminathanmanish · 2024-04-03T14:30:04Z

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

            _segmentLogger.error(errorMessage, e);
            _realtimeTableDataManager.addSegmentError(_segmentNameStr, new SegmentErrorInfo(now(), errorMessage, e));
          }
        }
      }
-      _currentOffset = messageOffset;
+      _currentOffset = nextOffset;
      _numRowsIndexed = _realtimeSegment.getNumDocsIndexed();
      _numRowsConsumed++;
      streamMessageCount++;
    }


Thanks for cleaning this up. Hope this can be verified.

if (indexedMessageCount > 0) {
// Record Ingestion delay for this partition with metadata for last message we processed
updateIngestionDelay(_lastRowMetadata);
} else if (!hasTransformedRows && (msgMetadata != null)) {
// If all messages were filtered by transformation, we still attempt to update ingestion delay using
// the metadata for the last message we processed if any.
updateIngestionDelay(msgMetadata);
}

The logic was added in #10418, and unfortunately seems like there is no test added.

Since we already provided a default implementation of getLastMessageMetadata(), it should be safe to simplify this logic.

swaminathanmanish · 2024-04-03T22:39:36Z

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

-  @Override
-  public StreamMessage getStreamMessage(int index) {
-    return _messageList.get(index);
+  public StreamPartitionMsgOffset getFirstMessageOffset() {


is this backwards compatible, since it can return null

We perform a null check after reading the value (the logic already exists), so this is actually a bug fix. When there is no value consumed, we should return null instead of negative offset. It is backward compatible.
Added an @Nullable annotation to this method

swaminathanmanish · 2024-04-03T22:49:01Z

...ka-2.0/src/main/java/org/apache/pinot/plugin/stream/kafka20/KafkaPartitionLevelConsumer.java

+    if (!records.isEmpty()) {
+      firstOffset = records.get(0).offset();
+      _lastFetchedOffset = records.get(records.size() - 1).offset();
+      offsetOfNextBatch = _lastFetchedOffset + 1;


This makes sense and simplifies the logic. Hope we can comprehensively verify through integ tests.

This is verified. If this value is not properly set, the consumption won't match the expected rows

swaminathanmanish · 2024-04-04T20:41:03Z

...pulsar/src/main/java/org/apache/pinot/plugin/stream/pulsar/PulsarPartitionLevelConsumer.java


-  public PulsarPartitionLevelConsumer(String clientId, StreamConfig streamConfig,
-      PartitionGroupConsumptionStatus partitionGroupConsumptionStatus) {
+  // TODO: Revisit the logic of using a separate executor to manage the request timeout. Currently it is not thread safe


Is enableKeyValueStitch not used any more?

@KKcorps - Could you verify as well?

It is handled within PulsarUtils.buildPulsarStreamMessage()

swaminathanmanish · 2024-04-04T20:46:47Z

...tion/pinot-kinesis/src/main/java/org/apache/pinot/plugin/stream/kinesis/KinesisConsumer.java

    }
+    return buildKinesisMessageBatch(startOffset, messages, false);


Is there still a race condition if the executor is running along with buildKinesisMessageBatch?

Yes. This race condition exists in the current code as well. Will try to address it separately.

swaminathanmanish · 2024-04-04T20:50:16Z

...tion/pinot-kinesis/src/main/java/org/apache/pinot/plugin/stream/kinesis/KinesisConsumer.java

-    _shardIteratorType = kinesisConfig.getShardIteratorType();
-    _rpsLimit = kinesisConfig.getRpsLimit();
-    _executorService = Executors.newSingleThreadExecutor();
+  // TODO: Revisit the logic of using a separate executor to manage the request timeout. Currently it is not thread safe


Will be good to understand why we created an executor.

IMO we should not. Currently I believe we created it to enforce the timeout. @KKcorps to confirm

swaminathanmanish

Thanks for the cleanup.

satishd · 2024-04-18T10:41:57Z

Thanks @Jackie-Jiang for cleaning up and fixing these interfaces. It helps address issues like ##12602

jadami10

sorry for the delay reviewing. Everything looks backwards compatible to me and should still work for those of us within at least 1 version up to date with these interfaces.

jadami10 · 2024-04-20T17:12:09Z

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

-   * @return A {@link GenericRow} that encapsulates the headers in the ingested row
+   * Returns the stream offset of the message.
+   */
+  @Nullable


out of curiosity, why do we use nulls and nullable annnotation instead of Optionals?

Both historic reason and performance reason:

Optional was introduced in java 8 (it is handy to be used along with function APIs), and majority of the code is using null

There could be performance overhead because it is an extra Object wrapper (24 bytes) over a reference. In majority of the places it doesn't matter, but checkout this interesting post

that's a solid post. thank you for the reference!

jadami10 · 2024-04-20T17:15:01Z

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

        if (metadata != null) {
          if (metadata.getHeaders() != null) {
            metadata.getHeaders().getFieldToValueMap()
                .forEach((key, value) -> row.putValue(HEADER_KEY_PREFIX + key, value));
          }
-          metadata.getRecordMetadata()
-                  .forEach((key, value) -> row.putValue(METADATA_KEY_PREFIX + key, value));
+          if (metadata.getRecordMetadata() != null) {


maybe

Optional.ofNullable(metadata.getRecordMetadata()) .orElse(Collections.emptyMap())

is a better pattern for the future so we don't call get twice

Jackie-Jiang added enhancement ingestion cleanup labels Mar 22, 2024

Jackie-Jiang added the real-time label Mar 22, 2024

Jackie-Jiang force-pushed the message_batch branch 2 times, most recently from 1e66caf to a9f3cf7 Compare March 23, 2024 01:12

Jackie-Jiang added the bugfix label Mar 23, 2024

Jackie-Jiang force-pushed the message_batch branch from a9f3cf7 to d7d7a51 Compare March 23, 2024 01:41

Jackie-Jiang requested review from KKcorps, xiangfu0, mcvsubbu, sajjad-moradi and snleee March 26, 2024 05:10

KKcorps reviewed Mar 26, 2024

View reviewed changes

...tion/pinot-kinesis/src/main/java/org/apache/pinot/plugin/stream/kinesis/KinesisConsumer.java Outdated Show resolved Hide resolved

KKcorps reviewed Mar 26, 2024

View reviewed changes

...tion/pinot-kinesis/src/main/java/org/apache/pinot/plugin/stream/kinesis/KinesisConsumer.java Show resolved Hide resolved

Jackie-Jiang force-pushed the message_batch branch from d7d7a51 to 9a75dbf Compare April 2, 2024 20:58

Jackie-Jiang force-pushed the message_batch branch from 9a75dbf to f4bab13 Compare April 2, 2024 22:23

swaminathanmanish reviewed Apr 3, 2024

View reviewed changes

Cleanup the consumer interfaces and legacy code

d7b690f

Jackie-Jiang force-pushed the message_batch branch from f4bab13 to 5c693ce Compare April 4, 2024 00:51

Add nullable annotation

082aada

Jackie-Jiang force-pushed the message_batch branch from 5c693ce to 082aada Compare April 4, 2024 00:53

swaminathanmanish reviewed Apr 4, 2024

View reviewed changes

swaminathanmanish approved these changes Apr 4, 2024

View reviewed changes

KKcorps approved these changes Apr 6, 2024

View reviewed changes

Jackie-Jiang merged commit 7f87cd3 into apache:master Apr 7, 2024
21 checks passed

Jackie-Jiang deleted the message_batch branch April 7, 2024 21:19

Jackie-Jiang mentioned this pull request Apr 8, 2024

Make KafkaConsumerFactory method less restrictive to avoid incompatibility #12815

Merged

jadami10 reviewed Apr 20, 2024

View reviewed changes

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Cleanup the consumer interfaces and legacy code #12697

Cleanup the consumer interfaces and legacy code #12697

Jackie-Jiang commented Mar 22, 2024 •

edited

Loading

codecov-commenter commented Mar 22, 2024 •

edited

Loading

Jackie-Jiang commented Apr 2, 2024

swaminathanmanish left a comment

swaminathanmanish Apr 3, 2024

Jackie-Jiang Apr 4, 2024

swaminathanmanish Apr 3, 2024

Jackie-Jiang Apr 4, 2024

swaminathanmanish Apr 3, 2024

Jackie-Jiang Apr 4, 2024

swaminathanmanish Apr 4, 2024

Jackie-Jiang Apr 4, 2024

swaminathanmanish Apr 4, 2024

Jackie-Jiang Apr 4, 2024

swaminathanmanish Apr 4, 2024

Jackie-Jiang Apr 4, 2024

swaminathanmanish left a comment

satishd commented Apr 18, 2024

jadami10 left a comment

jadami10 Apr 20, 2024

Jackie-Jiang Apr 24, 2024

jadami10 Apr 25, 2024

jadami10 Apr 20, 2024

		}
		return buildKinesisMessageBatch(startOffset, messages, false);

Cleanup the consumer interfaces and legacy code #12697

Cleanup the consumer interfaces and legacy code #12697

Conversation

Jackie-Jiang commented Mar 22, 2024 • edited Loading

codecov-commenter commented Mar 22, 2024 • edited Loading

Codecov Report

Jackie-Jiang commented Apr 2, 2024

swaminathanmanish left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

swaminathanmanish left a comment

Choose a reason for hiding this comment

satishd commented Apr 18, 2024

jadami10 left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Jackie-Jiang commented Mar 22, 2024 •

edited

Loading

codecov-commenter commented Mar 22, 2024 •

edited

Loading