You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kafka.apache.org by ij...@apache.org on 2017/09/28 20:59:24 UTC
[3/3] kafka git commit: KAFKA-5746;
Add new metrics to support health checks (KIP-188)
KAFKA-5746; Add new metrics to support health checks (KIP-188)
Adds new metrics to support health checks:
1. Error rates for each request type, per-error code
2. Request size and temporary memory size
3. Message conversion rate and time
4. Successful and failed authentication rates
5. ZooKeeper latency and status
6. Client version
Author: Rajini Sivaram <ra...@googlemail.com>
Reviewers: Ismael Juma <is...@juma.me.uk>
Closes #3705 from rajinisivaram/KAFKA-5746-new-metrics
Project: http://git-wip-us.apache.org/repos/asf/kafka/repo
Commit: http://git-wip-us.apache.org/repos/asf/kafka/commit/021d8a8e
Tree: http://git-wip-us.apache.org/repos/asf/kafka/tree/021d8a8e
Diff: http://git-wip-us.apache.org/repos/asf/kafka/diff/021d8a8e
Branch: refs/heads/trunk
Commit: 021d8a8e9698dce454e0e801092460b98f0a8a4d
Parents: dd6347a
Author: Rajini Sivaram <ra...@googlemail.com>
Authored: Thu Sep 28 21:58:47 2017 +0100
Committer: Ismael Juma <is...@juma.me.uk>
Committed: Thu Sep 28 21:58:59 2017 +0100
----------------------------------------------------------------------
checkstyle/import-control.xml | 4 +
checkstyle/suppressions.xml | 2 +-
.../kafka/clients/admin/KafkaAdminClient.java | 4 +-
.../kafka/clients/consumer/KafkaConsumer.java | 4 +-
.../kafka/clients/producer/KafkaProducer.java | 4 +-
.../clients/producer/internals/Sender.java | 2 +-
.../java/org/apache/kafka/common/Metric.java | 12 +-
.../org/apache/kafka/common/metrics/Gauge.java | 31 ++
.../kafka/common/metrics/JmxReporter.java | 2 +-
.../kafka/common/metrics/KafkaMetric.java | 40 ++-
.../apache/kafka/common/metrics/Measurable.java | 4 +-
.../common/metrics/MetricValueProvider.java | 28 ++
.../apache/kafka/common/metrics/Metrics.java | 36 ++-
.../org/apache/kafka/common/metrics/Sensor.java | 2 +-
.../apache/kafka/common/network/Selector.java | 33 ++-
.../kafka/common/record/AbstractRecords.java | 46 ++-
.../kafka/common/record/ConvertedRecords.java | 36 +++
.../apache/kafka/common/record/FileRecords.java | 17 +-
.../kafka/common/record/MemoryRecords.java | 5 +-
.../common/record/MemoryRecordsBuilder.java | 40 +--
.../org/apache/kafka/common/record/Records.java | 9 +-
.../common/record/RecordsProcessingStats.java | 60 ++++
.../kafka/common/requests/AbstractRequest.java | 14 +
.../kafka/common/requests/AbstractResponse.java | 29 ++
.../requests/AddOffsetsToTxnResponse.java | 6 +
.../requests/AddPartitionsToTxnResponse.java | 5 +
.../common/requests/AlterConfigsResponse.java | 6 +
.../requests/AlterReplicaDirResponse.java | 5 +
.../common/requests/ApiVersionsResponse.java | 5 +
.../requests/ControlledShutdownResponse.java | 6 +
.../common/requests/CreateAclsResponse.java | 11 +
.../requests/CreatePartitionsRequest.java | 2 +-
.../requests/CreatePartitionsResponse.java | 6 +
.../common/requests/CreateTopicsResponse.java | 6 +
.../common/requests/DeleteAclsResponse.java | 11 +
.../common/requests/DeleteRecordsResponse.java | 8 +
.../common/requests/DeleteTopicsResponse.java | 5 +
.../common/requests/DescribeAclsResponse.java | 6 +
.../requests/DescribeConfigsResponse.java | 9 +
.../common/requests/DescribeGroupsResponse.java | 7 +
.../requests/DescribeLogDirsResponse.java | 8 +
.../kafka/common/requests/EndTxnResponse.java | 6 +
.../kafka/common/requests/FetchResponse.java | 9 +
.../requests/FindCoordinatorResponse.java | 6 +
.../common/requests/HeartbeatResponse.java | 6 +
.../common/requests/InitProducerIdResponse.java | 6 +
.../common/requests/JoinGroupResponse.java | 5 +
.../common/requests/LeaderAndIsrResponse.java | 5 +
.../common/requests/LeaveGroupResponse.java | 6 +
.../common/requests/ListGroupsResponse.java | 6 +
.../common/requests/ListOffsetResponse.java | 8 +
.../kafka/common/requests/MetadataResponse.java | 8 +
.../common/requests/OffsetCommitResponse.java | 5 +
.../common/requests/OffsetFetchResponse.java | 5 +
.../requests/OffsetsForLeaderEpochResponse.java | 8 +
.../kafka/common/requests/ProduceRequest.java | 7 +
.../kafka/common/requests/ProduceResponse.java | 8 +
.../requests/SaslAuthenticateResponse.java | 6 +
.../common/requests/SaslHandshakeResponse.java | 6 +
.../common/requests/StopReplicaResponse.java | 5 +
.../common/requests/SyncGroupResponse.java | 6 +
.../requests/TxnOffsetCommitResponse.java | 5 +
.../common/requests/UpdateMetadataResponse.java | 6 +
.../requests/WriteTxnMarkersResponse.java | 10 +
.../authenticator/SaslServerAuthenticator.java | 13 +-
.../kafka/common/utils/AppInfoParser.java | 43 ++-
.../clients/consumer/internals/FetcherTest.java | 1 +
.../clients/producer/internals/SenderTest.java | 3 +
.../kafka/common/metrics/MetricsTest.java | 1 +
.../common/metrics/stats/FrequenciesTest.java | 3 +-
.../kafka/common/network/NioEchoServer.java | 51 +++-
.../kafka/common/network/SelectorTest.java | 2 +-
.../common/network/SslTransportLayerTest.java | 7 +
.../kafka/common/record/FileRecordsTest.java | 12 +-
.../common/record/MemoryRecordsBuilderTest.java | 43 ++-
.../authenticator/SaslAuthenticatorTest.java | 11 +
.../SaslServerAuthenticatorTest.java | 5 +-
.../kafka/connect/runtime/ConnectMetrics.java | 4 +-
.../runtime/distributed/WorkerGroupMember.java | 4 +-
.../main/scala/kafka/admin/ConfigCommand.scala | 2 +-
.../kafka/controller/KafkaController.scala | 4 +-
core/src/main/scala/kafka/log/Log.scala | 12 +-
.../src/main/scala/kafka/log/LogValidator.scala | 52 +++-
.../scala/kafka/metrics/KafkaMetricsGroup.scala | 15 +-
.../scala/kafka/network/RequestChannel.scala | 97 ++++++-
.../main/scala/kafka/server/AdminManager.scala | 2 +-
.../scala/kafka/server/ClientQuotaManager.scala | 2 -
.../src/main/scala/kafka/server/KafkaApis.scala | 49 +++-
.../scala/kafka/server/KafkaHealthcheck.scala | 21 +-
.../kafka/server/KafkaRequestHandler.scala | 6 +
.../main/scala/kafka/server/KafkaServer.scala | 17 +-
.../scala/kafka/server/ReplicaManager.scala | 5 +-
core/src/main/scala/kafka/utils/ZkUtils.scala | 122 +++++---
.../integration/kafka/api/MetricsTest.scala | 288 +++++++++++++++++++
.../kafka/api/PlaintextConsumerTest.scala | 49 ----
.../group/GroupCoordinatorTest.scala | 9 +-
.../group/GroupMetadataManagerTest.scala | 6 +-
.../TransactionStateManagerTest.scala | 4 +-
.../scala/unit/kafka/log/LogValidatorTest.scala | 71 +++++
.../unit/kafka/network/SocketServerTest.scala | 4 +-
.../scala/unit/kafka/server/KafkaApisTest.scala | 3 +
docs/upgrade.html | 6 +
.../java/org/apache/kafka/tools/ToolsUtils.java | 14 +-
103 files changed, 1544 insertions(+), 243 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/kafka/blob/021d8a8e/checkstyle/import-control.xml
----------------------------------------------------------------------
diff --git a/checkstyle/import-control.xml b/checkstyle/import-control.xml
index f4d9655..ddb13bc 100644
--- a/checkstyle/import-control.xml
+++ b/checkstyle/import-control.xml
@@ -140,6 +140,10 @@
<allow class="org.apache.kafka.common.errors.SerializationException" />
<allow class="org.apache.kafka.common.header.Headers" />
</subpackage>
+
+ <subpackage name="utils">
+ <allow pkg="org.apache.kafka.common.metrics" />
+ </subpackage>
</subpackage>
<subpackage name="clients">
http://git-wip-us.apache.org/repos/asf/kafka/blob/021d8a8e/checkstyle/suppressions.xml
----------------------------------------------------------------------
diff --git a/checkstyle/suppressions.xml b/checkstyle/suppressions.xml
index 734b07e..b747e4c 100644
--- a/checkstyle/suppressions.xml
+++ b/checkstyle/suppressions.xml
@@ -60,7 +60,7 @@
files="AbstractRequest.java"/>
<suppress checks="NPathComplexity"
- files="(BufferPool|MetricName|Node|ConfigDef|SslTransportLayer|MetadataResponse|KerberosLogin|SslTransportLayer|Sender|Serdes|Agent).java"/>
+ files="(BufferPool|MetricName|Node|ConfigDef|SslTransportLayer|MetadataResponse|KerberosLogin|Selector|Sender|Serdes|Agent).java"/>
<!-- clients tests -->
<suppress checks="ClassDataAbstractionCoupling"
http://git-wip-us.apache.org/repos/asf/kafka/blob/021d8a8e/clients/src/main/java/org/apache/kafka/clients/admin/KafkaAdminClient.java
----------------------------------------------------------------------
diff --git a/clients/src/main/java/org/apache/kafka/clients/admin/KafkaAdminClient.java b/clients/src/main/java/org/apache/kafka/clients/admin/KafkaAdminClient.java
index 88339bf..5f37b8e 100644
--- a/clients/src/main/java/org/apache/kafka/clients/admin/KafkaAdminClient.java
+++ b/clients/src/main/java/org/apache/kafka/clients/admin/KafkaAdminClient.java
@@ -377,7 +377,7 @@ public class KafkaAdminClient extends AdminClient {
new TimeoutProcessorFactory() : timeoutProcessorFactory;
this.maxRetries = config.getInt(AdminClientConfig.RETRIES_CONFIG);
config.logUnused();
- AppInfoParser.registerAppInfo(JMX_PREFIX, sanitizedClientId);
+ AppInfoParser.registerAppInfo(JMX_PREFIX, sanitizedClientId, metrics);
log.debug("Kafka admin client initialized");
thread.start();
}
@@ -418,7 +418,7 @@ public class KafkaAdminClient extends AdminClient {
// Wait for the thread to be joined.
thread.join();
- AppInfoParser.unregisterAppInfo(JMX_PREFIX, Sanitizer.sanitize(clientId));
+ AppInfoParser.unregisterAppInfo(JMX_PREFIX, Sanitizer.sanitize(clientId), metrics);
log.debug("Kafka admin client closed.");
} catch (InterruptedException e) {
http://git-wip-us.apache.org/repos/asf/kafka/blob/021d8a8e/clients/src/main/java/org/apache/kafka/clients/consumer/KafkaConsumer.java
----------------------------------------------------------------------
diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/KafkaConsumer.java b/clients/src/main/java/org/apache/kafka/clients/consumer/KafkaConsumer.java
index f4af39c..d6764ca 100644
--- a/clients/src/main/java/org/apache/kafka/clients/consumer/KafkaConsumer.java
+++ b/clients/src/main/java/org/apache/kafka/clients/consumer/KafkaConsumer.java
@@ -771,7 +771,7 @@ public class KafkaConsumer<K, V> implements Consumer<K, V> {
isolationLevel);
config.logUnused();
- AppInfoParser.registerAppInfo(JMX_PREFIX, sanitizedClientId);
+ AppInfoParser.registerAppInfo(JMX_PREFIX, sanitizedClientId, metrics);
log.debug("Kafka consumer initialized");
} catch (Throwable t) {
@@ -1726,7 +1726,7 @@ public class KafkaConsumer<K, V> implements Consumer<K, V> {
ClientUtils.closeQuietly(client, "consumer network client", firstException);
ClientUtils.closeQuietly(keyDeserializer, "consumer key deserializer", firstException);
ClientUtils.closeQuietly(valueDeserializer, "consumer value deserializer", firstException);
- AppInfoParser.unregisterAppInfo(JMX_PREFIX, Sanitizer.sanitize(clientId));
+ AppInfoParser.unregisterAppInfo(JMX_PREFIX, Sanitizer.sanitize(clientId), metrics);
log.debug("Kafka consumer has been closed");
Throwable exception = firstException.get();
if (exception != null && !swallowException) {
http://git-wip-us.apache.org/repos/asf/kafka/blob/021d8a8e/clients/src/main/java/org/apache/kafka/clients/producer/KafkaProducer.java
----------------------------------------------------------------------
diff --git a/clients/src/main/java/org/apache/kafka/clients/producer/KafkaProducer.java b/clients/src/main/java/org/apache/kafka/clients/producer/KafkaProducer.java
index d40a16c..a202217 100644
--- a/clients/src/main/java/org/apache/kafka/clients/producer/KafkaProducer.java
+++ b/clients/src/main/java/org/apache/kafka/clients/producer/KafkaProducer.java
@@ -427,7 +427,7 @@ public class KafkaProducer<K, V> implements Producer<K, V> {
this.ioThread.start();
this.errors = this.metrics.sensor("errors");
config.logUnused();
- AppInfoParser.registerAppInfo(JMX_PREFIX, sanitizedClientId);
+ AppInfoParser.registerAppInfo(JMX_PREFIX, sanitizedClientId, metrics);
log.debug("Kafka producer started");
} catch (Throwable t) {
// call close methods if internal objects are already constructed this is to prevent resource leak. see KAFKA-2121
@@ -1075,7 +1075,7 @@ public class KafkaProducer<K, V> implements Producer<K, V> {
ClientUtils.closeQuietly(keySerializer, "producer keySerializer", firstException);
ClientUtils.closeQuietly(valueSerializer, "producer valueSerializer", firstException);
ClientUtils.closeQuietly(partitioner, "producer partitioner", firstException);
- AppInfoParser.unregisterAppInfo(JMX_PREFIX, Sanitizer.sanitize(clientId));
+ AppInfoParser.unregisterAppInfo(JMX_PREFIX, Sanitizer.sanitize(clientId), metrics);
log.debug("Kafka producer has been closed");
if (firstException.get() != null && !swallowException)
throw new KafkaException("Failed to close kafka producer", firstException.get());
http://git-wip-us.apache.org/repos/asf/kafka/blob/021d8a8e/clients/src/main/java/org/apache/kafka/clients/producer/internals/Sender.java
----------------------------------------------------------------------
diff --git a/clients/src/main/java/org/apache/kafka/clients/producer/internals/Sender.java b/clients/src/main/java/org/apache/kafka/clients/producer/internals/Sender.java
index 6d9021c..a15a250 100644
--- a/clients/src/main/java/org/apache/kafka/clients/producer/internals/Sender.java
+++ b/clients/src/main/java/org/apache/kafka/clients/producer/internals/Sender.java
@@ -675,7 +675,7 @@ public class Sender implements Runnable {
// not all support the same message format version. For example, if a partition migrates from a broker
// which is supporting the new magic version to one which doesn't, then we will need to convert.
if (!records.hasMatchingMagic(minUsedMagic))
- records = batch.records().downConvert(minUsedMagic, 0);
+ records = batch.records().downConvert(minUsedMagic, 0, time).records();
produceRecordsByPartition.put(tp, records);
recordsByPartition.put(tp, batch);
}
http://git-wip-us.apache.org/repos/asf/kafka/blob/021d8a8e/clients/src/main/java/org/apache/kafka/common/Metric.java
----------------------------------------------------------------------
diff --git a/clients/src/main/java/org/apache/kafka/common/Metric.java b/clients/src/main/java/org/apache/kafka/common/Metric.java
index 908c5c7..1f8d618 100644
--- a/clients/src/main/java/org/apache/kafka/common/Metric.java
+++ b/clients/src/main/java/org/apache/kafka/common/Metric.java
@@ -17,7 +17,7 @@
package org.apache.kafka.common;
/**
- * A numerical metric tracked for monitoring purposes
+ * A metric tracked for monitoring purposes.
*/
public interface Metric {
@@ -27,8 +27,16 @@ public interface Metric {
public MetricName metricName();
/**
- * The value of the metric
+ * The value of the metric as double if the metric is measurable
+ * @throws IllegalStateException if this metric does not have a measurable double value
+ * @deprecated As of 1.0.0, use {@link #metricValue()} instead. This will be removed in a future major release.
*/
+ @Deprecated
public double value();
+ /**
+ * The value of the metric, which may be measurable or a non-measurable gauge
+ */
+ public Object metricValue();
+
}
http://git-wip-us.apache.org/repos/asf/kafka/blob/021d8a8e/clients/src/main/java/org/apache/kafka/common/metrics/Gauge.java
----------------------------------------------------------------------
diff --git a/clients/src/main/java/org/apache/kafka/common/metrics/Gauge.java b/clients/src/main/java/org/apache/kafka/common/metrics/Gauge.java
new file mode 100644
index 0000000..647942b
--- /dev/null
+++ b/clients/src/main/java/org/apache/kafka/common/metrics/Gauge.java
@@ -0,0 +1,31 @@
+/*
+ * 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.metrics;
+
+/**
+ * A gauge metric is an instantaneous reading of a particular value.
+ */
+public interface Gauge<T> extends MetricValueProvider<T> {
+
+ /**
+ * Returns the current value associated with this gauge.
+ * @param config The configuration for this metric
+ * @param now The POSIX time in milliseconds the measurement is being taken
+ */
+ T value(MetricConfig config, long now);
+
+}
http://git-wip-us.apache.org/repos/asf/kafka/blob/021d8a8e/clients/src/main/java/org/apache/kafka/common/metrics/JmxReporter.java
----------------------------------------------------------------------
diff --git a/clients/src/main/java/org/apache/kafka/common/metrics/JmxReporter.java b/clients/src/main/java/org/apache/kafka/common/metrics/JmxReporter.java
index de02a23..294e1d8 100644
--- a/clients/src/main/java/org/apache/kafka/common/metrics/JmxReporter.java
+++ b/clients/src/main/java/org/apache/kafka/common/metrics/JmxReporter.java
@@ -184,7 +184,7 @@ public class JmxReporter implements MetricsReporter {
@Override
public Object getAttribute(String name) throws AttributeNotFoundException, MBeanException, ReflectionException {
if (this.metrics.containsKey(name))
- return this.metrics.get(name).value();
+ return this.metrics.get(name).metricValue();
else
throw new AttributeNotFoundException("Could not find attribute " + name);
}
http://git-wip-us.apache.org/repos/asf/kafka/blob/021d8a8e/clients/src/main/java/org/apache/kafka/common/metrics/KafkaMetric.java
----------------------------------------------------------------------
diff --git a/clients/src/main/java/org/apache/kafka/common/metrics/KafkaMetric.java b/clients/src/main/java/org/apache/kafka/common/metrics/KafkaMetric.java
index ef53b89..0cdd844 100644
--- a/clients/src/main/java/org/apache/kafka/common/metrics/KafkaMetric.java
+++ b/clients/src/main/java/org/apache/kafka/common/metrics/KafkaMetric.java
@@ -25,13 +25,16 @@ public final class KafkaMetric implements Metric {
private MetricName metricName;
private final Object lock;
private final Time time;
- private final Measurable measurable;
+ private final MetricValueProvider<?> metricValueProvider;
private MetricConfig config;
- KafkaMetric(Object lock, MetricName metricName, Measurable measurable, MetricConfig config, Time time) {
+ KafkaMetric(Object lock, MetricName metricName, MetricValueProvider<?> valueProvider,
+ MetricConfig config, Time time) {
this.metricName = metricName;
this.lock = lock;
- this.measurable = measurable;
+ if (!(valueProvider instanceof Measurable) && !(valueProvider instanceof Gauge))
+ throw new IllegalArgumentException("Unsupported metric value provider of class " + valueProvider.getClass());
+ this.metricValueProvider = valueProvider;
this.config = config;
this.time = time;
}
@@ -45,19 +48,42 @@ public final class KafkaMetric implements Metric {
return this.metricName;
}
+ /**
+ * See {@link Metric#value()} for the details on why this is deprecated.
+ */
@Override
+ @Deprecated
public double value() {
synchronized (this.lock) {
- return value(time.milliseconds());
+ return measurableValue(time.milliseconds());
+ }
+ }
+
+ @Override
+ public Object metricValue() {
+ long now = time.milliseconds();
+ synchronized (this.lock) {
+ if (this.metricValueProvider instanceof Measurable)
+ return ((Measurable) metricValueProvider).measure(config, now);
+ else if (this.metricValueProvider instanceof Gauge)
+ return ((Gauge<?>) metricValueProvider).value(config, now);
+ else
+ throw new IllegalStateException("Not a valid metric: " + this.metricValueProvider.getClass());
}
}
public Measurable measurable() {
- return this.measurable;
+ if (this.metricValueProvider instanceof Measurable)
+ return (Measurable) metricValueProvider;
+ else
+ throw new IllegalStateException("Not a measurable: " + this.metricValueProvider.getClass());
}
- double value(long timeMs) {
- return this.measurable.measure(config, timeMs);
+ double measurableValue(long timeMs) {
+ if (this.metricValueProvider instanceof Measurable)
+ return ((Measurable) metricValueProvider).measure(config, timeMs);
+ else
+ throw new IllegalStateException("Not a measurable metric");
}
public void config(MetricConfig config) {
http://git-wip-us.apache.org/repos/asf/kafka/blob/021d8a8e/clients/src/main/java/org/apache/kafka/common/metrics/Measurable.java
----------------------------------------------------------------------
diff --git a/clients/src/main/java/org/apache/kafka/common/metrics/Measurable.java b/clients/src/main/java/org/apache/kafka/common/metrics/Measurable.java
index d068895..866caba 100644
--- a/clients/src/main/java/org/apache/kafka/common/metrics/Measurable.java
+++ b/clients/src/main/java/org/apache/kafka/common/metrics/Measurable.java
@@ -19,7 +19,7 @@ package org.apache.kafka.common.metrics;
/**
* A measurable quantity that can be registered as a metric
*/
-public interface Measurable {
+public interface Measurable extends MetricValueProvider<Double> {
/**
* Measure this quantity and return the result as a double
@@ -27,6 +27,6 @@ public interface Measurable {
* @param now The POSIX time in milliseconds the measurement is being taken
* @return The measured value
*/
- public double measure(MetricConfig config, long now);
+ double measure(MetricConfig config, long now);
}
http://git-wip-us.apache.org/repos/asf/kafka/blob/021d8a8e/clients/src/main/java/org/apache/kafka/common/metrics/MetricValueProvider.java
----------------------------------------------------------------------
diff --git a/clients/src/main/java/org/apache/kafka/common/metrics/MetricValueProvider.java b/clients/src/main/java/org/apache/kafka/common/metrics/MetricValueProvider.java
new file mode 100644
index 0000000..68028e7
--- /dev/null
+++ b/clients/src/main/java/org/apache/kafka/common/metrics/MetricValueProvider.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.metrics;
+
+/**
+ * Super-interface for {@link Measurable} or {@link Gauge} that provides
+ * metric values.
+ * <p>
+ * In the future for Java8 and above, {@link Gauge#value(MetricConfig, long)} will be
+ * moved to this interface with a default implementation in {@link Measurable} that returns
+ * {@link Measurable#measure(MetricConfig, long)}.
+ * </p>
+ */
+public interface MetricValueProvider<T> { }
http://git-wip-us.apache.org/repos/asf/kafka/blob/021d8a8e/clients/src/main/java/org/apache/kafka/common/metrics/Metrics.java
----------------------------------------------------------------------
diff --git a/clients/src/main/java/org/apache/kafka/common/metrics/Metrics.java b/clients/src/main/java/org/apache/kafka/common/metrics/Metrics.java
index 0b4507b..8127022 100644
--- a/clients/src/main/java/org/apache/kafka/common/metrics/Metrics.java
+++ b/clients/src/main/java/org/apache/kafka/common/metrics/Metrics.java
@@ -449,6 +449,10 @@ public class Metrics implements Closeable {
/**
* Add a metric to monitor an object that implements measurable. This metric won't be associated with any sensor.
* This is a way to expose existing values as metrics.
+ *
+ * This method is kept for binary compatibility purposes, it has the same behaviour as
+ * {@link #addMetric(MetricName, MetricValue)}.
+ *
* @param metricName The name of the metric
* @param measurable The measurable that will be measured by this metric
*/
@@ -457,22 +461,48 @@ public class Metrics implements Closeable {
}
/**
- * Add a metric to monitor an object that implements measurable. This metric won't be associated with any sensor.
+ * Add a metric to monitor an object that implements Measurable. This metric won't be associated with any sensor.
* This is a way to expose existing values as metrics.
+ *
+ * This method is kept for binary compatibility purposes, it has the same behaviour as
+ * {@link #addMetric(MetricName, MetricConfig, MetricValueProvider)}.
+ *
* @param metricName The name of the metric
* @param config The configuration to use when measuring this measurable
* @param measurable The measurable that will be measured by this metric
*/
- public synchronized void addMetric(MetricName metricName, MetricConfig config, Measurable measurable) {
+ public void addMetric(MetricName metricName, MetricConfig config, Measurable measurable) {
+ addMetric(metricName, config, (MetricValueProvider<?>) measurable);
+ }
+
+ /**
+ * Add a metric to monitor an object that implements MetricValueProvider. This metric won't be associated with any
+ * sensor. This is a way to expose existing values as metrics.
+ *
+ * @param metricName The name of the metric
+ * @param metricValueProvider The metric value provider associated with this metric
+ */
+ public void addMetric(MetricName metricName, MetricConfig config, MetricValueProvider<?> metricValueProvider) {
KafkaMetric m = new KafkaMetric(new Object(),
Utils.notNull(metricName),
- Utils.notNull(measurable),
+ Utils.notNull(metricValueProvider),
config == null ? this.config : config,
time);
registerMetric(m);
}
/**
+ * Add a metric to monitor an object that implements MetricValueProvider. This metric won't be associated with any
+ * sensor. This is a way to expose existing values as metrics.
+ *
+ * @param metricName The name of the metric
+ * @param metricValueProvider The metric value provider associated with this metric
+ */
+ public void addMetric(MetricName metricName, MetricValueProvider<?> metricValueProvider) {
+ addMetric(metricName, null, metricValueProvider);
+ }
+
+ /**
* Remove a metric if it exists and return it. Return null otherwise. If a metric is removed, `metricRemoval`
* will be invoked for each reporter.
*
http://git-wip-us.apache.org/repos/asf/kafka/blob/021d8a8e/clients/src/main/java/org/apache/kafka/common/metrics/Sensor.java
----------------------------------------------------------------------
diff --git a/clients/src/main/java/org/apache/kafka/common/metrics/Sensor.java b/clients/src/main/java/org/apache/kafka/common/metrics/Sensor.java
index 55d75b1..321fab6 100644
--- a/clients/src/main/java/org/apache/kafka/common/metrics/Sensor.java
+++ b/clients/src/main/java/org/apache/kafka/common/metrics/Sensor.java
@@ -199,7 +199,7 @@ public final class Sensor {
if (config != null) {
Quota quota = config.quota();
if (quota != null) {
- double value = metric.value(timeMs);
+ double value = metric.measurableValue(timeMs);
if (!quota.acceptable(value)) {
throw new QuotaViolationException(metric.metricName(), value,
quota.bound());
http://git-wip-us.apache.org/repos/asf/kafka/blob/021d8a8e/clients/src/main/java/org/apache/kafka/common/network/Selector.java
----------------------------------------------------------------------
diff --git a/clients/src/main/java/org/apache/kafka/common/network/Selector.java b/clients/src/main/java/org/apache/kafka/common/network/Selector.java
index b753745..f70afe0 100644
--- a/clients/src/main/java/org/apache/kafka/common/network/Selector.java
+++ b/clients/src/main/java/org/apache/kafka/common/network/Selector.java
@@ -456,7 +456,14 @@ public class Selector implements Selectable, AutoCloseable {
/* if channel is not ready finish prepare */
if (channel.isConnected() && !channel.ready()) {
- channel.prepare();
+ try {
+ channel.prepare();
+ } catch (AuthenticationException e) {
+ sensors.failedAuthentication.record();
+ throw e;
+ }
+ if (channel.ready())
+ sensors.successfulAuthentication.record();
}
attemptRead(key, channel);
@@ -839,6 +846,8 @@ public class Selector implements Selectable, AutoCloseable {
public final Sensor connectionClosed;
public final Sensor connectionCreated;
+ public final Sensor successfulAuthentication;
+ public final Sensor failedAuthentication;
public final Sensor bytesTransferred;
public final Sensor bytesSent;
public final Sensor bytesReceived;
@@ -863,19 +872,27 @@ public class Selector implements Selectable, AutoCloseable {
tagsSuffix.append(tag.getValue());
}
- this.connectionClosed = sensor("connections-closed:" + tagsSuffix.toString());
+ this.connectionClosed = sensor("connections-closed:" + tagsSuffix);
this.connectionClosed.add(createMeter(metrics, metricGrpName, metricTags,
"connection-close", "connections closed"));
- this.connectionCreated = sensor("connections-created:" + tagsSuffix.toString());
+ this.connectionCreated = sensor("connections-created:" + tagsSuffix);
this.connectionCreated.add(createMeter(metrics, metricGrpName, metricTags,
"connection-creation", "new connections established"));
- this.bytesTransferred = sensor("bytes-sent-received:" + tagsSuffix.toString());
+ this.successfulAuthentication = sensor("successful-authentication:" + tagsSuffix);
+ this.successfulAuthentication.add(createMeter(metrics, metricGrpName, metricTags,
+ "successful-authentication", "connections with successful authentication"));
+
+ this.failedAuthentication = sensor("failed-authentication:" + tagsSuffix);
+ this.failedAuthentication.add(createMeter(metrics, metricGrpName, metricTags,
+ "failed-authentication", "connections with failed authentication"));
+
+ this.bytesTransferred = sensor("bytes-sent-received:" + tagsSuffix);
bytesTransferred.add(createMeter(metrics, metricGrpName, metricTags, new Count(),
"network-io", "network operations (reads or writes) on all connections"));
- this.bytesSent = sensor("bytes-sent:" + tagsSuffix.toString(), bytesTransferred);
+ this.bytesSent = sensor("bytes-sent:" + tagsSuffix, bytesTransferred);
this.bytesSent.add(createMeter(metrics, metricGrpName, metricTags,
"outgoing-byte", "outgoing bytes sent to all servers"));
this.bytesSent.add(createMeter(metrics, metricGrpName, metricTags, new Count(),
@@ -885,20 +902,20 @@ public class Selector implements Selectable, AutoCloseable {
metricName = metrics.metricName("request-size-max", metricGrpName, "The maximum size of any request sent.", metricTags);
this.bytesSent.add(metricName, new Max());
- this.bytesReceived = sensor("bytes-received:" + tagsSuffix.toString(), bytesTransferred);
+ this.bytesReceived = sensor("bytes-received:" + tagsSuffix, bytesTransferred);
this.bytesReceived.add(createMeter(metrics, metricGrpName, metricTags,
"incoming-byte", "bytes read off all sockets"));
this.bytesReceived.add(createMeter(metrics, metricGrpName, metricTags,
new Count(), "response", "responses received"));
- this.selectTime = sensor("select-time:" + tagsSuffix.toString());
+ this.selectTime = sensor("select-time:" + tagsSuffix);
this.selectTime.add(createMeter(metrics, metricGrpName, metricTags,
new Count(), "select", "times the I/O layer checked for new I/O to perform"));
metricName = metrics.metricName("io-wait-time-ns-avg", metricGrpName, "The average length of time the I/O thread spent waiting for a socket ready for reads or writes in nanoseconds.", metricTags);
this.selectTime.add(metricName, new Avg());
this.selectTime.add(createIOThreadRatioMeter(metrics, metricGrpName, metricTags, "io-wait", "waiting"));
- this.ioTime = sensor("io-time:" + tagsSuffix.toString());
+ this.ioTime = sensor("io-time:" + tagsSuffix);
metricName = metrics.metricName("io-time-ns-avg", metricGrpName, "The average length of time for I/O per select call in nanoseconds.", metricTags);
this.ioTime.add(metricName, new Avg());
this.ioTime.add(createIOThreadRatioMeter(metrics, metricGrpName, metricTags, "io", "doing I/O"));
http://git-wip-us.apache.org/repos/asf/kafka/blob/021d8a8e/clients/src/main/java/org/apache/kafka/common/record/AbstractRecords.java
----------------------------------------------------------------------
diff --git a/clients/src/main/java/org/apache/kafka/common/record/AbstractRecords.java b/clients/src/main/java/org/apache/kafka/common/record/AbstractRecords.java
index 9ba2048..6a1b95c 100644
--- a/clients/src/main/java/org/apache/kafka/common/record/AbstractRecords.java
+++ b/clients/src/main/java/org/apache/kafka/common/record/AbstractRecords.java
@@ -18,6 +18,7 @@ package org.apache.kafka.common.record;
import org.apache.kafka.common.header.Header;
import org.apache.kafka.common.utils.AbstractIterator;
+import org.apache.kafka.common.utils.Time;
import org.apache.kafka.common.utils.Utils;
import java.nio.ByteBuffer;
@@ -59,10 +60,12 @@ public abstract class AbstractRecords implements Records {
* need to drop records from the batch during the conversion. Some versions of librdkafka rely on this for
* correctness.
*/
- protected MemoryRecords downConvert(Iterable<? extends RecordBatch> batches, byte toMagic, long firstOffset) {
+ protected ConvertedRecords<MemoryRecords> downConvert(Iterable<? extends RecordBatch> batches, byte toMagic,
+ long firstOffset, Time time) {
// maintain the batch along with the decompressed records to avoid the need to decompress again
List<RecordBatchAndRecords> recordBatchAndRecordsList = new ArrayList<>();
int totalSizeEstimate = 0;
+ long startNanos = time.nanoseconds();
for (RecordBatch batch : batches) {
if (toMagic < RecordBatch.MAGIC_VALUE_V2 && batch.isControlBatch())
@@ -91,22 +94,31 @@ public abstract class AbstractRecords implements Records {
}
ByteBuffer buffer = ByteBuffer.allocate(totalSizeEstimate);
+ long temporaryMemoryBytes = 0;
+ long conversionCount = 0;
for (RecordBatchAndRecords recordBatchAndRecords : recordBatchAndRecordsList) {
- if (recordBatchAndRecords.batch.magic() <= toMagic)
+ temporaryMemoryBytes += recordBatchAndRecords.batch.sizeInBytes();
+ if (recordBatchAndRecords.batch.magic() <= toMagic) {
recordBatchAndRecords.batch.writeTo(buffer);
- else
- buffer = convertRecordBatch(toMagic, buffer, recordBatchAndRecords);
+ } else {
+ MemoryRecordsBuilder builder = convertRecordBatch(toMagic, buffer, recordBatchAndRecords);
+ buffer = builder.buffer();
+ temporaryMemoryBytes += builder.uncompressedBytesWritten();
+ conversionCount++;
+ }
}
buffer.flip();
- return MemoryRecords.readableRecords(buffer);
+ RecordsProcessingStats stats = new RecordsProcessingStats(temporaryMemoryBytes, conversionCount,
+ time.nanoseconds() - startNanos);
+ return new ConvertedRecords<>(MemoryRecords.readableRecords(buffer), stats);
}
/**
* Return a buffer containing the converted record batches. The returned buffer may not be the same as the received
* one (e.g. it may require expansion).
*/
- private ByteBuffer convertRecordBatch(byte magic, ByteBuffer buffer, RecordBatchAndRecords recordBatchAndRecords) {
+ private MemoryRecordsBuilder convertRecordBatch(byte magic, ByteBuffer buffer, RecordBatchAndRecords recordBatchAndRecords) {
RecordBatch batch = recordBatchAndRecords.batch;
final TimestampType timestampType = batch.timestampType();
long logAppendTime = timestampType == TimestampType.LOG_APPEND_TIME ? batch.maxTimestamp() : RecordBatch.NO_TIMESTAMP;
@@ -117,7 +129,7 @@ public abstract class AbstractRecords implements Records {
builder.append(record);
builder.close();
- return builder.buffer();
+ return builder;
}
/**
@@ -192,7 +204,8 @@ public abstract class AbstractRecords implements Records {
* Get an upper bound estimate on the batch size needed to hold a record with the given fields. This is only
* an estimate because it does not take into account overhead from the compression algorithm.
*/
- public static int estimateSizeInBytesUpperBound(byte magic, CompressionType compressionType, ByteBuffer key, ByteBuffer value, Header[] headers) {
+ public static int estimateSizeInBytesUpperBound(byte magic, CompressionType compressionType, ByteBuffer key,
+ ByteBuffer value, Header[] headers) {
if (magic >= RecordBatch.MAGIC_VALUE_V2)
return DefaultRecordBatch.estimateBatchSizeUpperBound(key, value, headers);
else if (compressionType != CompressionType.NONE)
@@ -201,6 +214,23 @@ public abstract class AbstractRecords implements Records {
return Records.LOG_OVERHEAD + LegacyRecord.recordSize(magic, key, value);
}
+ /**
+ * Return the size of the record batch header.
+ *
+ * For V0 and V1 with no compression, it's unclear if Records.LOG_OVERHEAD or 0 should be chosen. There is no header
+ * per batch, but a sequence of batches is preceded by the offset and size. This method returns `0` as it's what
+ * `MemoryRecordsBuilder` requires.
+ */
+ public static int recordBatchHeaderSizeInBytes(byte magic, CompressionType compressionType) {
+ if (magic > RecordBatch.MAGIC_VALUE_V1) {
+ return DefaultRecordBatch.RECORD_BATCH_OVERHEAD;
+ } else if (compressionType != CompressionType.NONE) {
+ return Records.LOG_OVERHEAD + LegacyRecord.recordOverhead(magic);
+ } else {
+ return 0;
+ }
+ }
+
private static class RecordBatchAndRecords {
private final RecordBatch batch;
private final List<Record> records;
http://git-wip-us.apache.org/repos/asf/kafka/blob/021d8a8e/clients/src/main/java/org/apache/kafka/common/record/ConvertedRecords.java
----------------------------------------------------------------------
diff --git a/clients/src/main/java/org/apache/kafka/common/record/ConvertedRecords.java b/clients/src/main/java/org/apache/kafka/common/record/ConvertedRecords.java
new file mode 100644
index 0000000..fe37c48
--- /dev/null
+++ b/clients/src/main/java/org/apache/kafka/common/record/ConvertedRecords.java
@@ -0,0 +1,36 @@
+/*
+ * 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.record;
+
+public class ConvertedRecords<T extends Records> {
+
+ private final T records;
+ private final RecordsProcessingStats recordsProcessingStats;
+
+ public ConvertedRecords(T records, RecordsProcessingStats recordsProcessingStats) {
+ this.records = records;
+ this.recordsProcessingStats = recordsProcessingStats;
+ }
+
+ public T records() {
+ return records;
+ }
+
+ public RecordsProcessingStats recordsProcessingStats() {
+ return recordsProcessingStats;
+ }
+}
http://git-wip-us.apache.org/repos/asf/kafka/blob/021d8a8e/clients/src/main/java/org/apache/kafka/common/record/FileRecords.java
----------------------------------------------------------------------
diff --git a/clients/src/main/java/org/apache/kafka/common/record/FileRecords.java b/clients/src/main/java/org/apache/kafka/common/record/FileRecords.java
index a898634..ae99db3 100644
--- a/clients/src/main/java/org/apache/kafka/common/record/FileRecords.java
+++ b/clients/src/main/java/org/apache/kafka/common/record/FileRecords.java
@@ -19,6 +19,7 @@ package org.apache.kafka.common.record;
import org.apache.kafka.common.KafkaException;
import org.apache.kafka.common.network.TransportLayer;
import org.apache.kafka.common.record.FileLogInputStream.FileChannelRecordBatch;
+import org.apache.kafka.common.utils.Time;
import org.apache.kafka.common.utils.Utils;
import java.io.Closeable;
@@ -237,19 +238,19 @@ public class FileRecords extends AbstractRecords implements Closeable {
}
@Override
- public Records downConvert(byte toMagic, long firstOffset) {
+ public ConvertedRecords<? extends Records> downConvert(byte toMagic, long firstOffset, Time time) {
List<? extends RecordBatch> batches = Utils.toList(batches().iterator());
if (batches.isEmpty()) {
// This indicates that the message is too large, which means that the buffer is not large
- // enough to hold a full record batch. We just return all the bytes in the file message set.
- // Even though the message set does not have the right format version, we expect old clients
- // to raise an error to the user after reading the message size and seeing that there
- // are not enough available bytes in the response to read the full message. Note that this is
+ // enough to hold a full record batch. We just return all the bytes in this instance.
+ // Even though the record batch does not have the right format version, we expect old clients
+ // to raise an error to the user after reading the record batch size and seeing that there
+ // are not enough available bytes in the response to read it fully. Note that this is
// only possible prior to KIP-74, after which the broker was changed to always return at least
- // one full message, even if it requires exceeding the max fetch size requested by the client.
- return this;
+ // one full record batch, even if it requires exceeding the max fetch size requested by the client.
+ return new ConvertedRecords<>(this, RecordsProcessingStats.EMPTY);
} else {
- return downConvert(batches, toMagic, firstOffset);
+ return downConvert(batches, toMagic, firstOffset, time);
}
}
http://git-wip-us.apache.org/repos/asf/kafka/blob/021d8a8e/clients/src/main/java/org/apache/kafka/common/record/MemoryRecords.java
----------------------------------------------------------------------
diff --git a/clients/src/main/java/org/apache/kafka/common/record/MemoryRecords.java b/clients/src/main/java/org/apache/kafka/common/record/MemoryRecords.java
index d3f2444..2a25aad 100644
--- a/clients/src/main/java/org/apache/kafka/common/record/MemoryRecords.java
+++ b/clients/src/main/java/org/apache/kafka/common/record/MemoryRecords.java
@@ -20,6 +20,7 @@ import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.record.MemoryRecords.RecordFilter.BatchRetention;
import org.apache.kafka.common.utils.ByteBufferOutputStream;
import org.apache.kafka.common.utils.CloseableIterator;
+import org.apache.kafka.common.utils.Time;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -111,8 +112,8 @@ public class MemoryRecords extends AbstractRecords {
}
@Override
- public MemoryRecords downConvert(byte toMagic, long firstOffset) {
- return downConvert(batches(), toMagic, firstOffset);
+ public ConvertedRecords<MemoryRecords> downConvert(byte toMagic, long firstOffset, Time time) {
+ return downConvert(batches(), toMagic, firstOffset, time);
}
/**
http://git-wip-us.apache.org/repos/asf/kafka/blob/021d8a8e/clients/src/main/java/org/apache/kafka/common/record/MemoryRecordsBuilder.java
----------------------------------------------------------------------
diff --git a/clients/src/main/java/org/apache/kafka/common/record/MemoryRecordsBuilder.java b/clients/src/main/java/org/apache/kafka/common/record/MemoryRecordsBuilder.java
index fc83134..ad0bab7 100644
--- a/clients/src/main/java/org/apache/kafka/common/record/MemoryRecordsBuilder.java
+++ b/clients/src/main/java/org/apache/kafka/common/record/MemoryRecordsBuilder.java
@@ -54,6 +54,7 @@ public class MemoryRecordsBuilder {
private final boolean isControlBatch;
private final int partitionLeaderEpoch;
private final int writeLimit;
+ private final int batchHeaderSizeInBytes;
// Use a conservative estimate of the compression ratio. The producer overrides this using statistics
// from previous batches before appending any records.
@@ -64,8 +65,7 @@ public class MemoryRecordsBuilder {
private long producerId;
private short producerEpoch;
private int baseSequence;
- private int writtenUncompressed = 0; // Number of bytes (excluding the header) written before compression
- private int batchHeaderSize;
+ private int uncompressedRecordsSizeInBytes = 0; // Number of bytes (excluding the header) written before compression
private int numRecords = 0;
private float actualCompressionRatio = 1;
private long maxTimestamp = RecordBatch.NO_TIMESTAMP;
@@ -104,7 +104,7 @@ public class MemoryRecordsBuilder {
this.baseOffset = baseOffset;
this.logAppendTime = logAppendTime;
this.numRecords = 0;
- this.writtenUncompressed = 0;
+ this.uncompressedRecordsSizeInBytes = 0;
this.actualCompressionRatio = 1;
this.maxTimestamp = RecordBatch.NO_TIMESTAMP;
this.producerId = producerId;
@@ -115,18 +115,9 @@ public class MemoryRecordsBuilder {
this.partitionLeaderEpoch = partitionLeaderEpoch;
this.writeLimit = writeLimit;
this.initialPosition = bufferStream.position();
+ this.batchHeaderSizeInBytes = AbstractRecords.recordBatchHeaderSizeInBytes(magic, compressionType);
- if (magic > RecordBatch.MAGIC_VALUE_V1) {
- batchHeaderSize = DefaultRecordBatch.RECORDS_OFFSET;
- } else if (compressionType != CompressionType.NONE) {
- // for compressed records, leave space for the header and the shallow message metadata
- // and move the starting position to the value payload offset
- batchHeaderSize = Records.LOG_OVERHEAD + LegacyRecord.recordOverhead(magic);
- } else {
- batchHeaderSize = 0;
- }
-
- bufferStream.position(initialPosition + batchHeaderSize);
+ bufferStream.position(initialPosition + batchHeaderSizeInBytes);
this.bufferStream = bufferStream;
this.appendStream = new DataOutputStream(compressionType.wrapForOutput(this.bufferStream, magic));
}
@@ -238,6 +229,17 @@ public class MemoryRecordsBuilder {
}
}
+ public int numRecords() {
+ return numRecords;
+ }
+
+ /**
+ * Return the sum of the size of the batch header (always uncompressed) and the records (before compression).
+ */
+ public int uncompressedBytesWritten() {
+ return uncompressedRecordsSizeInBytes + batchHeaderSizeInBytes;
+ }
+
public void setProducerState(long producerId, short producerEpoch, int baseSequence, boolean isTransactional) {
if (isClosed()) {
// Sequence numbers are assigned when the batch is closed while the accumulator is being drained.
@@ -306,9 +308,9 @@ public class MemoryRecordsBuilder {
builtRecords = MemoryRecords.EMPTY;
} else {
if (magic > RecordBatch.MAGIC_VALUE_V1)
- this.actualCompressionRatio = (float) writeDefaultBatchHeader() / this.writtenUncompressed;
+ this.actualCompressionRatio = (float) writeDefaultBatchHeader() / this.uncompressedRecordsSizeInBytes;
else if (compressionType != CompressionType.NONE)
- this.actualCompressionRatio = (float) writeLegacyCompressedWrapperHeader() / this.writtenUncompressed;
+ this.actualCompressionRatio = (float) writeLegacyCompressedWrapperHeader() / this.uncompressedRecordsSizeInBytes;
ByteBuffer buffer = buffer().duplicate();
buffer.flip();
@@ -651,7 +653,7 @@ public class MemoryRecordsBuilder {
", last offset: " + offset);
numRecords += 1;
- writtenUncompressed += size;
+ uncompressedRecordsSizeInBytes += size;
lastOffset = offset;
if (magic > RecordBatch.MAGIC_VALUE_V0 && timestamp > maxTimestamp) {
@@ -678,10 +680,10 @@ public class MemoryRecordsBuilder {
*/
private int estimatedBytesWritten() {
if (compressionType == CompressionType.NONE) {
- return batchHeaderSize + writtenUncompressed;
+ return batchHeaderSizeInBytes + uncompressedRecordsSizeInBytes;
} else {
// estimate the written bytes to the underlying byte buffer based on uncompressed written bytes
- return batchHeaderSize + (int) (writtenUncompressed * estimatedCompressionRatio * COMPRESSION_RATE_ESTIMATION_FACTOR);
+ return batchHeaderSizeInBytes + (int) (uncompressedRecordsSizeInBytes * estimatedCompressionRatio * COMPRESSION_RATE_ESTIMATION_FACTOR);
}
}
http://git-wip-us.apache.org/repos/asf/kafka/blob/021d8a8e/clients/src/main/java/org/apache/kafka/common/record/Records.java
----------------------------------------------------------------------
diff --git a/clients/src/main/java/org/apache/kafka/common/record/Records.java b/clients/src/main/java/org/apache/kafka/common/record/Records.java
index ec2e717..19152ba 100644
--- a/clients/src/main/java/org/apache/kafka/common/record/Records.java
+++ b/clients/src/main/java/org/apache/kafka/common/record/Records.java
@@ -19,6 +19,8 @@ package org.apache.kafka.common.record;
import java.io.IOException;
import java.nio.channels.GatheringByteChannel;
+import org.apache.kafka.common.utils.Time;
+
/**
* Interface for accessing the records contained in a log. The log itself is represented as a sequence of record
* batches (see {@link RecordBatch}).
@@ -97,10 +99,11 @@ public interface Records {
* deep iteration since all of the deep records must also be converted to the desired format.
* @param toMagic The magic value to convert to
* @param firstOffset The starting offset for returned records. This only impacts some cases. See
- * {@link AbstractRecords#downConvert(Iterable, byte, long)} for an explanation.
- * @return A Records instance (which may or may not be the same instance)
+ * {@link AbstractRecords#downConvert(Iterable, byte, long, Time) for an explanation.
+ * @param time instance used for reporting stats
+ * @return A ConvertedRecords instance which may or may not contain the same instance in its records field.
*/
- Records downConvert(byte toMagic, long firstOffset);
+ ConvertedRecords<? extends Records> downConvert(byte toMagic, long firstOffset, Time time);
/**
* Get an iterator over the records in this log. Note that this generally requires decompression,
http://git-wip-us.apache.org/repos/asf/kafka/blob/021d8a8e/clients/src/main/java/org/apache/kafka/common/record/RecordsProcessingStats.java
----------------------------------------------------------------------
diff --git a/clients/src/main/java/org/apache/kafka/common/record/RecordsProcessingStats.java b/clients/src/main/java/org/apache/kafka/common/record/RecordsProcessingStats.java
new file mode 100644
index 0000000..cbb76b7
--- /dev/null
+++ b/clients/src/main/java/org/apache/kafka/common/record/RecordsProcessingStats.java
@@ -0,0 +1,60 @@
+/*
+ * 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.record;
+
+public class RecordsProcessingStats {
+
+ public static final RecordsProcessingStats EMPTY = new RecordsProcessingStats(0L, 0L, -1);
+
+ private final long temporaryMemoryBytes;
+ private final long conversionCount;
+ private final long conversionTimeNanos;
+
+ public RecordsProcessingStats(long temporaryMemoryBytes, long conversionCount, long conversionTimeNanos) {
+ this.temporaryMemoryBytes = temporaryMemoryBytes;
+ this.conversionCount = conversionCount;
+ this.conversionTimeNanos = conversionTimeNanos;
+ }
+
+ /**
+ * Returns the number of temporary memory bytes allocated to process the records.
+ * This size depends on whether the records need decompression and/or conversion:
+ * <ul>
+ * <li>Non compressed, no conversion: zero</li>
+ * <li>Non compressed, with conversion: size of the converted buffer</li>
+ * <li>Compressed, no conversion: size of the original buffer after decompression</li>
+ * <li>Compressed, with conversion: size of the original buffer after decompression + size of the converted buffer uncompressed</li>
+ * </ul>
+ */
+ public long temporaryMemoryBytes() {
+ return temporaryMemoryBytes;
+ }
+
+ public long conversionCount() {
+ return conversionCount;
+ }
+
+ public long conversionTimeNanos() {
+ return conversionTimeNanos;
+ }
+
+ @Override
+ public String toString() {
+ return String.format("RecordsProcessingStats(temporaryMemoryBytes=%d, conversionCount=%d, conversionTimeNanos=%d)",
+ temporaryMemoryBytes, conversionCount, conversionTimeNanos);
+ }
+}
http://git-wip-us.apache.org/repos/asf/kafka/blob/021d8a8e/clients/src/main/java/org/apache/kafka/common/requests/AbstractRequest.java
----------------------------------------------------------------------
diff --git a/clients/src/main/java/org/apache/kafka/common/requests/AbstractRequest.java b/clients/src/main/java/org/apache/kafka/common/requests/AbstractRequest.java
index da1d147..e6dd6da 100644
--- a/clients/src/main/java/org/apache/kafka/common/requests/AbstractRequest.java
+++ b/clients/src/main/java/org/apache/kafka/common/requests/AbstractRequest.java
@@ -19,9 +19,11 @@ package org.apache.kafka.common.requests;
import org.apache.kafka.common.network.NetworkSend;
import org.apache.kafka.common.network.Send;
import org.apache.kafka.common.protocol.ApiKeys;
+import org.apache.kafka.common.protocol.Errors;
import org.apache.kafka.common.protocol.types.Struct;
import java.nio.ByteBuffer;
+import java.util.Map;
public abstract class AbstractRequest extends AbstractRequestResponse {
@@ -120,6 +122,18 @@ public abstract class AbstractRequest extends AbstractRequestResponse {
public abstract AbstractResponse getErrorResponse(int throttleTimeMs, Throwable e);
/**
+ * Get the error counts corresponding to an error response. This is overridden for requests
+ * where response may be null (e.g produce with acks=0).
+ */
+ public Map<Errors, Integer> errorCounts(Throwable e) {
+ AbstractResponse response = getErrorResponse(0, e);
+ if (response == null)
+ throw new IllegalStateException("Error counts could not be obtained for request " + this);
+ else
+ return response.errorCounts();
+ }
+
+ /**
* Factory method for getting a request object based on ApiKey ID and a version
*/
public static AbstractRequest parseRequest(ApiKeys apiKey, short apiVersion, Struct struct) {
http://git-wip-us.apache.org/repos/asf/kafka/blob/021d8a8e/clients/src/main/java/org/apache/kafka/common/requests/AbstractResponse.java
----------------------------------------------------------------------
diff --git a/clients/src/main/java/org/apache/kafka/common/requests/AbstractResponse.java b/clients/src/main/java/org/apache/kafka/common/requests/AbstractResponse.java
index 4cff798..12fe3c8 100644
--- a/clients/src/main/java/org/apache/kafka/common/requests/AbstractResponse.java
+++ b/clients/src/main/java/org/apache/kafka/common/requests/AbstractResponse.java
@@ -19,9 +19,13 @@ package org.apache.kafka.common.requests;
import org.apache.kafka.common.network.NetworkSend;
import org.apache.kafka.common.network.Send;
import org.apache.kafka.common.protocol.ApiKeys;
+import org.apache.kafka.common.protocol.Errors;
import org.apache.kafka.common.protocol.types.Struct;
import java.nio.ByteBuffer;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
public abstract class AbstractResponse extends AbstractRequestResponse {
public static final int DEFAULT_THROTTLE_TIME = 0;
@@ -37,6 +41,31 @@ public abstract class AbstractResponse extends AbstractRequestResponse {
return serialize(responseHeader.toStruct(), toStruct(version));
}
+ public abstract Map<Errors, Integer> errorCounts();
+
+ protected Map<Errors, Integer> errorCounts(Errors error) {
+ return Collections.singletonMap(error, 1);
+ }
+
+ protected Map<Errors, Integer> errorCounts(Map<?, Errors> errors) {
+ Map<Errors, Integer> errorCounts = new HashMap<>();
+ for (Errors error : errors.values())
+ updateErrorCounts(errorCounts, error);
+ return errorCounts;
+ }
+
+ protected Map<Errors, Integer> apiErrorCounts(Map<?, ApiError> errors) {
+ Map<Errors, Integer> errorCounts = new HashMap<>();
+ for (ApiError apiError : errors.values())
+ updateErrorCounts(errorCounts, apiError.error());
+ return errorCounts;
+ }
+
+ protected void updateErrorCounts(Map<Errors, Integer> errorCounts, Errors error) {
+ Integer count = errorCounts.get(error);
+ errorCounts.put(error, count == null ? 1 : count + 1);
+ }
+
protected abstract Struct toStruct(short version);
public static AbstractResponse parseResponse(ApiKeys apiKey, Struct struct) {
http://git-wip-us.apache.org/repos/asf/kafka/blob/021d8a8e/clients/src/main/java/org/apache/kafka/common/requests/AddOffsetsToTxnResponse.java
----------------------------------------------------------------------
diff --git a/clients/src/main/java/org/apache/kafka/common/requests/AddOffsetsToTxnResponse.java b/clients/src/main/java/org/apache/kafka/common/requests/AddOffsetsToTxnResponse.java
index 10dc279..69b8ad4 100644
--- a/clients/src/main/java/org/apache/kafka/common/requests/AddOffsetsToTxnResponse.java
+++ b/clients/src/main/java/org/apache/kafka/common/requests/AddOffsetsToTxnResponse.java
@@ -22,6 +22,7 @@ import org.apache.kafka.common.protocol.types.Schema;
import org.apache.kafka.common.protocol.types.Struct;
import java.nio.ByteBuffer;
+import java.util.Map;
import static org.apache.kafka.common.protocol.CommonFields.ERROR_CODE;
import static org.apache.kafka.common.protocol.CommonFields.THROTTLE_TIME_MS;
@@ -67,6 +68,11 @@ public class AddOffsetsToTxnResponse extends AbstractResponse {
}
@Override
+ public Map<Errors, Integer> errorCounts() {
+ return errorCounts(error);
+ }
+
+ @Override
protected Struct toStruct(short version) {
Struct struct = new Struct(ApiKeys.ADD_OFFSETS_TO_TXN.responseSchema(version));
struct.set(THROTTLE_TIME_MS, throttleTimeMs);
http://git-wip-us.apache.org/repos/asf/kafka/blob/021d8a8e/clients/src/main/java/org/apache/kafka/common/requests/AddPartitionsToTxnResponse.java
----------------------------------------------------------------------
diff --git a/clients/src/main/java/org/apache/kafka/common/requests/AddPartitionsToTxnResponse.java b/clients/src/main/java/org/apache/kafka/common/requests/AddPartitionsToTxnResponse.java
index e9f6088..4472449 100644
--- a/clients/src/main/java/org/apache/kafka/common/requests/AddPartitionsToTxnResponse.java
+++ b/clients/src/main/java/org/apache/kafka/common/requests/AddPartitionsToTxnResponse.java
@@ -95,6 +95,11 @@ public class AddPartitionsToTxnResponse extends AbstractResponse {
}
@Override
+ public Map<Errors, Integer> errorCounts() {
+ return errorCounts(errors);
+ }
+
+ @Override
protected Struct toStruct(short version) {
Struct struct = new Struct(ApiKeys.ADD_PARTITIONS_TO_TXN.responseSchema(version));
struct.set(THROTTLE_TIME_MS, throttleTimeMs);
http://git-wip-us.apache.org/repos/asf/kafka/blob/021d8a8e/clients/src/main/java/org/apache/kafka/common/requests/AlterConfigsResponse.java
----------------------------------------------------------------------
diff --git a/clients/src/main/java/org/apache/kafka/common/requests/AlterConfigsResponse.java b/clients/src/main/java/org/apache/kafka/common/requests/AlterConfigsResponse.java
index df9416e..f292ef6 100644
--- a/clients/src/main/java/org/apache/kafka/common/requests/AlterConfigsResponse.java
+++ b/clients/src/main/java/org/apache/kafka/common/requests/AlterConfigsResponse.java
@@ -21,6 +21,7 @@ import org.apache.kafka.common.protocol.ApiKeys;
import org.apache.kafka.common.protocol.types.ArrayOf;
import org.apache.kafka.common.protocol.types.Field;
import org.apache.kafka.common.protocol.types.Schema;
+import org.apache.kafka.common.protocol.Errors;
import org.apache.kafka.common.protocol.types.Struct;
import java.nio.ByteBuffer;
@@ -81,6 +82,11 @@ public class AlterConfigsResponse extends AbstractResponse {
return errors;
}
+ @Override
+ public Map<Errors, Integer> errorCounts() {
+ return apiErrorCounts(errors);
+ }
+
public int throttleTimeMs() {
return throttleTimeMs;
}
http://git-wip-us.apache.org/repos/asf/kafka/blob/021d8a8e/clients/src/main/java/org/apache/kafka/common/requests/AlterReplicaDirResponse.java
----------------------------------------------------------------------
diff --git a/clients/src/main/java/org/apache/kafka/common/requests/AlterReplicaDirResponse.java b/clients/src/main/java/org/apache/kafka/common/requests/AlterReplicaDirResponse.java
index 1767d45..b875104 100644
--- a/clients/src/main/java/org/apache/kafka/common/requests/AlterReplicaDirResponse.java
+++ b/clients/src/main/java/org/apache/kafka/common/requests/AlterReplicaDirResponse.java
@@ -124,6 +124,11 @@ public class AlterReplicaDirResponse extends AbstractResponse {
return this.responses;
}
+ @Override
+ public Map<Errors, Integer> errorCounts() {
+ return errorCounts(responses);
+ }
+
public static AlterReplicaDirResponse parse(ByteBuffer buffer, short version) {
return new AlterReplicaDirResponse(ApiKeys.ALTER_REPLICA_DIR.responseSchema(version).read(buffer));
}
http://git-wip-us.apache.org/repos/asf/kafka/blob/021d8a8e/clients/src/main/java/org/apache/kafka/common/requests/ApiVersionsResponse.java
----------------------------------------------------------------------
diff --git a/clients/src/main/java/org/apache/kafka/common/requests/ApiVersionsResponse.java b/clients/src/main/java/org/apache/kafka/common/requests/ApiVersionsResponse.java
index 2bdc8aa..753db9e 100644
--- a/clients/src/main/java/org/apache/kafka/common/requests/ApiVersionsResponse.java
+++ b/clients/src/main/java/org/apache/kafka/common/requests/ApiVersionsResponse.java
@@ -159,6 +159,11 @@ public class ApiVersionsResponse extends AbstractResponse {
return error;
}
+ @Override
+ public Map<Errors, Integer> errorCounts() {
+ return errorCounts(error);
+ }
+
public static ApiVersionsResponse parse(ByteBuffer buffer, short version) {
return new ApiVersionsResponse(ApiKeys.API_VERSIONS.parseResponse(version, buffer));
}
http://git-wip-us.apache.org/repos/asf/kafka/blob/021d8a8e/clients/src/main/java/org/apache/kafka/common/requests/ControlledShutdownResponse.java
----------------------------------------------------------------------
diff --git a/clients/src/main/java/org/apache/kafka/common/requests/ControlledShutdownResponse.java b/clients/src/main/java/org/apache/kafka/common/requests/ControlledShutdownResponse.java
index dfd68e7..f9852dd 100644
--- a/clients/src/main/java/org/apache/kafka/common/requests/ControlledShutdownResponse.java
+++ b/clients/src/main/java/org/apache/kafka/common/requests/ControlledShutdownResponse.java
@@ -28,6 +28,7 @@ import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.HashSet;
import java.util.List;
+import java.util.Map;
import java.util.Set;
import static org.apache.kafka.common.protocol.CommonFields.ERROR_CODE;
@@ -85,6 +86,11 @@ public class ControlledShutdownResponse extends AbstractResponse {
return error;
}
+ @Override
+ public Map<Errors, Integer> errorCounts() {
+ return errorCounts(error);
+ }
+
public Set<TopicPartition> partitionsRemaining() {
return partitionsRemaining;
}
http://git-wip-us.apache.org/repos/asf/kafka/blob/021d8a8e/clients/src/main/java/org/apache/kafka/common/requests/CreateAclsResponse.java
----------------------------------------------------------------------
diff --git a/clients/src/main/java/org/apache/kafka/common/requests/CreateAclsResponse.java b/clients/src/main/java/org/apache/kafka/common/requests/CreateAclsResponse.java
index 836215e..6c35798 100644
--- a/clients/src/main/java/org/apache/kafka/common/requests/CreateAclsResponse.java
+++ b/clients/src/main/java/org/apache/kafka/common/requests/CreateAclsResponse.java
@@ -20,11 +20,14 @@ import org.apache.kafka.common.protocol.ApiKeys;
import org.apache.kafka.common.protocol.types.ArrayOf;
import org.apache.kafka.common.protocol.types.Field;
import org.apache.kafka.common.protocol.types.Schema;
+import org.apache.kafka.common.protocol.Errors;
import org.apache.kafka.common.protocol.types.Struct;
import java.nio.ByteBuffer;
import java.util.ArrayList;
+import java.util.HashMap;
import java.util.List;
+import java.util.Map;
import static org.apache.kafka.common.protocol.CommonFields.ERROR_CODE;
import static org.apache.kafka.common.protocol.CommonFields.ERROR_MESSAGE;
@@ -101,6 +104,14 @@ public class CreateAclsResponse extends AbstractResponse {
return aclCreationResponses;
}
+ @Override
+ public Map<Errors, Integer> errorCounts() {
+ Map<Errors, Integer> errorCounts = new HashMap<>();
+ for (AclCreationResponse response : aclCreationResponses)
+ updateErrorCounts(errorCounts, response.error.error());
+ return errorCounts;
+ }
+
public static CreateAclsResponse parse(ByteBuffer buffer, short version) {
return new CreateAclsResponse(ApiKeys.CREATE_ACLS.responseSchema(version).read(buffer));
}
http://git-wip-us.apache.org/repos/asf/kafka/blob/021d8a8e/clients/src/main/java/org/apache/kafka/common/requests/CreatePartitionsRequest.java
----------------------------------------------------------------------
diff --git a/clients/src/main/java/org/apache/kafka/common/requests/CreatePartitionsRequest.java b/clients/src/main/java/org/apache/kafka/common/requests/CreatePartitionsRequest.java
index 3e3cc30..41c5327 100644
--- a/clients/src/main/java/org/apache/kafka/common/requests/CreatePartitionsRequest.java
+++ b/clients/src/main/java/org/apache/kafka/common/requests/CreatePartitionsRequest.java
@@ -122,7 +122,7 @@ public class CreatePartitionsRequest extends AbstractRequest {
Object[] assignmentsArray = partitionCountStruct.getArray(ASSIGNMENT_KEY_NAME);
NewPartitions newPartition;
if (assignmentsArray != null) {
- List<List<Integer>> assignments = new ArrayList(assignmentsArray.length);
+ List<List<Integer>> assignments = new ArrayList<>(assignmentsArray.length);
for (Object replicas : assignmentsArray) {
Object[] replicasArray = (Object[]) replicas;
List<Integer> replicasList = new ArrayList<>(replicasArray.length);
http://git-wip-us.apache.org/repos/asf/kafka/blob/021d8a8e/clients/src/main/java/org/apache/kafka/common/requests/CreatePartitionsResponse.java
----------------------------------------------------------------------
diff --git a/clients/src/main/java/org/apache/kafka/common/requests/CreatePartitionsResponse.java b/clients/src/main/java/org/apache/kafka/common/requests/CreatePartitionsResponse.java
index 390221f..73a6b56 100644
--- a/clients/src/main/java/org/apache/kafka/common/requests/CreatePartitionsResponse.java
+++ b/clients/src/main/java/org/apache/kafka/common/requests/CreatePartitionsResponse.java
@@ -18,6 +18,7 @@
package org.apache.kafka.common.requests;
import org.apache.kafka.common.protocol.ApiKeys;
+import org.apache.kafka.common.protocol.Errors;
import org.apache.kafka.common.protocol.types.ArrayOf;
import org.apache.kafka.common.protocol.types.Field;
import org.apache.kafka.common.protocol.types.Schema;
@@ -94,6 +95,11 @@ public class CreatePartitionsResponse extends AbstractResponse {
return errors;
}
+ @Override
+ public Map<Errors, Integer> errorCounts() {
+ return apiErrorCounts(errors);
+ }
+
public int throttleTimeMs() {
return throttleTimeMs;
}
http://git-wip-us.apache.org/repos/asf/kafka/blob/021d8a8e/clients/src/main/java/org/apache/kafka/common/requests/CreateTopicsResponse.java
----------------------------------------------------------------------
diff --git a/clients/src/main/java/org/apache/kafka/common/requests/CreateTopicsResponse.java b/clients/src/main/java/org/apache/kafka/common/requests/CreateTopicsResponse.java
index b3c052b..1f73eda 100644
--- a/clients/src/main/java/org/apache/kafka/common/requests/CreateTopicsResponse.java
+++ b/clients/src/main/java/org/apache/kafka/common/requests/CreateTopicsResponse.java
@@ -21,6 +21,7 @@ import org.apache.kafka.common.protocol.ApiKeys;
import org.apache.kafka.common.protocol.types.ArrayOf;
import org.apache.kafka.common.protocol.types.Field;
import org.apache.kafka.common.protocol.types.Schema;
+import org.apache.kafka.common.protocol.Errors;
import org.apache.kafka.common.protocol.types.Struct;
import java.nio.ByteBuffer;
@@ -125,6 +126,11 @@ public class CreateTopicsResponse extends AbstractResponse {
return errors;
}
+ @Override
+ public Map<Errors, Integer> errorCounts() {
+ return apiErrorCounts(errors);
+ }
+
public static CreateTopicsResponse parse(ByteBuffer buffer, short version) {
return new CreateTopicsResponse(ApiKeys.CREATE_TOPICS.responseSchema(version).read(buffer));
}
http://git-wip-us.apache.org/repos/asf/kafka/blob/021d8a8e/clients/src/main/java/org/apache/kafka/common/requests/DeleteAclsResponse.java
----------------------------------------------------------------------
diff --git a/clients/src/main/java/org/apache/kafka/common/requests/DeleteAclsResponse.java b/clients/src/main/java/org/apache/kafka/common/requests/DeleteAclsResponse.java
index 0857287..7ae25da 100644
--- a/clients/src/main/java/org/apache/kafka/common/requests/DeleteAclsResponse.java
+++ b/clients/src/main/java/org/apache/kafka/common/requests/DeleteAclsResponse.java
@@ -22,6 +22,7 @@ import org.apache.kafka.common.protocol.ApiKeys;
import org.apache.kafka.common.protocol.types.ArrayOf;
import org.apache.kafka.common.protocol.types.Field;
import org.apache.kafka.common.protocol.types.Schema;
+import org.apache.kafka.common.protocol.Errors;
import org.apache.kafka.common.protocol.types.Struct;
import org.apache.kafka.common.resource.Resource;
import org.apache.kafka.common.utils.Utils;
@@ -31,7 +32,9 @@ import org.slf4j.LoggerFactory;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.Collection;
+import java.util.HashMap;
import java.util.List;
+import java.util.Map;
import static org.apache.kafka.common.protocol.CommonFields.ERROR_CODE;
import static org.apache.kafka.common.protocol.CommonFields.ERROR_MESSAGE;
@@ -182,6 +185,14 @@ public class DeleteAclsResponse extends AbstractResponse {
return responses;
}
+ @Override
+ public Map<Errors, Integer> errorCounts() {
+ Map<Errors, Integer> errorCounts = new HashMap<>();
+ for (AclFilterResponse response : responses)
+ updateErrorCounts(errorCounts, response.error.error());
+ return errorCounts;
+ }
+
public static DeleteAclsResponse parse(ByteBuffer buffer, short version) {
return new DeleteAclsResponse(ApiKeys.DELETE_ACLS.responseSchema(version).read(buffer));
}
http://git-wip-us.apache.org/repos/asf/kafka/blob/021d8a8e/clients/src/main/java/org/apache/kafka/common/requests/DeleteRecordsResponse.java
----------------------------------------------------------------------
diff --git a/clients/src/main/java/org/apache/kafka/common/requests/DeleteRecordsResponse.java b/clients/src/main/java/org/apache/kafka/common/requests/DeleteRecordsResponse.java
index 5bfdec8..b839bf5 100644
--- a/clients/src/main/java/org/apache/kafka/common/requests/DeleteRecordsResponse.java
+++ b/clients/src/main/java/org/apache/kafka/common/requests/DeleteRecordsResponse.java
@@ -161,6 +161,14 @@ public class DeleteRecordsResponse extends AbstractResponse {
return this.responses;
}
+ @Override
+ public Map<Errors, Integer> errorCounts() {
+ Map<Errors, Integer> errorCounts = new HashMap<>();
+ for (PartitionResponse response : responses.values())
+ updateErrorCounts(errorCounts, response.error);
+ return errorCounts;
+ }
+
public static DeleteRecordsResponse parse(ByteBuffer buffer, short version) {
return new DeleteRecordsResponse(ApiKeys.DELETE_RECORDS.responseSchema(version).read(buffer));
}
http://git-wip-us.apache.org/repos/asf/kafka/blob/021d8a8e/clients/src/main/java/org/apache/kafka/common/requests/DeleteTopicsResponse.java
----------------------------------------------------------------------
diff --git a/clients/src/main/java/org/apache/kafka/common/requests/DeleteTopicsResponse.java b/clients/src/main/java/org/apache/kafka/common/requests/DeleteTopicsResponse.java
index 9c84c11..d13e441 100644
--- a/clients/src/main/java/org/apache/kafka/common/requests/DeleteTopicsResponse.java
+++ b/clients/src/main/java/org/apache/kafka/common/requests/DeleteTopicsResponse.java
@@ -110,6 +110,11 @@ public class DeleteTopicsResponse extends AbstractResponse {
return errors;
}
+ @Override
+ public Map<Errors, Integer> errorCounts() {
+ return errorCounts(errors);
+ }
+
public static DeleteTopicsResponse parse(ByteBuffer buffer, short version) {
return new DeleteTopicsResponse(ApiKeys.DELETE_TOPICS.responseSchema(version).read(buffer));
}
http://git-wip-us.apache.org/repos/asf/kafka/blob/021d8a8e/clients/src/main/java/org/apache/kafka/common/requests/DescribeAclsResponse.java
----------------------------------------------------------------------
diff --git a/clients/src/main/java/org/apache/kafka/common/requests/DescribeAclsResponse.java b/clients/src/main/java/org/apache/kafka/common/requests/DescribeAclsResponse.java
index f8b9695..a21230b 100644
--- a/clients/src/main/java/org/apache/kafka/common/requests/DescribeAclsResponse.java
+++ b/clients/src/main/java/org/apache/kafka/common/requests/DescribeAclsResponse.java
@@ -23,6 +23,7 @@ import org.apache.kafka.common.protocol.ApiKeys;
import org.apache.kafka.common.protocol.types.ArrayOf;
import org.apache.kafka.common.protocol.types.Field;
import org.apache.kafka.common.protocol.types.Schema;
+import org.apache.kafka.common.protocol.Errors;
import org.apache.kafka.common.protocol.types.Struct;
import org.apache.kafka.common.resource.Resource;
@@ -133,6 +134,11 @@ public class DescribeAclsResponse extends AbstractResponse {
return error;
}
+ @Override
+ public Map<Errors, Integer> errorCounts() {
+ return errorCounts(error.error());
+ }
+
public Collection<AclBinding> acls() {
return acls;
}
http://git-wip-us.apache.org/repos/asf/kafka/blob/021d8a8e/clients/src/main/java/org/apache/kafka/common/requests/DescribeConfigsResponse.java
----------------------------------------------------------------------
diff --git a/clients/src/main/java/org/apache/kafka/common/requests/DescribeConfigsResponse.java b/clients/src/main/java/org/apache/kafka/common/requests/DescribeConfigsResponse.java
index 9b2289d..91bf30e 100644
--- a/clients/src/main/java/org/apache/kafka/common/requests/DescribeConfigsResponse.java
+++ b/clients/src/main/java/org/apache/kafka/common/requests/DescribeConfigsResponse.java
@@ -21,6 +21,7 @@ import org.apache.kafka.common.protocol.ApiKeys;
import org.apache.kafka.common.protocol.types.ArrayOf;
import org.apache.kafka.common.protocol.types.Field;
import org.apache.kafka.common.protocol.types.Schema;
+import org.apache.kafka.common.protocol.Errors;
import org.apache.kafka.common.protocol.types.Struct;
import java.nio.ByteBuffer;
@@ -176,6 +177,14 @@ public class DescribeConfigsResponse extends AbstractResponse {
}
@Override
+ public Map<Errors, Integer> errorCounts() {
+ Map<Errors, Integer> errorCounts = new HashMap<>();
+ for (Config response : configs.values())
+ updateErrorCounts(errorCounts, response.error.error());
+ return errorCounts;
+ }
+
+ @Override
protected Struct toStruct(short version) {
Struct struct = new Struct(ApiKeys.DESCRIBE_CONFIGS.responseSchema(version));
struct.set(THROTTLE_TIME_MS, throttleTimeMs);
http://git-wip-us.apache.org/repos/asf/kafka/blob/021d8a8e/clients/src/main/java/org/apache/kafka/common/requests/DescribeGroupsResponse.java
----------------------------------------------------------------------
diff --git a/clients/src/main/java/org/apache/kafka/common/requests/DescribeGroupsResponse.java b/clients/src/main/java/org/apache/kafka/common/requests/DescribeGroupsResponse.java
index 9241165..61c5a36 100644
--- a/clients/src/main/java/org/apache/kafka/common/requests/DescribeGroupsResponse.java
+++ b/clients/src/main/java/org/apache/kafka/common/requests/DescribeGroupsResponse.java
@@ -141,6 +141,13 @@ public class DescribeGroupsResponse extends AbstractResponse {
return groups;
}
+ @Override
+ public Map<Errors, Integer> errorCounts() {
+ Map<Errors, Integer> errorCounts = new HashMap<>();
+ for (GroupMetadata response : groups.values())
+ updateErrorCounts(errorCounts, response.error);
+ return errorCounts;
+ }
public static class GroupMetadata {
private final Errors error;
http://git-wip-us.apache.org/repos/asf/kafka/blob/021d8a8e/clients/src/main/java/org/apache/kafka/common/requests/DescribeLogDirsResponse.java
----------------------------------------------------------------------
diff --git a/clients/src/main/java/org/apache/kafka/common/requests/DescribeLogDirsResponse.java b/clients/src/main/java/org/apache/kafka/common/requests/DescribeLogDirsResponse.java
index dc226d8..6613dfe 100644
--- a/clients/src/main/java/org/apache/kafka/common/requests/DescribeLogDirsResponse.java
+++ b/clients/src/main/java/org/apache/kafka/common/requests/DescribeLogDirsResponse.java
@@ -162,6 +162,14 @@ public class DescribeLogDirsResponse extends AbstractResponse {
return throttleTimeMs;
}
+ @Override
+ public Map<Errors, Integer> errorCounts() {
+ Map<Errors, Integer> errorCounts = new HashMap<>();
+ for (LogDirInfo logDirInfo : logDirInfos.values())
+ updateErrorCounts(errorCounts, logDirInfo.error);
+ return errorCounts;
+ }
+
public Map<String, LogDirInfo> logDirInfos() {
return logDirInfos;
}
http://git-wip-us.apache.org/repos/asf/kafka/blob/021d8a8e/clients/src/main/java/org/apache/kafka/common/requests/EndTxnResponse.java
----------------------------------------------------------------------
diff --git a/clients/src/main/java/org/apache/kafka/common/requests/EndTxnResponse.java b/clients/src/main/java/org/apache/kafka/common/requests/EndTxnResponse.java
index a3bae58..a0a453d 100644
--- a/clients/src/main/java/org/apache/kafka/common/requests/EndTxnResponse.java
+++ b/clients/src/main/java/org/apache/kafka/common/requests/EndTxnResponse.java
@@ -22,6 +22,7 @@ import org.apache.kafka.common.protocol.types.Schema;
import org.apache.kafka.common.protocol.types.Struct;
import java.nio.ByteBuffer;
+import java.util.Map;
import static org.apache.kafka.common.protocol.CommonFields.ERROR_CODE;
import static org.apache.kafka.common.protocol.CommonFields.THROTTLE_TIME_MS;
@@ -66,6 +67,11 @@ public class EndTxnResponse extends AbstractResponse {
}
@Override
+ public Map<Errors, Integer> errorCounts() {
+ return errorCounts(error);
+ }
+
+ @Override
protected Struct toStruct(short version) {
Struct struct = new Struct(ApiKeys.END_TXN.responseSchema(version));
struct.set(THROTTLE_TIME_MS, throttleTimeMs);
http://git-wip-us.apache.org/repos/asf/kafka/blob/021d8a8e/clients/src/main/java/org/apache/kafka/common/requests/FetchResponse.java
----------------------------------------------------------------------
diff --git a/clients/src/main/java/org/apache/kafka/common/requests/FetchResponse.java b/clients/src/main/java/org/apache/kafka/common/requests/FetchResponse.java
index f8d3090..0d09027 100644
--- a/clients/src/main/java/org/apache/kafka/common/requests/FetchResponse.java
+++ b/clients/src/main/java/org/apache/kafka/common/requests/FetchResponse.java
@@ -30,6 +30,7 @@ import org.apache.kafka.common.record.Records;
import java.nio.ByteBuffer;
import java.util.ArrayList;
+import java.util.HashMap;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
@@ -353,6 +354,14 @@ public class FetchResponse extends AbstractResponse {
return this.throttleTimeMs;
}
+ @Override
+ public Map<Errors, Integer> errorCounts() {
+ Map<Errors, Integer> errorCounts = new HashMap<>();
+ for (PartitionData response : responseData.values())
+ updateErrorCounts(errorCounts, response.error);
+ return errorCounts;
+ }
+
public static FetchResponse parse(ByteBuffer buffer, short version) {
return new FetchResponse(ApiKeys.FETCH.responseSchema(version).read(buffer));
}
http://git-wip-us.apache.org/repos/asf/kafka/blob/021d8a8e/clients/src/main/java/org/apache/kafka/common/requests/FindCoordinatorResponse.java
----------------------------------------------------------------------
diff --git a/clients/src/main/java/org/apache/kafka/common/requests/FindCoordinatorResponse.java b/clients/src/main/java/org/apache/kafka/common/requests/FindCoordinatorResponse.java
index f5d9805..bddc41f 100644
--- a/clients/src/main/java/org/apache/kafka/common/requests/FindCoordinatorResponse.java
+++ b/clients/src/main/java/org/apache/kafka/common/requests/FindCoordinatorResponse.java
@@ -24,6 +24,7 @@ import org.apache.kafka.common.protocol.types.Schema;
import org.apache.kafka.common.protocol.types.Struct;
import java.nio.ByteBuffer;
+import java.util.Map;
import static org.apache.kafka.common.protocol.CommonFields.ERROR_CODE;
import static org.apache.kafka.common.protocol.CommonFields.ERROR_MESSAGE;
@@ -104,6 +105,11 @@ public class FindCoordinatorResponse extends AbstractResponse {
return error;
}
+ @Override
+ public Map<Errors, Integer> errorCounts() {
+ return errorCounts(error);
+ }
+
public Node node() {
return node;
}
http://git-wip-us.apache.org/repos/asf/kafka/blob/021d8a8e/clients/src/main/java/org/apache/kafka/common/requests/HeartbeatResponse.java
----------------------------------------------------------------------
diff --git a/clients/src/main/java/org/apache/kafka/common/requests/HeartbeatResponse.java b/clients/src/main/java/org/apache/kafka/common/requests/HeartbeatResponse.java
index e41d937..b52a993 100644
--- a/clients/src/main/java/org/apache/kafka/common/requests/HeartbeatResponse.java
+++ b/clients/src/main/java/org/apache/kafka/common/requests/HeartbeatResponse.java
@@ -22,6 +22,7 @@ import org.apache.kafka.common.protocol.types.Schema;
import org.apache.kafka.common.protocol.types.Struct;
import java.nio.ByteBuffer;
+import java.util.Map;
import static org.apache.kafka.common.protocol.CommonFields.ERROR_CODE;
import static org.apache.kafka.common.protocol.CommonFields.THROTTLE_TIME_MS;
@@ -74,6 +75,11 @@ public class HeartbeatResponse extends AbstractResponse {
}
@Override
+ public Map<Errors, Integer> errorCounts() {
+ return errorCounts(error);
+ }
+
+ @Override
protected Struct toStruct(short version) {
Struct struct = new Struct(ApiKeys.HEARTBEAT.responseSchema(version));
struct.setIfExists(THROTTLE_TIME_MS, throttleTimeMs);