AHeise commented on a change in pull request #15972: URL: https://github.com/apache/flink/pull/15972#discussion_r664275730
########## File path: flink-metrics/flink-metrics-core/src/main/java/org/apache/flink/metrics/groups/SourceMetricGroup.java ########## @@ -0,0 +1,71 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.flink.metrics.groups; + +import org.apache.flink.metrics.Counter; +import org.apache.flink.metrics.Gauge; +import org.apache.flink.metrics.SettableGauge; + +import javax.annotation.concurrent.NotThreadSafe; + +/** + * Pre-defined metrics for sources. + * + * <p>All metrics can only be accessed in the main operator thread. + */ +@NotThreadSafe +public interface SourceMetricGroup extends OperatorMetricGroup { + /** The total number of record that failed to consume, process, or emit. */ + Counter getNumRecordsInErrorsCounter(); + + /** + * Adds an optional gauge for last fetch time. Source readers can use this gauge to indicate the + * timestamp in milliseconds that Flink used to fetch a record. + * + * <p>The timestamp will be used to calculate the currentFetchEventTimeLag metric <code> + * currentFetchEventTimeLag = FetchTime - EventTime</code>. + * + * <p>Note that this time must strictly reflect the time of the last polled record. For sources + * that retrieve batches from the external system, the best way is to attach the timestamp to + * the batch and return the time of that batch. For multi-threaded sources, the timestamp should + * be embedded into the hand-over data structure. + * + * @see SettableGauge SettableGauge to continuously update the value. + */ + Gauge<Long> addLastFetchTimeGauge(Gauge<Long> lastFetchTimeGauge); Review comment: This argument holds more for `addPendingBytesGauge` and `addPendingRecordsGauge`. Here this gauge is used to supply half of the information needed to calculate `currentFetchEventTimeLag = FetchTime - EventTime`. So while I think it's okay to have `addLastFetchTimeGauge`, we can discuss if the pendingXGauge makes sense. Alternatively, we could also add the name constants to the interface? Another related question: do we want to return the gauges here? I don't really see the benefit but the same can be set about `MetricGroup#gauge`. -- 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: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
