KAFKA-15778 & KAFKA-15779: Implement metrics manager (KIP-714)

checkstyle/import-control-core.xml

@@ -85,6 +85,12 @@
     <allow pkg="org.apache.kafka.test" />
   </subpackage>
 
+  <subpackage name="metrics">
+    <allow class="org.apache.kafka.server.authorizer.AuthorizableRequestContext" />
+    <allow pkg="org.apache.kafka.server.telemetry" />
+    <allow pkg="kafka.metrics" />
+  </subpackage>
+
   <subpackage name="server">
     <allow pkg="kafka" />
     <allow pkg="org.apache.kafka" />

clients/src/main/java/org/apache/kafka/common/errors/TelemetryTooLargeException.java

@@ -0,0 +1,28 @@
+/*
+ * 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.kafka.common.errors;
+
+/**
+ * This exception indicates that the size of the telemetry metrics data is too large.
+ */
+public class TelemetryTooLargeException extends ApiException {
+
+    public TelemetryTooLargeException(String message) {
+        super(message);
+    }
+}
+

clients/src/main/java/org/apache/kafka/common/errors/UnknownSubscriptionIdException.java

@@ -0,0 +1,28 @@
+/*
+ * 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.kafka.common.errors;
+
+/**
+ * This exception indicates that the client sent an invalid or outdated SubscriptionId
+ */
+public class UnknownSubscriptionIdException extends ApiException {
+
+    public UnknownSubscriptionIdException(String message) {
+        super(message);
+    }
+}
+

clients/src/main/java/org/apache/kafka/common/protocol/Errors.java

@@ -20,6 +20,7 @@
 import org.apache.kafka.common.errors.ApiException;
 import org.apache.kafka.common.errors.BrokerIdNotRegisteredException;
 import org.apache.kafka.common.errors.BrokerNotAvailableException;
+import org.apache.kafka.common.errors.TelemetryTooLargeException;
 import org.apache.kafka.common.errors.ClusterAuthorizationException;
 import org.apache.kafka.common.errors.ConcurrentTransactionsException;
 import org.apache.kafka.common.errors.ControllerMovedException;
@@ -118,6 +119,7 @@
 import org.apache.kafka.common.errors.TransactionCoordinatorFencedException;
 import org.apache.kafka.common.errors.TransactionalIdAuthorizationException;
 import org.apache.kafka.common.errors.TransactionalIdNotFoundException;
+import org.apache.kafka.common.errors.UnknownSubscriptionIdException;
 import org.apache.kafka.common.errors.UnacceptableCredentialException;
 import org.apache.kafka.common.errors.UnknownControllerIdException;
 import org.apache.kafka.common.errors.UnknownLeaderEpochException;
@@ -386,7 +388,9 @@ public enum Errors {
     STALE_MEMBER_EPOCH(113, "The member epoch is stale. The member must retry after receiving its updated member epoch via the ConsumerGroupHeartbeat API.", StaleMemberEpochException::new),
     MISMATCHED_ENDPOINT_TYPE(114, "The request was sent to an endpoint of the wrong type.", MismatchedEndpointTypeException::new),
     UNSUPPORTED_ENDPOINT_TYPE(115, "This endpoint type is not supported yet.", UnsupportedEndpointTypeException::new),
-    UNKNOWN_CONTROLLER_ID(116, "This controller ID is not known.", UnknownControllerIdException::new);
+    UNKNOWN_CONTROLLER_ID(116, "This controller ID is not known.", UnknownControllerIdException::new),
+    UNKNOWN_SUBSCRIPTION_ID(117, "Client sent a push telemetry request with an invalid or outdated subscription ID", UnknownSubscriptionIdException::new),
+    TELEMETRY_TOO_LARGE(118, "Client sent a push telemetry request larger than the maximum size the broker will accept", TelemetryTooLargeException::new);
 
     private static final Logger log = LoggerFactory.getLogger(Errors.class);
 

clients/src/main/java/org/apache/kafka/common/requests/PushTelemetryRequest.java

@@ -14,19 +14,21 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-
 package org.apache.kafka.common.requests;
 
 import org.apache.kafka.common.message.PushTelemetryRequestData;
 import org.apache.kafka.common.message.PushTelemetryResponseData;
 import org.apache.kafka.common.protocol.ApiKeys;
 import org.apache.kafka.common.protocol.ByteBufferAccessor;
 import org.apache.kafka.common.protocol.Errors;
+import org.apache.kafka.common.record.CompressionType;
 
 import java.nio.ByteBuffer;
 
 public class PushTelemetryRequest extends AbstractRequest {
 
+    private static final String OTLP_CONTENT_TYPE = "OTLP";
+
     public static class Builder extends AbstractRequest.Builder<PushTelemetryRequest> {
 
         private final PushTelemetryRequestData data;
@@ -71,6 +73,31 @@ public PushTelemetryRequestData data() {
         return data;
     }
 
+    public PushTelemetryResponse createResponse(int throttleTimeMs, Errors errors) {
+        PushTelemetryResponseData responseData = new PushTelemetryResponseData();
+        responseData.setErrorCode(errors.code());
+        responseData.setThrottleTimeMs(throttleTimeMs);
+        return new PushTelemetryResponse(responseData);
+    }
+
+    public String getMetricsContentType() {
+        // Future versions of PushTelemetryRequest and GetTelemetrySubscriptionsRequest may include a content-type
+        // field to allow for updated OTLP format versions (or additional formats), but this field is currently not
+        // included since only one format is specified in the current proposal of the kip-714
+        return OTLP_CONTENT_TYPE;
+    }
+
+    public ByteBuffer getMetricsData() {
+        CompressionType cType = CompressionType.forId(this.data.compressionType());
+        return (cType == CompressionType.NONE) ?
+            ByteBuffer.wrap(this.data.metrics()) : decompressMetricsData(cType, this.data.metrics());
+    }
+
+    private static ByteBuffer decompressMetricsData(CompressionType compressionType, byte[] metrics) {
+        // TODO: Add support for decompression of metrics data
+        return ByteBuffer.wrap(metrics);
+    }
+
     public static PushTelemetryRequest parse(ByteBuffer buffer, short version) {
         return new PushTelemetryRequest(new PushTelemetryRequestData(
                 new ByteBufferAccessor(buffer), version), version);

core/src/main/java/kafka/metrics/ClientMetricsConfigs.java

@@ -16,13 +16,15 @@
  */
 package kafka.metrics;
 
+import org.apache.kafka.common.config.AbstractConfig;
 import org.apache.kafka.common.config.ConfigDef;
 import org.apache.kafka.common.config.ConfigDef.Importance;
 import org.apache.kafka.common.config.ConfigDef.Type;
 import org.apache.kafka.common.errors.InvalidConfigurationException;
 import org.apache.kafka.common.errors.InvalidRequestException;
 
 import java.util.Arrays;
+import java.util.Collections;
 import java.util.HashMap;
 import java.util.HashSet;
 import java.util.List;
@@ -67,7 +69,7 @@
  * For more information please look at kip-714:
  * https://cwiki.apache.org/confluence/display/KAFKA/KIP-714%3A+Client+metrics+and+observability#KIP714:Clientmetricsandobservability-Clientmetricsconfiguration
  */
-public class ClientMetricsConfigs {
+public class ClientMetricsConfigs extends AbstractConfig {
 
     public static final String SUBSCRIPTION_METRICS = "metrics";
     public static final String PUSH_INTERVAL_MS = "interval.ms";
@@ -80,6 +82,11 @@ public class ClientMetricsConfigs {
     public static final String CLIENT_SOURCE_ADDRESS = "client_source_address";
     public static final String CLIENT_SOURCE_PORT = "client_source_port";
 
+    // Empty string in client-metrics resource configs indicates that all the metrics are subscribed.
+    public static final String ALL_SUBSCRIBED_METRICS_CONFIG = "\"\"";
+    // Empty string representing all subscribed metrics in subscription response.
+    public static final String ALL_SUBSCRIBED_METRICS = "";
+
     public static final int DEFAULT_INTERVAL_MS = 5 * 60 * 1000; // 5 minutes
     private static final int MIN_INTERVAL_MS = 100; // 100ms
     private static final int MAX_INTERVAL_MS = 3600000; // 1 hour
@@ -94,9 +101,13 @@ public class ClientMetricsConfigs {
     ));
 
     private static final ConfigDef CONFIG = new ConfigDef()
-        .define(SUBSCRIPTION_METRICS, Type.LIST, Importance.MEDIUM, "Subscription metrics list")
-        .define(PUSH_INTERVAL_MS, Type.INT, Importance.MEDIUM, "Push interval in milliseconds")
-        .define(CLIENT_MATCH_PATTERN, Type.LIST, Importance.MEDIUM, "Client match pattern list");
+        .define(SUBSCRIPTION_METRICS, Type.LIST, Collections.emptyList(), Importance.MEDIUM, "Subscription metrics list")
+        .define(PUSH_INTERVAL_MS, Type.INT, DEFAULT_INTERVAL_MS, Importance.MEDIUM, "Push interval in milliseconds")
+        .define(CLIENT_MATCH_PATTERN, Type.LIST, Collections.emptyList(), Importance.MEDIUM, "Client match pattern list");
+
+    public ClientMetricsConfigs(Properties props) {
+        super(CONFIG, props);
+    }
 
     public static ConfigDef configDef() {
         return CONFIG;
@@ -143,46 +154,43 @@ private static void validateProperties(Properties properties) {
     /**
      * Parses the client matching patterns and builds a map with entries that has
      * (PatternName, PatternValue) as the entries.
-     *  Ex: "VERSION=1.2.3" would be converted to a map entry of (Version, 1.2.3)
+     * Ex: "VERSION=1.2.3" would be converted to a map entry of (Version, 1.2.3)
      * <p>
-     *  NOTES:
-     *  Client match pattern splits the input into two parts separated by first occurrence of the character '='
+     * NOTES:
+     * Client match pattern splits the input into two parts separated by first occurrence of the character '='
      *
-     *  @param patterns List of client matching pattern strings
+     * @param patterns List of client matching pattern strings
      * @return map of client matching pattern entries
      */
-    public static Map<String, String> parseMatchingPatterns(List<String> patterns) {
-        Map<String, String> patternsMap = new HashMap<>();
-        if (patterns != null) {
-            patterns.forEach(pattern -> {
-                String[] nameValuePair = pattern.split("=");
-                if (nameValuePair.length != 2) {
-                    throw new InvalidConfigurationException("Illegal client matching pattern: " + pattern);
-                }
-
-                String param = nameValuePair[0].trim();
-                String patternValue = nameValuePair[1].trim();
-                if (isValidParam(param) && isValidRegExPattern(patternValue)) {
-                    patternsMap.put(param, patternValue);
-                } else {
-                    throw new InvalidConfigurationException("Illegal client matching pattern: " + pattern);
-                }
-            });
+    public static Map<String, Pattern> parseMatchingPatterns(List<String> patterns) {
+        if (patterns == null || patterns.isEmpty()) {
+            return Collections.emptyMap();
         }
 
+        Map<String, Pattern> patternsMap = new HashMap<>();
+        patterns.forEach(pattern -> {
+            String[] nameValuePair = pattern.split("=");
+            if (nameValuePair.length != 2) {
+                throw new InvalidConfigurationException("Illegal client matching pattern: " + pattern);
+            }
+
+            String param = nameValuePair[0].trim();
+            if (!isValidParam(param)) {
+                throw new InvalidConfigurationException("Illegal client matching pattern: " + pattern);
+            }
+
+            try {
+                Pattern patternValue = Pattern.compile(nameValuePair[1].trim());
+                patternsMap.put(param, patternValue);
+            } catch (PatternSyntaxException e) {
+                throw new InvalidConfigurationException("Illegal client matching pattern: " + pattern);
+            }
+        });
+
         return patternsMap;
     }
 
     private static boolean isValidParam(String paramName) {
         return ALLOWED_MATCH_PARAMS.contains(paramName);
     }
-
-    private static boolean isValidRegExPattern(String inputPattern) {
-        try {
-            Pattern.compile(inputPattern);
-        } catch (PatternSyntaxException e) {
-            return false;
-        }
-        return true;
-    }
 }