1) If no partition is specified but a key is present, choose a partition based on a hash of the key." + - "
2) If no partition or key is present, choose the sticky partition that changes when at least " + BATCH_SIZE_CONFIG + " bytes are produced to the partition." + + "
org.apache.kafka.clients.producer.RoundRobinPartitioner: A partitioning strategy where " +
"each record in a series of consecutive records is sent to a different partition, regardless of whether the 'key' is provided or not, " +
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 2cdc3b876d..99bc1d68b0 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
@@ -706,9 +706,8 @@ private void completeBatch(ProducerBatch batch, ProduceResponse.PartitionRespons
"topic-partition may not exist or the user may not have Describe access to it",
batch.topicPartition);
} else {
- log.warn("Received invalid metadata error in produce request on partition {} due to {}. Going " +
- "to request metadata update now", batch.topicPartition,
- error.exception(response.errorMessage).toString());
+ log.warn("Received invalid metadata error in produce request on partition {} due to {} Going " +
+ "to request metadata update now", batch.topicPartition, error.exception(response.errorMessage).toString());
}
if (error.exception() instanceof NotLeaderOrFollowerException || error.exception() instanceof FencedLeaderEpochException) {
log.debug("For {}, received error {}, with leaderIdAndEpoch {}", batch.topicPartition, error, response.currentLeader);
diff --git a/clients/src/main/java/org/apache/kafka/clients/producer/internals/TransactionManager.java b/clients/src/main/java/org/apache/kafka/clients/producer/internals/TransactionManager.java
index 14a519a7a0..4ff54b1759 100644
--- a/clients/src/main/java/org/apache/kafka/clients/producer/internals/TransactionManager.java
+++ b/clients/src/main/java/org/apache/kafka/clients/producer/internals/TransactionManager.java
@@ -1330,6 +1330,8 @@ public void handleResponse(AbstractResponse response) {
// We could still receive INVALID_PRODUCER_EPOCH from old versioned transaction coordinator,
// just treat it the same as PRODUCE_FENCED.
fatalError(Errors.PRODUCER_FENCED.exception());
+ } else if (error == Errors.ABORTABLE_TRANSACTION) {
+ abortableError(error.exception());
} else {
fatalError(new KafkaException("Unexpected error in InitProducerIdResponse; " + error.message()));
}
@@ -1386,10 +1388,8 @@ public void handleResponse(AbstractResponse response) {
// just treat it the same as PRODUCE_FENCED.
fatalError(Errors.PRODUCER_FENCED.exception());
return;
- } else if (error == Errors.TRANSACTIONAL_ID_AUTHORIZATION_FAILED) {
- fatalError(error.exception());
- return;
- } else if (error == Errors.INVALID_TXN_STATE) {
+ } else if (error == Errors.TRANSACTIONAL_ID_AUTHORIZATION_FAILED ||
+ error == Errors.INVALID_TXN_STATE) {
fatalError(error.exception());
return;
} else if (error == Errors.TOPIC_AUTHORIZATION_FAILED) {
@@ -1401,6 +1401,9 @@ public void handleResponse(AbstractResponse response) {
} else if (error == Errors.UNKNOWN_PRODUCER_ID || error == Errors.INVALID_PRODUCER_ID_MAPPING) {
abortableErrorIfPossible(error.exception());
return;
+ } else if (error == Errors.ABORTABLE_TRANSACTION) {
+ abortableError(error.exception());
+ return;
} else {
log.error("Could not add partition {} due to unexpected error {}", topicPartition, error);
hasPartitionErrors = true;
@@ -1504,6 +1507,8 @@ public void handleResponse(AbstractResponse response) {
fatalError(error.exception());
} else if (error == Errors.GROUP_AUTHORIZATION_FAILED) {
abortableError(GroupAuthorizationException.forGroupId(key));
+ } else if (error == Errors.ABORTABLE_TRANSACTION) {
+ abortableError(error.exception());
} else {
fatalError(new KafkaException(String.format("Could not find a coordinator with type %s with key %s due to " +
"unexpected error: %s", coordinatorType, key,
@@ -1552,12 +1557,13 @@ public void handleResponse(AbstractResponse response) {
// We could still receive INVALID_PRODUCER_EPOCH from old versioned transaction coordinator,
// just treat it the same as PRODUCE_FENCED.
fatalError(Errors.PRODUCER_FENCED.exception());
- } else if (error == Errors.TRANSACTIONAL_ID_AUTHORIZATION_FAILED) {
- fatalError(error.exception());
- } else if (error == Errors.INVALID_TXN_STATE) {
+ } else if (error == Errors.TRANSACTIONAL_ID_AUTHORIZATION_FAILED ||
+ error == Errors.INVALID_TXN_STATE) {
fatalError(error.exception());
} else if (error == Errors.UNKNOWN_PRODUCER_ID || error == Errors.INVALID_PRODUCER_ID_MAPPING) {
abortableErrorIfPossible(error.exception());
+ } else if (error == Errors.ABORTABLE_TRANSACTION) {
+ abortableError(error.exception());
} else {
fatalError(new KafkaException("Unhandled error in EndTxnResponse: " + error.message()));
}
@@ -1611,12 +1617,13 @@ public void handleResponse(AbstractResponse response) {
// We could still receive INVALID_PRODUCER_EPOCH from old versioned transaction coordinator,
// just treat it the same as PRODUCE_FENCED.
fatalError(Errors.PRODUCER_FENCED.exception());
- } else if (error == Errors.TRANSACTIONAL_ID_AUTHORIZATION_FAILED) {
- fatalError(error.exception());
- } else if (error == Errors.INVALID_TXN_STATE) {
+ } else if (error == Errors.TRANSACTIONAL_ID_AUTHORIZATION_FAILED ||
+ error == Errors.INVALID_TXN_STATE) {
fatalError(error.exception());
} else if (error == Errors.GROUP_AUTHORIZATION_FAILED) {
abortableError(GroupAuthorizationException.forGroupId(builder.data.groupId()));
+ } else if (error == Errors.ABORTABLE_TRANSACTION) {
+ abortableError(error.exception());
} else {
fatalError(new KafkaException("Unexpected error in AddOffsetsToTxnResponse: " + error.message()));
}
@@ -1679,7 +1686,8 @@ public void handleResponse(AbstractResponse response) {
} else if (error == Errors.GROUP_AUTHORIZATION_FAILED) {
abortableError(GroupAuthorizationException.forGroupId(builder.data.groupId()));
break;
- } else if (error == Errors.FENCED_INSTANCE_ID) {
+ } else if (error == Errors.FENCED_INSTANCE_ID ||
+ error == Errors.ABORTABLE_TRANSACTION) {
abortableError(error.exception());
break;
} else if (error == Errors.UNKNOWN_MEMBER_ID
diff --git a/clients/src/main/java/org/apache/kafka/common/errors/AbortableTransactionException.java b/clients/src/main/java/org/apache/kafka/common/errors/AbortableTransactionException.java
new file mode 100644
index 0000000000..cfea649e75
--- /dev/null
+++ b/clients/src/main/java/org/apache/kafka/common/errors/AbortableTransactionException.java
@@ -0,0 +1,23 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.common.errors;
+
+public class AbortableTransactionException extends ApiException {
+ public AbortableTransactionException(String message) {
+ super(message);
+ }
+}
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 6d395fa3fb..c8e53ffc6c 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
@@ -26,9 +26,17 @@ public final class KafkaMetric implements Metric {
private final Object lock;
private final Time time;
private final MetricValueProvider metricValueProvider;
- private MetricConfig config;
+ private volatile MetricConfig config;
// public for testing
+ /**
+ * Create a metric to monitor an object that implements MetricValueProvider.
+ * @param lock The lock used to prevent race condition
+ * @param metricName The name of the metric
+ * @param valueProvider The metric value provider associated with this metric
+ * @param config The configuration of the metric
+ * @param time The time instance to use with the metrics
+ */
public KafkaMetric(Object lock, MetricName metricName, MetricValueProvider valueProvider,
MetricConfig config, Time time) {
this.metricName = metricName;
@@ -40,15 +48,29 @@ public KafkaMetric(Object lock, MetricName metricName, MetricValueProvider va
this.time = time;
}
+ /**
+ * Get the configuration of this metric.
+ * This is supposed to be used by server only.
+ * @return Return the config of this metric
+ */
public MetricConfig config() {
return this.config;
}
+ /**
+ * Get the metric name
+ * @return Return the name of this metric
+ */
@Override
public MetricName metricName() {
return this.metricName;
}
+ /**
+ * Take the metric and return the value, which could be a {@link Measurable} or a {@link Gauge}
+ * @return Return the metric value
+ * @throws IllegalStateException if the underlying metric is not a {@link Measurable} or a {@link Gauge}.
+ */
@Override
public Object metricValue() {
long now = time.milliseconds();
@@ -62,6 +84,11 @@ else if (this.metricValueProvider instanceof Gauge)
}
}
+ /**
+ * Get the underlying metric provider, which should be a {@link Measurable}
+ * @return Return the metric provider
+ * @throws IllegalStateException if the underlying metric is not a {@link Measurable}.
+ */
public Measurable measurable() {
if (this.metricValueProvider instanceof Measurable)
return (Measurable) metricValueProvider;
@@ -69,6 +96,11 @@ public Measurable measurable() {
throw new IllegalStateException("Not a measurable: " + this.metricValueProvider.getClass());
}
+ /**
+ * Take the metric and return the value, where the underlying metric provider should be a {@link Measurable}
+ * @param timeMs The time that this metric is taken
+ * @return Return the metric value if it's measurable, otherwise 0
+ */
double measurableValue(long timeMs) {
synchronized (this.lock) {
if (this.metricValueProvider instanceof Measurable)
@@ -78,6 +110,11 @@ public Measurable measurable() {
}
}
+ /**
+ * Set the metric config.
+ * This is supposed to be used by server only.
+ * @param config configuration for this metrics
+ */
public void config(MetricConfig config) {
synchronized (lock) {
this.config = config;
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 6447cdb5c7..b52285dac6 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
@@ -277,40 +277,30 @@ public static String toHtmlTable(String domain, Iterable| Metric/Attribute name | \n"); + b.append("Description | \n"); + b.append("Mbean name | \n"); + b.append("|
|---|---|---|---|
| "); - b.append(e.getKey()); - b.append(" | "); - b.append("|||
| \n"); - b.append(" | Attribute name | \n"); - b.append("Description | \n"); - b.append("|
| "); b.append(" | "); b.append(e2.getKey()); - b.append(" | "); + b.append("\n"); b.append(""); b.append(e2.getValue()); - b.append(" | "); + b.append("\n"); + b.append(""); + b.append(e.getKey()); + b.append(" | \n"); b.append("