NIFI-16380 Add Records Metrics to ConsumeKinesis - #11712
exceptionfactory wants to merge 2 commits into
Conversation
| } | ||
| } | ||
|
|
||
| if (recordCount == 0) { |
There was a problem hiding this comment.
Should the lag gauge be recorded independently of recordCount, using the shard lag observations, so an empty successful poll can update a previously positive lag to zero?
There was a problem hiding this comment.
That's a good point, will adjust to always report the milliseconds behind regardless of record count
| final Map<String, String> attributes = getMetricAttributes(streamName, shardId); | ||
| session.adjustCounter(KinesisMetricName.RECORDS_CONSUMED.getMetricName(), recordCount, attributes, CommitTiming.NOW); | ||
| session.adjustCounter(KinesisMetricName.BYTES_CONSUMED.getMetricName(), bytesConsumed, attributes, CommitTiming.NOW); | ||
| // Kinesis uses -1 when millisBehindLatest is absent so record that as 0 |
There was a problem hiding this comment.
Should an absent MillisBehindLatest value skip the gauge instead of converting the -1 sentinel to zero?
There was a problem hiding this comment.
I considered that approach, but decided to set 0 as a definitive indicator. With this approach, the milliseconds behind is always reported with everything else, and 0 indicates that the consumer is not behind, versus having to infer it from the lack of a change. Are there reasons you think not reporting would be better?
There was a problem hiding this comment.
My concern is that AWS defines 0 as an observed caught-up state, while -1 means no MillisBehindLatest value was provided. Converting unknown to zero would report that the consumer is caught up without evidence, so I think the gauge should be skipped until a real value is observed. But, to be fair, this should be an exceptional situation (I assume) so I don't feel strongly about it and would be OK with the current approach.
There was a problem hiding this comment.
Thanks, that makes sense. On balance, it seems better to mirror the indicator that AWS provides, so I will switch this to check for -1 and not report if found
| } | ||
|
|
||
| @Test | ||
| void testEmptyConsumeDoesNotRecordMetrics() throws Exception { |
There was a problem hiding this comment.
Can we add a test where lag changes from positive to zero on an empty successful response, to verify that the gauge reports the caught-up state?
|
Thanks for the review @pvillard31, I pushed an update to record |
2 participants
Summary
NIFI-16380 Adds the following metrics to the
ConsumeKinesisProcessor:The implementation follows the pattern of NIFI-16375 adding similar metrics to the
ConsumeKafkaProcessor. The primary difference is theconsumer.milliseconds.behind, which tracks the MillisBehindLatest from AWS Kinesis, providing a rough gauge of latency depending on the expected behavior of the Kinesis stream.The implementation includes the Stream Name and Shard ID as attributes, using the same attribute names already used for FlowFiles.
Tracking
Please complete the following tracking steps prior to pull request creation.
Issue Tracking
Pull Request Tracking
NIFI-00000NIFI-00000VerifiedstatusPull Request Formatting
mainbranchVerification
Please indicate the verification steps performed prior to pull request creation.
Build
./mvnw clean install -P contrib-checkLicensing
LICENSEandNOTICEfilesDocumentation