KIP 714 with compression support (#4721)

KIP 714 with compression support (#4721)

implemented GetTelemetrySubscriptions and PushTelemetry to send client telemetry to the requesting broker.
Available metrics:

* producer.connection.creation.rate
* producer.connection.creation.total
* producer.node.request.latency.avg
* producer.node.request.latency.max
* producer.produce.throttle.time.avg
* producer.produce.throttle.time.max
* producer.record.queue.time.avg
* producer.record.queue.time.max
* consumer.connection.creation.rate
* consumer.connection.creation.total
* consumer.node.request.latency.avg
* consumer.node.request.latency.max
* consumer.coordinator.assigned.partitions

Compression is supported with zstd, zlib, lz4, or snappy.

---------

Co-authored-by: Milind L <[email protected]>

.formatignore

@@ -18,3 +18,16 @@ src/snappy_compat.h
 src/tinycthread.c
 src/tinycthread.h
 src/regexp.h
+src/nanopb/pb_common.c
+src/nanopb/pb_common.h
+src/nanopb/pb_decode.c
+src/nanopb/pb_decode.h
+src/nanopb/pb_encode.c
+src/nanopb/pb_encode.h
+src/nanopb/pb.h
+src/opentelemetry/common.pb.c
+src/opentelemetry/common.pb.h
+src/opentelemetry/metrics.pb.c
+src/opentelemetry/metrics.pb.h
+src/opentelemetry/resource.pb.c
+src/opentelemetry/resource.pb.h

CHANGELOG.md

@@ -8,6 +8,8 @@ librdkafka v2.5.0 is a feature release.
 * Fix for an idempotent producer error, with a message batch not reconstructed
   identically when retried (#4750)
 * Removed support for CentOS 6 and CentOS 7 (#4775).
+* [KIP-714](https://cwiki.apache.org/confluence/display/KAFKA/KIP-714%3A+Client+metrics+and+observability) Client 
+  metrics and observability (#4721).
 
 ## Upgrade considerations
 

CONFIGURATION.md

@@ -19,7 +19,7 @@ topic.metadata.refresh.fast.cnt          |  *  | 0 .. 1000       |            10
 topic.metadata.refresh.sparse            |  *  | true, false     |          true | low        | Sparse metadata requests (consumes less network bandwidth) <br>*Type: boolean*
 topic.metadata.propagation.max.ms        |  *  | 0 .. 3600000    |         30000 | low        | Apache Kafka topic creation is asynchronous and it takes some time for a new topic to propagate throughout the cluster to all brokers. If a client requests topic metadata after manual topic creation but before the topic has been fully propagated to the broker the client is requesting metadata from, the topic will seem to be non-existent and the client will mark the topic as such, failing queued produced messages with `ERR__UNKNOWN_TOPIC`. This setting delays marking a topic as non-existent until the configured propagation max time has passed. The maximum propagation time is calculated from the time the topic is first referenced in the client, e.g., on produce(). <br>*Type: integer*
 topic.blacklist                          |  *  |                 |               | low        | Topic blacklist, a comma-separated list of regular expressions for matching topic names that should be ignored in broker metadata information as if the topics did not exist. <br>*Type: pattern list*
-debug                                    |  *  | generic, broker, topic, metadata, feature, queue, msg, protocol, cgrp, security, fetch, interceptor, plugin, consumer, admin, eos, mock, assignor, conf, all |               | medium     | A comma-separated list of debug contexts to enable. Detailed Producer debugging: broker,topic,msg. Consumer: consumer,cgrp,topic,fetch <br>*Type: CSV flags*
+debug                                    |  *  | generic, broker, topic, metadata, feature, queue, msg, protocol, cgrp, security, fetch, interceptor, plugin, consumer, admin, eos, mock, assignor, conf, telemetry, all |               | medium     | A comma-separated list of debug contexts to enable. Detailed Producer debugging: broker,topic,msg. Consumer: consumer,cgrp,topic,fetch <br>*Type: CSV flags*
 socket.timeout.ms                        |  *  | 10 .. 300000    |         60000 | low        | Default timeout for network requests. Producer: ProduceRequests will use the lesser value of `socket.timeout.ms` and remaining `message.timeout.ms` for the first message in the batch. Consumer: FetchRequests will use `fetch.wait.max.ms` + `socket.timeout.ms`. Admin: Admin requests will use `socket.timeout.ms` or explicitly set `rd_kafka_AdminOptions_set_operation_timeout()` value. <br>*Type: integer*
 socket.blocking.max.ms                   |  *  | 1 .. 60000      |          1000 | low        | **DEPRECATED** No longer used. <br>*Type: integer*
 socket.send.buffer.bytes                 |  *  | 0 .. 100000000  |             0 | low        | Broker socket send buffer size. System default is used if 0. <br>*Type: integer*
@@ -156,6 +156,7 @@ dr_cb                                    |  P  |                 |
 dr_msg_cb                                |  P  |                 |               | low        | Delivery report callback (set with rd_kafka_conf_set_dr_msg_cb()) <br>*Type: see dedicated API*
 sticky.partitioning.linger.ms            |  P  | 0 .. 900000     |            10 | low        | Delay in milliseconds to wait to assign new sticky partitions for each topic. By default, set to double the time of linger.ms. To disable sticky behavior, set to 0. This behavior affects messages with the key NULL in all cases, and messages with key lengths of zero when the consistent_random partitioner is in use. These messages would otherwise be assigned randomly. A higher value allows for more effective batching of these messages. <br>*Type: integer*
 client.dns.lookup                        |  *  | use_all_dns_ips, resolve_canonical_bootstrap_servers_only | use_all_dns_ips | low        | Controls how the client uses DNS lookups. By default, when the lookup returns multiple IP addresses for a hostname, they will all be attempted for connection before the connection is considered failed. This applies to both bootstrap and advertised servers. If the value is set to `resolve_canonical_bootstrap_servers_only`, each entry will be resolved and expanded into a list of canonical names. **WARNING**: `resolve_canonical_bootstrap_servers_only` must only be used with `GSSAPI` (Kerberos) as `sasl.mechanism`, as it's the only purpose of this configuration value. **NOTE**: Default here is different from the Java client's default behavior, which connects only to the first IP address returned for a hostname.  <br>*Type: enum value*
+enable.metrics.push                      |  *  | true, false     |          true | low        | Whether to enable pushing of client metrics to the cluster, if the cluster has a client metrics subscription which matches this client <br>*Type: boolean*
 
 
 ## Topic configuration properties

INTRODUCTION.md

@@ -2051,6 +2051,7 @@ The [Apache Kafka Implementation Proposals (KIPs)](https://cwiki.apache.org/conf
 | KIP-602 - Use all resolved addresses by default                          | 2.6.0                       | Supported                                                                                     |
 | KIP-651 - Support PEM format for SSL certs and keys                      | 2.7.0                       | Supported                                                                                     |
 | KIP-654 - Aborted txns with non-flushed msgs should not be fatal         | 2.7.0                       | Supported                                                                                     |
+| KIP-714 - Client metrics and observability                               | 3.7.0                       | Supported                                                                                     |
 | KIP-735 - Increase default consumer session timeout                      | 3.0.0                       | Supported                                                                                     |
 | KIP-768 - SASL/OAUTHBEARER OIDC support                                  | 3.0                         | Supported                                                                                     |
 | KIP-881 - Rack-aware Partition Assignment for Kafka Consumers            | 3.5.0                       | Supported                                                                                     |
@@ -2106,6 +2107,8 @@ release of librdkafka.
 | 50      | DescribeUserScramCredentials  | 0          | 0              |
 | 51      | AlterUserScramCredentials     | 0          | 0              |
 | 68      | ConsumerGroupHeartbeat        | 0          | 0              |
+| 71      | GetTelemetrySubscriptions     | 0          | 0              |
+| 72      | PushTelemetry                 | 0          | 0              |
 
 # Recommendations for language binding developers
 

LICENSE.nanopb

@@ -0,0 +1,22 @@
+For files in src/nanopb : https://github.com/nanopb/nanopb/blob/8ef41e0ebd45daaf19459a011f67e66224b247cd/LICENSE.txt
+
+Copyright (c) 2011 Petteri Aimonen <jpa at nanopb.mail.kapsi.fi>
+
+This software is provided 'as-is', without any express or 
+implied warranty. In no event will the authors be held liable 
+for any damages arising from the use of this software.
+
+Permission is granted to anyone to use this software for any 
+purpose, including commercial applications, and to alter it and 
+redistribute it freely, subject to the following restrictions:
+
+1. The origin of this software must not be misrepresented; you 
+   must not claim that you wrote the original software. If you use 
+   this software in a product, an acknowledgment in the product 
+   documentation would be appreciated but is not required.
+
+2. Altered source versions must be plainly marked as such, and 
+   must not be misrepresented as being the original software.
+
+3. This notice may not be removed or altered from any source 
+   distribution.