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_r216376586
##########
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);
+ }
+ } catch (Exception e) {
+ LOG.warn("An exception occurred while
processing a message.", e);
Review comment:
`"An exception occurred while processing a metric message."`?
----------------------------------------------------------------
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:
[email protected]
With regards,
Apache Git Services
