[
https://issues.apache.org/jira/browse/FLINK-10247?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16609426#comment-16609426
]
ASF GitHub Bot commented on FLINK-10247:
----------------------------------------
yanghua commented on a change in pull request #6676: [FLINK-10247][Metrics] Run
MetricQueryService in separate thread pool
URL: https://github.com/apache/flink/pull/6676#discussion_r216376070
##########
File path:
flink-runtime/src/main/java/org/apache/flink/runtime/metrics/dump/MetricQueryService.java
##########
@@ -221,4 +193,75 @@ public static Object getCreateDump() {
private static class CreateDump implements Serializable {
private static final CreateDump INSTANCE = new CreateDump();
}
+
+ /**
+ * This runnable executes add metric, remove metric and create dump
logic after notified.
+ */
+ private static final class MetricMessageHandlerRunnable implements
Runnable {
+ private final Object message;
+ private final Map<Gauge<?>, Tuple2<QueryScopeInfo, String>>
gauges;
+ private final Map<Counter, Tuple2<QueryScopeInfo, String>>
counters;
+ private final Map<Histogram, Tuple2<QueryScopeInfo, String>>
histograms;
+ private final Map<Meter, Tuple2<QueryScopeInfo, String>> meters;
+ private final ActorRef sender;
+ private final ActorRef self;
+ private final MetricDumpSerializer serializer;
+
+ public MetricMessageHandlerRunnable(Object message,
Map<Gauge<?>, Tuple2<QueryScopeInfo, String>> gauges,
+ Map<Counter, Tuple2<QueryScopeInfo, String>> counters,
Map<Histogram, Tuple2<QueryScopeInfo, String>> histograms,
+ Map<Meter, Tuple2<QueryScopeInfo, String>> meters,
ActorRef sender, ActorRef self,
+ MetricDumpSerializer serializer) {
+ this.message = message;
+ this.gauges = gauges;
+ this.counters = counters;
+ this.histograms = histograms;
+ this.meters = meters;
+ this.sender = sender;
+ this.self = self;
+ this.serializer = serializer;
+ }
+
+ @Override public void run() {
+ try {
+ if (message instanceof AddMetric) {
+ AddMetric added = (AddMetric) message;
+
+ String metricName = added.metricName;
+ Metric metric = added.metric;
+ AbstractMetricGroup group = added.group;
+
+ QueryScopeInfo info =
group.getQueryServiceMetricInfo(FILTER);
+
+ if (metric instanceof Counter) {
+ counters.put((Counter) metric,
new Tuple2<>(info, FILTER.filterCharacters(metricName)));
+ } else if (metric instanceof Gauge) {
+ gauges.put((Gauge<?>) metric,
new Tuple2<>(info, FILTER.filterCharacters(metricName)));
+ } else if (metric instanceof Histogram)
{
+ histograms.put((Histogram)
metric, new Tuple2<>(info, FILTER.filterCharacters(metricName)));
+ } else if (metric instanceof Meter) {
+ meters.put((Meter) metric, new
Tuple2<>(info, FILTER.filterCharacters(metricName)));
+ }
+ } else if (message instanceof RemoveMetric) {
+ Metric metric = (((RemoveMetric)
message).metric);
+ if (metric instanceof Counter) {
+ this.counters.remove(metric);
+ } else if (metric instanceof Gauge) {
+ this.gauges.remove(metric);
+ } else if (metric instanceof Histogram)
{
+ this.histograms.remove(metric);
+ } else if (metric instanceof Meter) {
+ this.meters.remove(metric);
+ }
+ } else if (message instanceof CreateDump) {
+
MetricDumpSerialization.MetricSerializationResult dump =
serializer.serialize(counters, gauges, histograms, meters);
+ sender.tell(dump, self);
+ } else {
+ LOG.warn("MetricQueryServiceActor
received an invalid message. " + message.toString());
+ sender.tell(new Status.Failure(new
IOException("MetricQueryServiceActor received an invalid message. " +
message.toString())), self);
Review comment:
the same
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
> Run MetricQueryService in separate thread pool
> ----------------------------------------------
>
> Key: FLINK-10247
> URL: https://issues.apache.org/jira/browse/FLINK-10247
> Project: Flink
> Issue Type: Sub-task
> Components: Metrics
> Affects Versions: 1.5.3, 1.6.0, 1.7.0
> Reporter: Till Rohrmann
> Assignee: Shimin Yang
> Priority: Critical
> Labels: pull-request-available
> Fix For: 1.6.1, 1.7.0, 1.5.4
>
>
> In order to make the {{MetricQueryService}} run independently of the main
> Flink components, it should get its own dedicated thread pool assigned.
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
