-
Notifications
You must be signed in to change notification settings - Fork 1k
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
fix: NPE when udf metrics enabled (#5960)
* fix: NPE when udf metrics enabled fixes: #5890 Fixes an NPE thrown when the second instance of a UDAF is initialized. The NPE was being thrown due to a bug in the code that failed to initialise the sensor fields in the `UdafAggregateFunction` class if the sensor already existed. The commit also refactors the code used to add function invocation metrics such that all three function types: UDF, UDTF and UDAF, use the same function to register metrics. Unit tests added to ensure this single code path doesn't throw an NPE or return null on the subsequent invocation. Also added `ksql.udf.collect.metrics` metric to the server config file, so users are more likely to be able to find it and try it out. Moved UDTF functions out of the `ksql-udf` group and into `ksql-udtf` group. Having them in the former was another bug. Co-authored-by: Andy Coates <big-andy-coates@users.noreply.github.com>
- Loading branch information
1 parent
ce2b6e4
commit e32183f
Showing
9 changed files
with
325 additions
and
167 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
106 changes: 106 additions & 0 deletions
106
ksqldb-engine/src/main/java/io/confluent/ksql/function/FunctionMetrics.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,106 @@ | ||
| /* | ||
| * Copyright 2020 Confluent Inc. | ||
| * | ||
| * Licensed under the Confluent Community License (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.confluent.io/confluent-community-license | ||
| * | ||
| * Unless required by applicable law or agreed to in writing, software | ||
| * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT | ||
| * WARRANTIES OF ANY KIND, either express or implied. See the License for the | ||
| * specific language governing permissions and limitations under the License. | ||
| */ | ||
|
|
||
| package io.confluent.ksql.function; | ||
|
|
||
| import java.util.Optional; | ||
| import java.util.concurrent.TimeUnit; | ||
| import java.util.function.BiFunction; | ||
| import org.apache.kafka.common.MetricName; | ||
| import org.apache.kafka.common.metrics.Metrics; | ||
| import org.apache.kafka.common.metrics.Sensor; | ||
| import org.apache.kafka.common.metrics.stats.Avg; | ||
| import org.apache.kafka.common.metrics.stats.Max; | ||
| import org.apache.kafka.common.metrics.stats.Rate; | ||
| import org.apache.kafka.common.metrics.stats.WindowedCount; | ||
|
|
||
| public final class FunctionMetrics { | ||
|
|
||
| static final String AVG_DESC = "Average time for invocations of the %s"; | ||
| static final String MAX_DESC = "Max time for invocations of the %s"; | ||
| static final String COUNT_DESC = "Total number of invocations of the %s"; | ||
| static final String RATE_DESC = "The rate of invocations (invocations per second) of the %s"; | ||
|
|
||
| private FunctionMetrics() { | ||
| } | ||
|
|
||
| /** | ||
| * Gets an existing invocation sensor, or creates one if needed. | ||
| * | ||
| * <p>Sensor created with avg, max, count and rate metrics. | ||
| * | ||
| * @param metrics the metrics service. | ||
| * @param sensorName the name of the sensor | ||
| * @param groupName the name of the group | ||
| * @param functionDescription the description of the function. | ||
| */ | ||
| public static void initInvocationSensor( | ||
| final Optional<Metrics> metrics, | ||
| final String sensorName, | ||
| final String groupName, | ||
| final String functionDescription | ||
| ) { | ||
| metrics.ifPresent(m -> getInvocationSensor(m, sensorName, groupName, functionDescription)); | ||
| } | ||
|
|
||
| /** | ||
| * Gets an existing invocation sensor, or creates one if needed. | ||
| * | ||
| * <p>Sensor created with avg, max, count and rate metrics. | ||
| * | ||
| * @param metrics the metrics service. | ||
| * @param sensorName the name of the sensor | ||
| * @param groupName the name of the group | ||
| * @param functionDescription the description of the function. | ||
| */ | ||
| public static Sensor getInvocationSensor( | ||
| final Metrics metrics, | ||
| final String sensorName, | ||
| final String groupName, | ||
| final String functionDescription | ||
| ) { | ||
| final Sensor sensor = metrics.sensor(sensorName); | ||
| if (sensor.hasMetrics()) { | ||
| return sensor; | ||
| } | ||
|
|
||
| final BiFunction<String, String, MetricName> metricNamer = (suffix, descPattern) -> { | ||
| final String description = String.format(descPattern, functionDescription); | ||
| return metrics.metricName(sensorName + "-" + suffix, groupName, description); | ||
| }; | ||
|
|
||
| sensor.add( | ||
| metricNamer.apply("avg", AVG_DESC), | ||
| new Avg() | ||
| ); | ||
|
|
||
| sensor.add( | ||
| metricNamer.apply("max", MAX_DESC), | ||
| new Max() | ||
| ); | ||
|
|
||
| sensor.add( | ||
| metricNamer.apply("count", COUNT_DESC), | ||
| new WindowedCount() | ||
| ); | ||
|
|
||
| sensor.add( | ||
| metricNamer.apply("rate", RATE_DESC), | ||
| new Rate(TimeUnit.SECONDS, new WindowedCount()) | ||
| ); | ||
|
|
||
| return sensor; | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.