You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pinot.apache.org by GitBox <gi...@apache.org> on 2022/10/18 18:46:38 UTC
[GitHub] [pinot] navina opened a new pull request, #9621: Add record availability lag for realtime connectors
navina opened a new pull request, #9621:
URL: https://github.com/apache/pinot/pull/9621
Introduces record availability lag for realtime connectors via the rest apis - `consumingSegmentsInfo`
and `/debug/tables/{tableName}`
Label: `release-notes`
`/consumingSegmentsInfo` and `debug/tables/{tableName}` API now provides the following per-partition numbers for Kafka data sources:
`currentConsumingOffset`, `latestUpstreamOffset`, `recordsLag`, `recordsAvailabilityLag`
--
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-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [pinot] navina commented on a diff in pull request #9621: Add record availability lag for Kafka connector
Posted by GitBox <gi...@apache.org>.
navina commented on code in PR #9621:
URL: https://github.com/apache/pinot/pull/9621#discussion_r1000970918
##########
pinot-spi/src/main/java/org/apache/pinot/spi/stream/PartitionLagState.java:
##########
@@ -27,12 +27,20 @@ public class PartitionLagState {
protected final static String NOT_CALCULATED = "NOT_CALCULATED";
/**
- * Defines how far away the current record's offset / pointer is from upstream latest record
+ * Defines how far behind the current record's offset / pointer is from upstream latest record
* The distance is based on actual record count.
*/
public String getRecordsLag() {
return NOT_CALCULATED;
}
- // TODO: Define record availability lag ($latest_record_consumption_time - $latest_record_ingestion_time)
+ /**
+ * Defines how soon after record ingestion was the record consumed by Pinot. That is, the difference between the
+ * time the record was consumed and the time at which the record was ingested upstream.
+ *
+ * @return Lag value in milliseconds
+ */
+ public String getRecordAvailabilityLag() {
Review Comment:
> Maybe getRecordsLagMs?
This would make one think that this lag is computed relative to the timestamp of the latest record in upstream (ie. latest_upstream_record_timestamp minus current_consumed_record_timestamp). This is the lag in terms of record time relative to upstream. This is actually a metric that kinesis provides and recommends applications to monitor - `millisBehindLatest` ([here](https://docs.aws.amazon.com/kinesis/latest/APIReference/API_GetRecords.html#API_GetRecords_ResponseSyntax)).
Where as, what we are computing here is relative to consumption time and the time at which the record was ingested upstream. I think it should be treated differently than the above and hence, I used the term `recordAvailabilityLag` as it shows how soon the record was consumed after it arrived in the stream.
> Not sure if availability lag is a common term in streaming
I thought it was. At least it was in the kafka ecosystem.
I am fine renaming the metric in this PR to `recordLagMs` and then, actually add another metric called `millisBehindLatest` (similar to kinesis). Wdyt?
--
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-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [pinot] navina commented on a diff in pull request #9621: Add record availability lag for Kafka connector
Posted by GitBox <gi...@apache.org>.
navina commented on code in PR #9621:
URL: https://github.com/apache/pinot/pull/9621#discussion_r1000144296
##########
pinot-spi/src/main/java/org/apache/pinot/spi/stream/PartitionLagState.java:
##########
@@ -27,12 +27,20 @@ public class PartitionLagState {
protected final static String NOT_CALCULATED = "NOT_CALCULATED";
/**
- * Defines how far away the current record's offset / pointer is from upstream latest record
+ * Defines how far behind the current record's offset / pointer is from upstream latest record
* The distance is based on actual record count.
*/
public String getRecordsLag() {
return NOT_CALCULATED;
}
- // TODO: Define record availability lag ($latest_record_consumption_time - $latest_record_ingestion_time)
+ /**
+ * Defines how soon after record ingestion was the record consumed by Pinot. That is, the difference between the
+ * time the record was consumed and the time at which the record was ingested upstream.
+ *
+ * @return Lag value in milliseconds
+ */
+ public String getRecordAvailabilityLag() {
Review Comment:
I didn't want to call it "offset" lag as that term doesn't makes sense for anything other than kafka. in fact, `recordsLag` can be computed only for Kafka. Hence, I am using the same term as kafka's metric name for lag.
For every other system, we can only compute a time-based lag. and it indicates when the record becomes available for consumption. I was considering calling it `getRecordAvailabilityLagMs`, if that helps clarify the meaning?
--
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-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [pinot] Jackie-Jiang commented on a diff in pull request #9621: Add record availability lag for Kafka connector
Posted by GitBox <gi...@apache.org>.
Jackie-Jiang commented on code in PR #9621:
URL: https://github.com/apache/pinot/pull/9621#discussion_r1000007263
##########
pinot-spi/src/main/java/org/apache/pinot/spi/stream/PartitionLagState.java:
##########
@@ -27,12 +27,20 @@ public class PartitionLagState {
protected final static String NOT_CALCULATED = "NOT_CALCULATED";
/**
- * Defines how far away the current record's offset / pointer is from upstream latest record
+ * Defines how far behind the current record's offset / pointer is from upstream latest record
* The distance is based on actual record count.
*/
public String getRecordsLag() {
return NOT_CALCULATED;
}
- // TODO: Define record availability lag ($latest_record_consumption_time - $latest_record_ingestion_time)
+ /**
+ * Defines how soon after record ingestion was the record consumed by Pinot. That is, the difference between the
+ * time the record was consumed and the time at which the record was ingested upstream.
+ *
+ * @return Lag value in milliseconds
+ */
+ public String getRecordAvailabilityLag() {
Review Comment:
Is my understanding correctly: recordsLag is the lag in offset, recordAvailabilityLag is the lag in millis? If so, shall we rename them so that it's more explicit?
--
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-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [pinot] Jackie-Jiang commented on a diff in pull request #9621: Add record availability lag for Kafka connector
Posted by GitBox <gi...@apache.org>.
Jackie-Jiang commented on code in PR #9621:
URL: https://github.com/apache/pinot/pull/9621#discussion_r1000931425
##########
pinot-spi/src/main/java/org/apache/pinot/spi/stream/PartitionLagState.java:
##########
@@ -27,12 +27,20 @@ public class PartitionLagState {
protected final static String NOT_CALCULATED = "NOT_CALCULATED";
/**
- * Defines how far away the current record's offset / pointer is from upstream latest record
+ * Defines how far behind the current record's offset / pointer is from upstream latest record
* The distance is based on actual record count.
*/
public String getRecordsLag() {
return NOT_CALCULATED;
}
- // TODO: Define record availability lag ($latest_record_consumption_time - $latest_record_ingestion_time)
+ /**
+ * Defines how soon after record ingestion was the record consumed by Pinot. That is, the difference between the
+ * time the record was consumed and the time at which the record was ingested upstream.
+ *
+ * @return Lag value in milliseconds
+ */
+ public String getRecordAvailabilityLag() {
Review Comment:
Maybe `getRecordsLagMs`?
Or do you want to track other kinds of lags in the future? Not sure if `availability lag` is a common term in streaming
--
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-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [pinot] Jackie-Jiang commented on a diff in pull request #9621: Add record availability lag for Kafka connector
Posted by GitBox <gi...@apache.org>.
Jackie-Jiang commented on code in PR #9621:
URL: https://github.com/apache/pinot/pull/9621#discussion_r1005002974
##########
pinot-server/src/main/java/org/apache/pinot/server/api/resources/TablesResource.java:
##########
@@ -517,8 +516,12 @@ public List<SegmentConsumerInfo> getConsumingSegmentsInfo(
RealtimeSegmentDataManager realtimeSegmentDataManager = (RealtimeSegmentDataManager) segmentDataManager;
Map<String, ConsumerPartitionState> partitionStateMap =
realtimeSegmentDataManager.getConsumerPartitionState();
- Map<String, PartitionLagState> partitionLagStateMap =
- realtimeSegmentDataManager.getPartitionToLagState(partitionStateMap);
+ Map<String, String> recordsLagMap = new HashMap<>();
+ Map<String, String> availabilityLagMap = new HashMap<>();
Review Comment:
(minor) Use the full name: `recordAvailabilityLagMsMap`
##########
pinot-plugins/pinot-stream-ingestion/pinot-kafka-2.0/src/main/java/org/apache/pinot/plugin/stream/kafka20/KafkaConsumerPartitionLag.java:
##########
@@ -23,12 +23,19 @@
public class KafkaConsumerPartitionLag extends PartitionLagState {
private final String _recordsLag;
+ private final String _recordAvailabilityLag;
- public KafkaConsumerPartitionLag(String recordsLag) {
+ public KafkaConsumerPartitionLag(String recordsLag, String recordAvailabilityLag) {
Review Comment:
```suggestion
public KafkaConsumerPartitionLag(String recordsLag, String recordAvailabilityLagMs) {
```
##########
pinot-plugins/pinot-stream-ingestion/pinot-kafka-2.0/src/main/java/org/apache/pinot/plugin/stream/kafka20/KafkaStreamMetadataProvider.java:
##########
@@ -113,14 +114,27 @@ public Map<String, PartitionLagState> getCurrentPartitionLagState(
Map<String, ConsumerPartitionState> currentPartitionStateMap) {
Map<String, PartitionLagState> perPartitionLag = new HashMap<>();
for (Map.Entry<String, ConsumerPartitionState> entry: currentPartitionStateMap.entrySet()) {
- StreamPartitionMsgOffset currentOffset = entry.getValue().getCurrentOffset();
- StreamPartitionMsgOffset upstreamLatest = entry.getValue().getUpstreamLatestOffset();
+ ConsumerPartitionState partitionState = entry.getValue();
+ // Compute records-lag
+ StreamPartitionMsgOffset currentOffset = partitionState.getCurrentOffset();
+ StreamPartitionMsgOffset upstreamLatest = partitionState.getUpstreamLatestOffset();
+ String offsetLagString = "UNKNOWN";
+
if (currentOffset instanceof LongMsgOffset && upstreamLatest instanceof LongMsgOffset) {
long offsetLag = ((LongMsgOffset) upstreamLatest).getOffset() - ((LongMsgOffset) currentOffset).getOffset();
- perPartitionLag.put(entry.getKey(), new KafkaConsumerPartitionLag(String.valueOf(offsetLag)));
- } else {
- perPartitionLag.put(entry.getKey(), new KafkaConsumerPartitionLag("UNKNOWN"));
+ offsetLagString = String.valueOf(offsetLag);
}
+
+ // Compute record-availability
+ String recordAvailabilityLag = "UNKNOWN";
Review Comment:
(minor)
```suggestion
String recordAvailabilityLagMs = "UNKNOWN";
```
##########
pinot-controller/src/main/java/org/apache/pinot/controller/util/ConsumingSegmentInfoReader.java:
##########
@@ -237,13 +238,18 @@ static public class PartitionOffsetInfo {
@JsonProperty("latestUpstreamOffsetMap")
public Map<String, String> _latestUpstreamOffsetMap;
+ @JsonProperty("recordsAvailabilityLagMap")
+ public Map<String, String> _availabilityLagMap;
+
public PartitionOffsetInfo(
@JsonProperty("currentOffsetsMap") Map<String, String> currentOffsetsMap,
@JsonProperty("latestUpstreamOffsetMap") Map<String, String> latestUpstreamOffsetMap,
- @JsonProperty("recordsLagMap") Map<String, String> recordsLagMap) {
+ @JsonProperty("recordsLagMap") Map<String, String> recordsLagMap,
+ @JsonProperty("recordsAvailabilityLagMap") Map<String, String> availabilityLagMap) {
Review Comment:
```suggestion
@JsonProperty("recordsAvailabilityLagMsMap") Map<String, String> recordAvailabilityLagMsMap) {
```
##########
pinot-controller/src/main/java/org/apache/pinot/controller/util/ConsumingSegmentInfoReader.java:
##########
@@ -237,13 +238,18 @@ static public class PartitionOffsetInfo {
@JsonProperty("latestUpstreamOffsetMap")
public Map<String, String> _latestUpstreamOffsetMap;
+ @JsonProperty("recordsAvailabilityLagMap")
+ public Map<String, String> _availabilityLagMap;
Review Comment:
```suggestion
@JsonProperty("recordsAvailabilityLagMsMap")
public Map<String, String> _recordAvailabilityLagMsMap;
```
##########
pinot-plugins/pinot-stream-ingestion/pinot-kafka-2.0/src/main/java/org/apache/pinot/plugin/stream/kafka20/KafkaConsumerPartitionLag.java:
##########
@@ -23,12 +23,19 @@
public class KafkaConsumerPartitionLag extends PartitionLagState {
private final String _recordsLag;
+ private final String _recordAvailabilityLag;
Review Comment:
```suggestion
private final String _recordAvailabilityMsLag;
```
##########
pinot-common/src/main/java/org/apache/pinot/common/restlet/resources/SegmentConsumerInfo.java:
##########
@@ -97,5 +102,9 @@ public Map<String, String> getRecordsLag() {
public Map<String, String> getLatestUpstreamOffsets() {
return _latestUpstreamOffsets;
}
+
+ public Map<String, String> getAvailabilityLag() {
+ return _availabilityLag;
Review Comment:
```suggestion
public Map<String, String> getRecordAvailabilityLagMs() {
return _recordAvailabilityLagMs;
```
##########
pinot-server/src/main/java/org/apache/pinot/server/api/resources/DebugResource.java:
##########
@@ -168,22 +168,24 @@ private SegmentConsumerInfo getSegmentConsumerInfo(SegmentDataManager segmentDat
if (tableType == TableType.REALTIME) {
RealtimeSegmentDataManager realtimeSegmentDataManager = (RealtimeSegmentDataManager) segmentDataManager;
Map<String, ConsumerPartitionState> partitionStateMap = realtimeSegmentDataManager.getConsumerPartitionState();
- Map<String, PartitionLagState> partitionLagStateMap =
- realtimeSegmentDataManager.getPartitionToLagState(partitionStateMap);
- Map<String, String> partitionToCurrentOffsetMap = realtimeSegmentDataManager.getPartitionToCurrentOffset();
+ Map<String, String> currentOffsets = realtimeSegmentDataManager.getPartitionToCurrentOffset();
+ Map<String, String> upstreamLatest = partitionStateMap.entrySet().stream().collect(
+ Collectors.toMap(Map.Entry::getKey, e -> e.getValue().getUpstreamLatestOffset().toString()));
+ Map<String, String> recordsLagMap = new HashMap<>();
+ Map<String, String> recordsAvailabilityMap = new HashMap<>();
Review Comment:
```suggestion
Map<String, String> recordAvailabilityLagMsMap = new HashMap<>();
```
##########
pinot-common/src/main/java/org/apache/pinot/common/restlet/resources/SegmentConsumerInfo.java:
##########
@@ -68,23 +68,28 @@ public PartitionOffsetInfo getPartitionOffsetInfo() {
@JsonIgnoreProperties(ignoreUnknown = true)
static public class PartitionOffsetInfo {
- @JsonProperty("currentOffsets")
- public Map<String, String> _currentOffsets;
-
- @JsonProperty("recordsLag")
- public Map<String, String> _recordsLag;
-
- @JsonProperty("latestUpstreamOffsets")
- public Map<String, String> _latestUpstreamOffsets;
-
- public PartitionOffsetInfo(
- @JsonProperty("currentOffsets") Map<String, String> currentOffsets,
- @JsonProperty("latestUpstreamOffsets") Map<String, String> latestUpstreamOffsets,
- @JsonProperty("recordsLag") Map<String, String> recordsLag) {
- _currentOffsets = currentOffsets;
- _latestUpstreamOffsets = latestUpstreamOffsets;
- _recordsLag = recordsLag;
- }
+ @JsonProperty("currentOffsets")
+ public Map<String, String> _currentOffsets;
+
+ @JsonProperty("recordsLag")
+ public Map<String, String> _recordsLag;
+
+ @JsonProperty("latestUpstreamOffsets")
+ public Map<String, String> _latestUpstreamOffsets;
+
+ @JsonProperty("recordsAvailabilityLag")
+ public Map<String, String> _availabilityLag;
+
+ public PartitionOffsetInfo(
+ @JsonProperty("currentOffsets") Map<String, String> currentOffsets,
+ @JsonProperty("latestUpstreamOffsets") Map<String, String> latestUpstreamOffsets,
+ @JsonProperty("recordsLag") Map<String, String> recordsLag,
+ @JsonProperty("recordsAvailabilityLag") Map<String, String> availabilityLag) {
Review Comment:
```suggestion
@JsonProperty("recordAvailabilityLagMs") Map<String, String> recordAvailabilityLagMs) {
```
##########
pinot-common/src/main/java/org/apache/pinot/common/restlet/resources/SegmentConsumerInfo.java:
##########
@@ -68,23 +68,28 @@ public PartitionOffsetInfo getPartitionOffsetInfo() {
@JsonIgnoreProperties(ignoreUnknown = true)
static public class PartitionOffsetInfo {
- @JsonProperty("currentOffsets")
- public Map<String, String> _currentOffsets;
-
- @JsonProperty("recordsLag")
- public Map<String, String> _recordsLag;
-
- @JsonProperty("latestUpstreamOffsets")
- public Map<String, String> _latestUpstreamOffsets;
-
- public PartitionOffsetInfo(
- @JsonProperty("currentOffsets") Map<String, String> currentOffsets,
- @JsonProperty("latestUpstreamOffsets") Map<String, String> latestUpstreamOffsets,
- @JsonProperty("recordsLag") Map<String, String> recordsLag) {
- _currentOffsets = currentOffsets;
- _latestUpstreamOffsets = latestUpstreamOffsets;
- _recordsLag = recordsLag;
- }
+ @JsonProperty("currentOffsets")
+ public Map<String, String> _currentOffsets;
+
+ @JsonProperty("recordsLag")
+ public Map<String, String> _recordsLag;
+
+ @JsonProperty("latestUpstreamOffsets")
+ public Map<String, String> _latestUpstreamOffsets;
+
+ @JsonProperty("recordsAvailabilityLag")
+ public Map<String, String> _availabilityLag;
Review Comment:
```suggestion
@JsonProperty("recordAvailabilityLagMs")
public Map<String, String> _recordAvailabilityLagMs;
```
--
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-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [pinot] navina commented on a diff in pull request #9621: Add record availability lag for Kafka connector
Posted by GitBox <gi...@apache.org>.
navina commented on code in PR #9621:
URL: https://github.com/apache/pinot/pull/9621#discussion_r1000977599
##########
pinot-spi/src/main/java/org/apache/pinot/spi/stream/PartitionLagState.java:
##########
@@ -27,12 +27,20 @@ public class PartitionLagState {
protected final static String NOT_CALCULATED = "NOT_CALCULATED";
/**
- * Defines how far away the current record's offset / pointer is from upstream latest record
+ * Defines how far behind the current record's offset / pointer is from upstream latest record
* The distance is based on actual record count.
*/
public String getRecordsLag() {
return NOT_CALCULATED;
}
- // TODO: Define record availability lag ($latest_record_consumption_time - $latest_record_ingestion_time)
+ /**
+ * Defines how soon after record ingestion was the record consumed by Pinot. That is, the difference between the
+ * time the record was consumed and the time at which the record was ingested upstream.
+ *
+ * @return Lag value in milliseconds
+ */
+ public String getRecordAvailabilityLag() {
Review Comment:
another alternative:
rename `recordAvailabilityLagMs` -> `consumeLagMs` and use `recordTimeLagMs` for `millisBehindLatest` type of metric
--
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-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [pinot] codecov-commenter commented on pull request #9621: Add record availability lag for Kafka connector
Posted by GitBox <gi...@apache.org>.
codecov-commenter commented on PR #9621:
URL: https://github.com/apache/pinot/pull/9621#issuecomment-1282909846
# [Codecov](https://codecov.io/gh/apache/pinot/pull/9621?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) Report
> Merging [#9621](https://codecov.io/gh/apache/pinot/pull/9621?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) (b13c20f) into [master](https://codecov.io/gh/apache/pinot/commit/d580bbcaec42281c7bc0667c4db228354b56a889?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) (d580bbc) will **decrease** coverage by `9.21%`.
> The diff coverage is `5.66%`.
```diff
@@ Coverage Diff @@
## master #9621 +/- ##
============================================
- Coverage 35.02% 25.80% -9.22%
+ Complexity 189 44 -145
============================================
Files 1943 1931 -12
Lines 103817 103488 -329
Branches 15743 15706 -37
============================================
- Hits 36361 26708 -9653
- Misses 64369 74072 +9703
+ Partials 3087 2708 -379
```
| Flag | Coverage Δ | |
|---|---|---|
| integration1 | `25.80% <5.66%> (-0.04%)` | :arrow_down: |
| unittests2 | `?` | |
Flags with carried forward coverage won't be shown. [Click here](https://docs.codecov.io/docs/carryforward-flags?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#carryforward-flags-in-the-pull-request-comment) to find out more.
| [Impacted Files](https://codecov.io/gh/apache/pinot/pull/9621?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) | Coverage Δ | |
|---|---|---|
| [.../common/restlet/resources/SegmentConsumerInfo.java](https://codecov.io/gh/apache/pinot/pull/9621/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-cGlub3QtY29tbW9uL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9waW5vdC9jb21tb24vcmVzdGxldC9yZXNvdXJjZXMvU2VnbWVudENvbnN1bWVySW5mby5qYXZh) | `0.00% <0.00%> (ø)` | |
| [...ot/controller/util/ConsumingSegmentInfoReader.java](https://codecov.io/gh/apache/pinot/pull/9621/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-cGlub3QtY29udHJvbGxlci9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcGlub3QvY29udHJvbGxlci91dGlsL0NvbnN1bWluZ1NlZ21lbnRJbmZvUmVhZGVyLmphdmE=) | `0.00% <0.00%> (-89.14%)` | :arrow_down: |
| [...ugin/stream/kafka20/KafkaConsumerPartitionLag.java](https://codecov.io/gh/apache/pinot/pull/9621/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-cGlub3QtcGx1Z2lucy9waW5vdC1zdHJlYW0taW5nZXN0aW9uL3Bpbm90LWthZmthLTIuMC9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcGlub3QvcGx1Z2luL3N0cmVhbS9rYWZrYTIwL0thZmthQ29uc3VtZXJQYXJ0aXRpb25MYWcuamF2YQ==) | `0.00% <0.00%> (ø)` | |
| [...in/stream/kafka20/KafkaStreamMetadataProvider.java](https://codecov.io/gh/apache/pinot/pull/9621/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-cGlub3QtcGx1Z2lucy9waW5vdC1zdHJlYW0taW5nZXN0aW9uL3Bpbm90LWthZmthLTIuMC9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcGlub3QvcGx1Z2luL3N0cmVhbS9rYWZrYTIwL0thZmthU3RyZWFtTWV0YWRhdGFQcm92aWRlci5qYXZh) | `25.80% <0.00%> (-28.74%)` | :arrow_down: |
| [...lugin/stream/kafka20/server/KafkaDataProducer.java](https://codecov.io/gh/apache/pinot/pull/9621/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-cGlub3QtcGx1Z2lucy9waW5vdC1zdHJlYW0taW5nZXN0aW9uL3Bpbm90LWthZmthLTIuMC9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcGlub3QvcGx1Z2luL3N0cmVhbS9rYWZrYTIwL3NlcnZlci9LYWZrYURhdGFQcm9kdWNlci5qYXZh) | `32.60% <0.00%> (-0.73%)` | :arrow_down: |
| [...ache/pinot/server/api/resources/DebugResource.java](https://codecov.io/gh/apache/pinot/pull/9621/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-cGlub3Qtc2VydmVyL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9waW5vdC9zZXJ2ZXIvYXBpL3Jlc291cmNlcy9EZWJ1Z1Jlc291cmNlLmphdmE=) | `0.00% <0.00%> (ø)` | |
| [...che/pinot/server/api/resources/TablesResource.java](https://codecov.io/gh/apache/pinot/pull/9621/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-cGlub3Qtc2VydmVyL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9waW5vdC9zZXJ2ZXIvYXBpL3Jlc291cmNlcy9UYWJsZXNSZXNvdXJjZS5qYXZh) | `27.31% <0.00%> (-0.25%)` | :arrow_down: |
| [...pache/pinot/spi/stream/ConsumerPartitionState.java](https://codecov.io/gh/apache/pinot/pull/9621/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-cGlub3Qtc3BpL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9waW5vdC9zcGkvc3RyZWFtL0NvbnN1bWVyUGFydGl0aW9uU3RhdGUuamF2YQ==) | `0.00% <0.00%> (ø)` | |
| [...org/apache/pinot/spi/stream/PartitionLagState.java](https://codecov.io/gh/apache/pinot/pull/9621/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-cGlub3Qtc3BpL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9waW5vdC9zcGkvc3RyZWFtL1BhcnRpdGlvbkxhZ1N0YXRlLmphdmE=) | `0.00% <0.00%> (ø)` | |
| [...manager/realtime/LLRealtimeSegmentDataManager.java](https://codecov.io/gh/apache/pinot/pull/9621/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-cGlub3QtY29yZS9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcGlub3QvY29yZS9kYXRhL21hbmFnZXIvcmVhbHRpbWUvTExSZWFsdGltZVNlZ21lbnREYXRhTWFuYWdlci5qYXZh) | `58.10% <60.00%> (+2.66%)` | :arrow_up: |
| ... and [358 more](https://codecov.io/gh/apache/pinot/pull/9621/diff?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) | |
:mega: We’re building smart automated test selection to slash your CI/CD build times. [Learn more](https://about.codecov.io/iterative-testing/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation)
--
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-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [pinot] navina commented on pull request #9621: Add record availability lag for Kafka connector
Posted by GitBox <gi...@apache.org>.
navina commented on PR #9621:
URL: https://github.com/apache/pinot/pull/9621#issuecomment-1283692055
@npawar / @KKcorps Please review!
I seem to have added [docs](https://app.gitbook.com/o/-LtRX9NwSr7Ga7zA4piL/s/-LtH6nl58DdnZnelPdTc-887967055/~/changes/T80vSbcXHeWkuMl9KzaN/) before merging the code :))
--
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-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [pinot] Jackie-Jiang commented on pull request #9621: Add record availability lag for Kafka connector
Posted by GitBox <gi...@apache.org>.
Jackie-Jiang commented on PR #9621:
URL: https://github.com/apache/pinot/pull/9621#issuecomment-1291443103
We may also pick a shorter name for them to be more concise, but let's keep the name consistent across classes:
- `recordsLag`: lag in records (offsets)
- `availabilityLagMs`: availability lag in millis
- `timeLagMs`: time lag in millis
--
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-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [pinot] Jackie-Jiang commented on a diff in pull request #9621: Add record availability lag for Kafka connector
Posted by GitBox <gi...@apache.org>.
Jackie-Jiang commented on code in PR #9621:
URL: https://github.com/apache/pinot/pull/9621#discussion_r1002381957
##########
pinot-spi/src/main/java/org/apache/pinot/spi/stream/PartitionLagState.java:
##########
@@ -27,12 +27,20 @@ public class PartitionLagState {
protected final static String NOT_CALCULATED = "NOT_CALCULATED";
/**
- * Defines how far away the current record's offset / pointer is from upstream latest record
+ * Defines how far behind the current record's offset / pointer is from upstream latest record
* The distance is based on actual record count.
*/
public String getRecordsLag() {
return NOT_CALCULATED;
}
- // TODO: Define record availability lag ($latest_record_consumption_time - $latest_record_ingestion_time)
+ /**
+ * Defines how soon after record ingestion was the record consumed by Pinot. That is, the difference between the
+ * time the record was consumed and the time at which the record was ingested upstream.
+ *
+ * @return Lag value in milliseconds
+ */
+ public String getRecordAvailabilityLag() {
Review Comment:
I see. `recordAvailabilityLagMs` and `recordTimeLagMs` sounds good to me.
--
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-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [pinot] Jackie-Jiang merged pull request #9621: Add record availability lag for Kafka connector
Posted by GitBox <gi...@apache.org>.
Jackie-Jiang merged PR #9621:
URL: https://github.com/apache/pinot/pull/9621
--
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-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org