Github user zentol commented on a diff in the pull request:
https://github.com/apache/flink/pull/5182#discussion_r157907331
--- Diff:
flink-connectors/flink-connector-kinesis/src/main/java/org/apache/flink/streaming/connectors/kinesis/internals/ShardConsumer.java
---
@@ -96,6 +100,14 @@ protected ShardConsumer(KinesisDataFetcher<T>
fetcherRef,
SequenceNumber
lastSequenceNum,
KinesisProxyInterface
kinesis) {
this.fetcherRef = checkNotNull(fetcherRef);
+ MetricGroup kinesisMetricGroup = fetcherRef.getRuntimeContext()
+ .getMetricGroup()
+ .addGroup("Kinesis");
+ kinesisMetricGroup
+ .getAllVariables()
+ .put("<shard_id>",
subscribedShard.getShard().getShardId());
+
+ kinesisMetricGroup.gauge("millisBehindLatest", (Gauge<Long>) ()
-> millisBehindLatest);
--- End diff --
the cast shouldn't be necessary
---
