You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kafka.apache.org by jg...@apache.org on 2019/01/09 08:03:40 UTC
[kafka] branch 2.1 updated (cf4355e -> 6a8ed79)
This is an automated email from the ASF dual-hosted git repository.
jgus pushed a change to branch 2.1
in repository https://gitbox.apache.org/repos/asf/kafka.git.
from cf4355e KAFKA-7253; The returned connector type is always null when creating connector (#5470)
new f3d4837 MINOR: Simplify handling of KafkaProducer serializer overrides
new 6a8ed79 KAFKA-6833; Producer should await metadata for unknown partitions (#6073)
The 2 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../kafka/clients/producer/KafkaProducer.java | 61 +++--
.../kafka/clients/producer/KafkaProducerTest.java | 305 ++++++++++++---------
.../kafka/api/BaseProducerSendTest.scala | 20 +-
.../kafka/api/ProducerFailureHandlingTest.scala | 14 +-
4 files changed, 236 insertions(+), 164 deletions(-)
[kafka] 01/02: MINOR: Simplify handling of KafkaProducer serializer
overrides
Posted by jg...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
jgus pushed a commit to branch 2.1
in repository https://gitbox.apache.org/repos/asf/kafka.git
commit f3d4837ceed3984c6914c6698afcea1e778a3d34
Author: Ismael Juma <is...@juma.me.uk>
AuthorDate: Thu Oct 25 16:26:55 2018 -0700
MINOR: Simplify handling of KafkaProducer serializer overrides
Author: Ismael Juma <is...@juma.me.uk>
Reviewers: Dong Lin <li...@gmail.com>
Closes #5781 from ijuma/simplify-producer-constructor
---
.../kafka/clients/producer/KafkaProducer.java | 30 +++--
.../kafka/clients/producer/KafkaProducerTest.java | 127 ++++++++++-----------
2 files changed, 85 insertions(+), 72 deletions(-)
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 9486c3d..5676d99 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
@@ -18,6 +18,7 @@ package org.apache.kafka.clients.producer;
import java.net.InetSocketAddress;
import java.util.Collections;
+import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Objects;
@@ -265,7 +266,7 @@ public class KafkaProducer<K, V> implements Producer<K, V> {
*
*/
public KafkaProducer(final Map<String, Object> configs) {
- this(new ProducerConfig(configs), null, null, null, null, null, Time.SYSTEM);
+ this(configs, null, null, null, null, null, Time.SYSTEM);
}
/**
@@ -282,8 +283,7 @@ public class KafkaProducer<K, V> implements Producer<K, V> {
* be called in the producer when the serializer is passed in directly.
*/
public KafkaProducer(Map<String, Object> configs, Serializer<K> keySerializer, Serializer<V> valueSerializer) {
- this(new ProducerConfig(ProducerConfig.addSerializerToConfig(configs, keySerializer, valueSerializer)),
- keySerializer, valueSerializer, null, null, null, Time.SYSTEM);
+ this(configs, keySerializer, valueSerializer, null, null, null, Time.SYSTEM);
}
/**
@@ -294,7 +294,7 @@ public class KafkaProducer<K, V> implements Producer<K, V> {
* @param properties The producer configs
*/
public KafkaProducer(Properties properties) {
- this(new ProducerConfig(properties), null, null, null, null, null, Time.SYSTEM);
+ this(propsToMap(properties), null, null, null, null, null, Time.SYSTEM);
}
/**
@@ -309,19 +309,21 @@ public class KafkaProducer<K, V> implements Producer<K, V> {
* be called in the producer when the serializer is passed in directly.
*/
public KafkaProducer(Properties properties, Serializer<K> keySerializer, Serializer<V> valueSerializer) {
- this(new ProducerConfig(ProducerConfig.addSerializerToConfig(properties, keySerializer, valueSerializer)),
- keySerializer, valueSerializer, null, null, null, Time.SYSTEM);
+ this(propsToMap(properties), keySerializer, valueSerializer, null, null, null,
+ Time.SYSTEM);
}
// visible for testing
@SuppressWarnings("unchecked")
- KafkaProducer(ProducerConfig config,
+ KafkaProducer(Map<String, Object> configs,
Serializer<K> keySerializer,
Serializer<V> valueSerializer,
Metadata metadata,
KafkaClient kafkaClient,
ProducerInterceptors interceptors,
Time time) {
+ ProducerConfig config = new ProducerConfig(ProducerConfig.addSerializerToConfig(configs, keySerializer,
+ valueSerializer));
try {
Map<String, Object> userProvidedConfigs = config.originals();
this.producerConfig = config;
@@ -1184,6 +1186,20 @@ public class KafkaProducer<K, V> implements Producer<K, V> {
}
}
+ private static Map<String, Object> propsToMap(Properties properties) {
+ Map<String, Object> map = new HashMap<>(properties.size());
+ for (Map.Entry<Object, Object> entry : properties.entrySet()) {
+ if (entry.getKey() instanceof String) {
+ String k = (String) entry.getKey();
+ map.put(k, properties.get(k));
+ }
+ else {
+ throw new ConfigException(entry.getKey().toString(), entry.getValue(), "Key must be a string.");
+ }
+ }
+ return map;
+ }
+
private ClusterResourceListeners configureClusterResourceListeners(Serializer<K> keySerializer, Serializer<V> valueSerializer, List<?>... candidateLists) {
ClusterResourceListeners clusterResourceListeners = new ClusterResourceListeners();
for (List<?> candidateList: candidateLists)
diff --git a/clients/src/test/java/org/apache/kafka/clients/producer/KafkaProducerTest.java b/clients/src/test/java/org/apache/kafka/clients/producer/KafkaProducerTest.java
index 77fcb51..b5d7709 100644
--- a/clients/src/test/java/org/apache/kafka/clients/producer/KafkaProducerTest.java
+++ b/clients/src/test/java/org/apache/kafka/clients/producer/KafkaProducerTest.java
@@ -128,6 +128,18 @@ public class KafkaProducerTest {
}
@Test
+ public void testConstructorWithNotStringKey() {
+ Properties props = new Properties();
+ props.setProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9999");
+ props.put(1, "not string key");
+ try (KafkaProducer<?, ?> ff = new KafkaProducer(props, new StringSerializer(), new StringSerializer())) {
+ fail("Constructor should throw exception");
+ } catch (ConfigException e) {
+ assertTrue("Unexpected exception message: " + e.getMessage(), e.getMessage().contains("not string key"));
+ }
+ }
+
+ @Test
public void testSerializerClose() {
Map<String, Object> configs = new HashMap<>();
configs.put(ProducerConfig.CLIENT_ID_CONFIG, "testConstructorClose");
@@ -197,10 +209,10 @@ public class KafkaProducerTest {
@Test
public void shouldCloseProperlyAndThrowIfInterrupted() throws Exception {
- Properties props = new Properties();
- props.setProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9999");
- props.setProperty(ProducerConfig.PARTITIONER_CLASS_CONFIG, MockPartitioner.class.getName());
- props.setProperty(ProducerConfig.BATCH_SIZE_CONFIG, "1");
+ Map<String, Object> configs = new HashMap();
+ configs.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9999");
+ configs.put(ProducerConfig.PARTITIONER_CLASS_CONFIG, MockPartitioner.class.getName());
+ configs.put(ProducerConfig.BATCH_SIZE_CONFIG, "1");
Time time = new MockTime();
Cluster cluster = TestUtils.singletonCluster("topic", 1);
@@ -212,9 +224,8 @@ public class KafkaProducerTest {
MockClient client = new MockClient(time, metadata);
client.setNode(node);
- final Producer<String, String> producer = new KafkaProducer<>(
- new ProducerConfig(ProducerConfig.addSerializerToConfig(props, new StringSerializer(), new StringSerializer())),
- new StringSerializer(), new StringSerializer(), metadata, client, null, time);
+ final Producer<String, String> producer = new KafkaProducer<>(configs, new StringSerializer(),
+ new StringSerializer(), metadata, client, null, time);
ExecutorService executor = Executors.newSingleThreadExecutor();
final AtomicReference<Exception> closeException = new AtomicReference<>();
@@ -278,10 +289,8 @@ public class KafkaProducerTest {
@Test
public void testMetadataFetch() throws InterruptedException {
- Properties props = new Properties();
- props.setProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9999");
- ProducerConfig config = new ProducerConfig(ProducerConfig.addSerializerToConfig(props, new StringSerializer(),
- new StringSerializer()));
+ Map<String, Object> configs = new HashMap<>();
+ configs.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9999");
String topic = "topic";
Collection<Node> nodes = Collections.singletonList(new Node(0, "host1", 1000));
final Cluster emptyCluster = new Cluster(null, nodes,
@@ -299,8 +308,8 @@ public class KafkaProducerTest {
// Return empty cluster 4 times and cluster from then on
when(metadata.fetch()).thenReturn(emptyCluster, emptyCluster, emptyCluster, emptyCluster, cluster);
- KafkaProducer<String, String> producer = new KafkaProducer<String, String>(config, null, null,
- metadata, new MockClient(Time.SYSTEM, metadata), null, Time.SYSTEM) {
+ KafkaProducer<String, String> producer = new KafkaProducer<String, String>(configs, new StringSerializer(),
+ new StringSerializer(), metadata, new MockClient(Time.SYSTEM, metadata), null, Time.SYSTEM) {
@Override
Sender newSender(LogContext logContext, KafkaClient kafkaClient, Metadata metadata) {
// give Sender its own Metadata instance so that we can isolate Metadata calls from KafkaProducer
@@ -332,10 +341,8 @@ public class KafkaProducerTest {
@Test
public void testMetadataFetchOnStaleMetadata() throws Exception {
- Properties props = new Properties();
- props.setProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9999");
- ProducerConfig config = new ProducerConfig(ProducerConfig.addSerializerToConfig(props, new StringSerializer(),
- new StringSerializer()));
+ Map<String, Object> configs = new HashMap<>();
+ configs.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9999");
String topic = "topic";
ProducerRecord<String, String> initialRecord = new ProducerRecord<>(topic, "value");
// Create a record with a partition higher than the initial (outdated) partition range
@@ -374,8 +381,8 @@ public class KafkaProducerTest {
return emptyCluster;
});
- KafkaProducer<String, String> producer = new KafkaProducer<String, String>(config, null, null,
- metadata, new MockClient(Time.SYSTEM, metadata), null, Time.SYSTEM) {
+ KafkaProducer<String, String> producer = new KafkaProducer<String, String>(configs, new StringSerializer(),
+ new StringSerializer(), metadata, new MockClient(Time.SYSTEM, metadata), null, Time.SYSTEM) {
@Override
Sender newSender(LogContext logContext, KafkaClient kafkaClient, Metadata metadata) {
// give Sender its own Metadata instance so that we can isolate Metadata calls from KafkaProducer
@@ -419,19 +426,17 @@ public class KafkaProducerTest {
@Test
public void testTopicRefreshInMetadata() throws InterruptedException {
- Properties props = new Properties();
- props.setProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9999");
- props.setProperty(ProducerConfig.MAX_BLOCK_MS_CONFIG, "600000");
- ProducerConfig config = new ProducerConfig(ProducerConfig.addSerializerToConfig(props, new StringSerializer(),
- new StringSerializer()));
+ Map<String, Object> configs = new HashMap<>();
+ configs.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9999");
+ configs.put(ProducerConfig.MAX_BLOCK_MS_CONFIG, "600000");
long refreshBackoffMs = 500L;
long metadataExpireMs = 60000L;
final Metadata metadata = new Metadata(refreshBackoffMs, metadataExpireMs, true,
true, new ClusterResourceListeners());
final Time time = new MockTime();
final String topic = "topic";
- try (KafkaProducer<String, String> producer = new KafkaProducer<>(config, null, null, metadata,
- null, null, time)) {
+ try (KafkaProducer<String, String> producer = new KafkaProducer<>(configs, new StringSerializer(),
+ new StringSerializer(), metadata, null, null, time)) {
Thread t = new Thread(() -> {
long startTimeMs = System.currentTimeMillis();
@@ -466,14 +471,12 @@ public class KafkaProducerTest {
}
private <T extends Serializer<String>> void doTestHeaders(Class<T> serializerClassToMock) {
- Properties props = new Properties();
- props.setProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9999");
+ Map<String, Object> configs = new HashMap<>();
+ configs.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9999");
@SuppressWarnings("unchecked") // it is safe to suppress, since this is a mock class
Serializer<String> keySerializer = mock(serializerClassToMock);
@SuppressWarnings("unchecked")
Serializer<String> valueSerializer = mock(serializerClassToMock);
- ProducerConfig config = new ProducerConfig(ProducerConfig.addSerializerToConfig(props, keySerializer,
- valueSerializer));
String topic = "topic";
final Cluster cluster = new Cluster(
@@ -485,8 +488,8 @@ public class KafkaProducerTest {
Metadata metadata = new Metadata(0, 90000, true);
metadata.update(cluster, Collections.emptySet(), Time.SYSTEM.milliseconds());
- KafkaProducer<String, String> producer = new KafkaProducer<>(config, keySerializer, valueSerializer,
- metadata, null, null, Time.SYSTEM);
+ KafkaProducer<String, String> producer = new KafkaProducer<>(configs, keySerializer, valueSerializer, metadata,
+ null, null, Time.SYSTEM);
when(keySerializer.serialize(any(), any(), any())).then(invocation ->
invocation.<String>getArgument(2).getBytes());
@@ -543,11 +546,9 @@ public class KafkaProducerTest {
@Test
public void testInterceptorPartitionSetOnTooLargeRecord() {
- Properties props = new Properties();
- props.setProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9999");
- props.setProperty(ProducerConfig.MAX_REQUEST_SIZE_CONFIG, "1");
- ProducerConfig config = new ProducerConfig(ProducerConfig.addSerializerToConfig(props, new StringSerializer(),
- new StringSerializer()));
+ Map<String, Object> configs = new HashMap<>();
+ configs.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9999");
+ configs.put(ProducerConfig.MAX_REQUEST_SIZE_CONFIG, "1");
String topic = "topic";
ProducerRecord<String, String> record = new ProducerRecord<>(topic, "value");
@@ -562,8 +563,8 @@ public class KafkaProducerTest {
@SuppressWarnings("unchecked") // it is safe to suppress, since this is a mock class
ProducerInterceptors<String, String> interceptors = mock(ProducerInterceptors.class);
- KafkaProducer<String, String> producer = new KafkaProducer<>(config, null, null,
- metadata, null, interceptors, Time.SYSTEM);
+ KafkaProducer<String, String> producer = new KafkaProducer<>(configs, new StringSerializer(),
+ new StringSerializer(), metadata, null, interceptors, Time.SYSTEM);
when(interceptors.onSend(any())).then(invocation -> invocation.getArgument(0));
@@ -589,10 +590,10 @@ public class KafkaProducerTest {
@Test(expected = TimeoutException.class)
public void testInitTransactionTimeout() {
- Properties props = new Properties();
- props.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, "bad-transaction");
- props.put(ProducerConfig.MAX_BLOCK_MS_CONFIG, 5);
- props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9000");
+ Map<String, Object> configs = new HashMap<>();
+ configs.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, "bad-transaction");
+ configs.put(ProducerConfig.MAX_BLOCK_MS_CONFIG, 5);
+ configs.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9000");
Time time = Time.SYSTEM;
Cluster cluster = TestUtils.singletonCluster("topic", 1);
@@ -604,9 +605,8 @@ public class KafkaProducerTest {
MockClient client = new MockClient(time, metadata);
client.setNode(node);
- try (Producer<String, String> producer = new KafkaProducer<>(
- new ProducerConfig(ProducerConfig.addSerializerToConfig(props, new StringSerializer(), new StringSerializer())),
- null, null, metadata, client, null, time)) {
+ try (Producer<String, String> producer = new KafkaProducer<>(configs, new StringSerializer(),
+ new StringSerializer(), metadata, client, null, time)) {
producer.initTransactions();
fail("initTransactions() should have raised TimeoutException");
}
@@ -614,10 +614,10 @@ public class KafkaProducerTest {
@Test(expected = KafkaException.class)
public void testOnlyCanExecuteCloseAfterInitTransactionsTimeout() {
- Properties props = new Properties();
- props.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, "bad-transaction");
- props.put(ProducerConfig.MAX_BLOCK_MS_CONFIG, 5);
- props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9000");
+ Map<String, Object> configs = new HashMap<>();
+ configs.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, "bad-transaction");
+ configs.put(ProducerConfig.MAX_BLOCK_MS_CONFIG, 5);
+ configs.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9000");
Time time = new MockTime();
Cluster cluster = TestUtils.singletonCluster("topic", 1);
@@ -629,9 +629,8 @@ public class KafkaProducerTest {
MockClient client = new MockClient(time, metadata);
client.setNode(node);
- Producer<String, String> producer = new KafkaProducer<>(
- new ProducerConfig(ProducerConfig.addSerializerToConfig(props, new StringSerializer(), new StringSerializer())),
- null, null, metadata, client, null, time);
+ Producer<String, String> producer = new KafkaProducer<>(configs, new StringSerializer(), new StringSerializer(),
+ metadata, client, null, time);
try {
producer.initTransactions();
} catch (TimeoutException e) {
@@ -647,9 +646,9 @@ public class KafkaProducerTest {
@Test
public void testSendToInvalidTopic() throws Exception {
- Properties props = new Properties();
- props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9000");
- props.put(ProducerConfig.MAX_BLOCK_MS_CONFIG, "15000");
+ Map<String, Object> configs = new HashMap<>();
+ configs.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9000");
+ configs.put(ProducerConfig.MAX_BLOCK_MS_CONFIG, "15000");
Time time = new MockTime();
Cluster cluster = TestUtils.singletonCluster();
@@ -661,9 +660,8 @@ public class KafkaProducerTest {
MockClient client = new MockClient(time, metadata);
client.setNode(node);
- Producer<String, String> producer = new KafkaProducer<>(new ProducerConfig(
- ProducerConfig.addSerializerToConfig(props, new StringSerializer(), new StringSerializer())),
- null, null, metadata, client, null, time);
+ Producer<String, String> producer = new KafkaProducer<>(configs, new StringSerializer(), new StringSerializer(),
+ metadata, client, null, time);
String invalidTopicName = "topic abc"; // Invalid topic name due to space
ProducerRecord<String, String> record = new ProducerRecord<>(invalidTopicName, "HelloKafka");
@@ -690,9 +688,9 @@ public class KafkaProducerTest {
@Test
public void testCloseWhenWaitingForMetadataUpdate() throws InterruptedException {
- Properties props = new Properties();
- props.put(ProducerConfig.MAX_BLOCK_MS_CONFIG, Long.MAX_VALUE);
- props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9000");
+ Map<String, Object> configs = new HashMap<>();
+ configs.put(ProducerConfig.MAX_BLOCK_MS_CONFIG, Long.MAX_VALUE);
+ configs.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9000");
// Simulate a case where metadata for a particular topic is not available. This will cause KafkaProducer#send to
// block in Metadata#awaitUpdate for the configured max.block.ms. When close() is invoked, KafkaProducer#send should
@@ -706,9 +704,8 @@ public class KafkaProducerTest {
MockClient client = new MockClient(time, metadata);
client.setNode(node);
- Producer<String, String> producer = new KafkaProducer<>(
- new ProducerConfig(ProducerConfig.addSerializerToConfig(props, new StringSerializer(), new StringSerializer())),
- new StringSerializer(), new StringSerializer(), metadata, client, null, time);
+ Producer<String, String> producer = new KafkaProducer<>(configs, new StringSerializer(), new StringSerializer(),
+ metadata, client, null, time);
ExecutorService executor = Executors.newSingleThreadExecutor();
final AtomicReference<Exception> sendException = new AtomicReference<>();
[kafka] 02/02: KAFKA-6833;
Producer should await metadata for unknown partitions (#6073)
Posted by jg...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
jgus pushed a commit to branch 2.1
in repository https://gitbox.apache.org/repos/asf/kafka.git
commit 6a8ed79f58e92498feed6fa4b3627d01bd9a242a
Author: Bob Barrett <bo...@outlook.com>
AuthorDate: Tue Jan 8 21:05:13 2019 -0500
KAFKA-6833; Producer should await metadata for unknown partitions (#6073)
This patch changes the behavior of KafkaProducer.waitOnMetadata to wait up to max.block.ms when the partition specified in the produce request is out of the range of partitions present in the metadata. This improves the user experience in the case when partitions are added to a topic and a client attempts to produce to one of the new partitions before the metadata has propagated to the brokers. Tested with unit tests.
Reviewers: Arjun Satish <ar...@confluent.io>, Jason Gustafson <ja...@confluent.io>
---
.../kafka/clients/producer/KafkaProducer.java | 31 ++--
.../kafka/clients/producer/KafkaProducerTest.java | 180 +++++++++++++--------
.../kafka/api/BaseProducerSendTest.scala | 20 ++-
.../kafka/api/ProducerFailureHandlingTest.scala | 14 +-
4 files changed, 152 insertions(+), 93 deletions(-)
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 5676d99..f4d1422 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
@@ -966,12 +966,15 @@ public class KafkaProducer<K, V> implements Producer<K, V> {
long begin = time.milliseconds();
long remainingWaitMs = maxWaitMs;
long elapsed;
- // Issue metadata requests until we have metadata for the topic or maxWaitTimeMs is exceeded.
- // In case we already have cached metadata for the topic, but the requested partition is greater
- // than expected, issue an update request only once. This is necessary in case the metadata
+ // Issue metadata requests until we have metadata for the topic and the requested partition,
+ // or until maxWaitTimeMs is exceeded. This is necessary in case the metadata
// is stale and the number of partitions for this topic has increased in the meantime.
do {
- log.trace("Requesting metadata update for topic {}.", topic);
+ if (partition != null) {
+ log.trace("Requesting metadata update for partition {} of topic {}.", partition, topic);
+ } else {
+ log.trace("Requesting metadata update for topic {}.", topic);
+ }
metadata.add(topic);
int version = metadata.requestUpdate();
sender.wakeup();
@@ -979,24 +982,26 @@ public class KafkaProducer<K, V> implements Producer<K, V> {
metadata.awaitUpdate(version, remainingWaitMs);
} catch (TimeoutException ex) {
// Rethrow with original maxWaitMs to prevent logging exception with remainingWaitMs
- throw new TimeoutException("Failed to update metadata after " + maxWaitMs + " ms.");
+ throw new TimeoutException(
+ String.format("Topic %s not present in metadata after %d ms.",
+ topic, maxWaitMs));
}
cluster = metadata.fetch();
elapsed = time.milliseconds() - begin;
- if (elapsed >= maxWaitMs)
- throw new TimeoutException("Failed to update metadata after " + maxWaitMs + " ms.");
+ if (elapsed >= maxWaitMs) {
+ throw new TimeoutException(partitionsCount == null ?
+ String.format("Topic %s not present in metadata after %d ms.",
+ topic, maxWaitMs) :
+ String.format("Partition %d of topic %s with partition count %d is not present in metadata after %d ms.",
+ partition, topic, partitionsCount, maxWaitMs));
+ }
if (cluster.unauthorizedTopics().contains(topic))
throw new TopicAuthorizationException(topic);
if (cluster.invalidTopics().contains(topic))
throw new InvalidTopicException(topic);
remainingWaitMs = maxWaitMs - elapsed;
partitionsCount = cluster.partitionCountForTopic(topic);
- } while (partitionsCount == null);
-
- if (partition != null && partition >= partitionsCount) {
- throw new KafkaException(
- String.format("Invalid partition given with record: %d is not in the range [0...%d).", partition, partitionsCount));
- }
+ } while (partitionsCount == null || (partition != null && partition >= partitionsCount));
return new ClusterAndWaitTime(cluster, elapsed);
}
diff --git a/clients/src/test/java/org/apache/kafka/clients/producer/KafkaProducerTest.java b/clients/src/test/java/org/apache/kafka/clients/producer/KafkaProducerTest.java
index b5d7709..7a442c2 100644
--- a/clients/src/test/java/org/apache/kafka/clients/producer/KafkaProducerTest.java
+++ b/clients/src/test/java/org/apache/kafka/clients/producer/KafkaProducerTest.java
@@ -58,6 +58,7 @@ import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
import java.util.Properties;
+import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
@@ -80,6 +81,27 @@ import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
public class KafkaProducerTest {
+ private String topic = "topic";
+ private Collection<Node> nodes = Collections.singletonList(new Node(0, "host1", 1000));
+ private final Cluster emptyCluster = new Cluster(null, nodes,
+ Collections.emptySet(),
+ Collections.emptySet(),
+ Collections.emptySet());
+ private final Cluster onePartitionCluster = new Cluster(
+ "dummy",
+ Collections.singletonList(new Node(0, "host1", 1000)),
+ Collections.singletonList(new PartitionInfo(topic, 0, null, null, null)),
+ Collections.emptySet(),
+ Collections.emptySet());
+ private final Cluster threePartitionCluster = new Cluster(
+ "dummy",
+ Collections.singletonList(new Node(0, "host1", 1000)),
+ Arrays.asList(
+ new PartitionInfo(topic, 0, null, null, null),
+ new PartitionInfo(topic, 1, null, null, null),
+ new PartitionInfo(topic, 2, null, null, null)),
+ Collections.emptySet(),
+ Collections.emptySet());
@Test
public void testMetricsReporterAutoGeneratedClientId() {
@@ -291,22 +313,10 @@ public class KafkaProducerTest {
public void testMetadataFetch() throws InterruptedException {
Map<String, Object> configs = new HashMap<>();
configs.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9999");
- String topic = "topic";
- Collection<Node> nodes = Collections.singletonList(new Node(0, "host1", 1000));
- final Cluster emptyCluster = new Cluster(null, nodes,
- Collections.emptySet(),
- Collections.emptySet(),
- Collections.emptySet());
- final Cluster cluster = new Cluster(
- "dummy",
- Collections.singletonList(new Node(0, "host1", 1000)),
- Collections.singletonList(new PartitionInfo(topic, 0, null, null, null)),
- Collections.emptySet(),
- Collections.emptySet());
Metadata metadata = mock(Metadata.class);
// Return empty cluster 4 times and cluster from then on
- when(metadata.fetch()).thenReturn(emptyCluster, emptyCluster, emptyCluster, emptyCluster, cluster);
+ when(metadata.fetch()).thenReturn(emptyCluster, emptyCluster, emptyCluster, emptyCluster, onePartitionCluster);
KafkaProducer<String, String> producer = new KafkaProducer<String, String>(configs, new StringSerializer(),
new StringSerializer(), metadata, new MockClient(Time.SYSTEM, metadata), null, Time.SYSTEM) {
@@ -340,91 +350,127 @@ public class KafkaProducerTest {
}
@Test
- public void testMetadataFetchOnStaleMetadata() throws Exception {
+ public void testMetadataTimeoutWithMissingTopic() throws Exception {
Map<String, Object> configs = new HashMap<>();
configs.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9999");
- String topic = "topic";
- ProducerRecord<String, String> initialRecord = new ProducerRecord<>(topic, "value");
+ configs.put(ProducerConfig.MAX_BLOCK_MS_CONFIG, 60000);
+
// Create a record with a partition higher than the initial (outdated) partition range
- ProducerRecord<String, String> extendedRecord = new ProducerRecord<>(topic, 2, null, "value");
- Collection<Node> nodes = Collections.singletonList(new Node(0, "host1", 1000));
- final Cluster emptyCluster = new Cluster(null, nodes,
- Collections.emptySet(),
- Collections.emptySet(),
- Collections.emptySet());
- final Cluster initialCluster = new Cluster(
- "dummy",
- Collections.singletonList(new Node(0, "host1", 1000)),
- Collections.singletonList(new PartitionInfo(topic, 0, null, null, null)),
- Collections.emptySet(),
- Collections.emptySet());
- final Cluster extendedCluster = new Cluster(
- "dummy",
- Collections.singletonList(new Node(0, "host1", 1000)),
- Arrays.asList(
- new PartitionInfo(topic, 0, null, null, null),
- new PartitionInfo(topic, 1, null, null, null),
- new PartitionInfo(topic, 2, null, null, null)),
- Collections.emptySet(),
- Collections.emptySet());
+ ProducerRecord<String, String> record = new ProducerRecord<>(topic, 2, null, "value");
Metadata metadata = mock(Metadata.class);
+ MockTime mockTime = new MockTime();
AtomicInteger invocationCount = new AtomicInteger(0);
-
- // Return empty cluster 4 times, initialCluster 5 times and extendedCluster after that
when(metadata.fetch()).then(invocation -> {
invocationCount.incrementAndGet();
- if (invocationCount.get() > 9)
- return extendedCluster;
- else if (invocationCount.get() > 4)
- return initialCluster;
+ if (invocationCount.get() == 5) {
+ mockTime.setCurrentTimeMs(mockTime.milliseconds() + 70000);
+ }
+
return emptyCluster;
});
KafkaProducer<String, String> producer = new KafkaProducer<String, String>(configs, new StringSerializer(),
- new StringSerializer(), metadata, new MockClient(Time.SYSTEM, metadata), null, Time.SYSTEM) {
+ new StringSerializer(), metadata, new MockClient(Time.SYSTEM, metadata), null, mockTime) {
@Override
Sender newSender(LogContext logContext, KafkaClient kafkaClient, Metadata metadata) {
// give Sender its own Metadata instance so that we can isolate Metadata calls from KafkaProducer
return super.newSender(logContext, kafkaClient, new Metadata(0, 100_000, true));
}
};
- producer.send(initialRecord);
- // One request update for each empty cluster returned
+ // Four request updates where the topic isn't present, at which point the timeout expires and a
+ // TimeoutException is thrown
+ Future future = producer.send(record);
verify(metadata, times(4)).requestUpdate();
verify(metadata, times(4)).awaitUpdate(anyInt(), anyLong());
verify(metadata, times(5)).fetch();
-
- // Should not request update if metadata is available and records are within range
- producer.send(initialRecord);
- verify(metadata, times(4)).requestUpdate();
- verify(metadata, times(4)).awaitUpdate(anyInt(), anyLong());
- verify(metadata, times(6)).fetch();
-
- // One request update followed by exception if topic metadata is available but metadata response still returns
- // the same partition size (either because metadata are still stale at the broker too or because
- // there weren't any partitions added in the first place)
try {
- producer.send(extendedRecord);
- fail("Expected KafkaException to be raised");
- } catch (KafkaException e) {
- // expected
+ future.get();
+ } catch (ExecutionException e) {
+ assertTrue(e.getCause() instanceof TimeoutException);
+ } finally {
+ producer.close(0, TimeUnit.MILLISECONDS);
}
- verify(metadata, times(5)).requestUpdate();
- verify(metadata, times(5)).awaitUpdate(anyInt(), anyLong());
- verify(metadata, times(8)).fetch();
+ }
+ @Test
+ public void testMetadataWithPartitionOutOfRange() throws Exception {
+ Map<String, Object> configs = new HashMap<>();
+ configs.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9999");
+ configs.put(ProducerConfig.MAX_BLOCK_MS_CONFIG, 60000);
+
+ // Create a record with a partition higher than the initial (outdated) partition range
+ ProducerRecord<String, String> record = new ProducerRecord<>(topic, 2, null, "value");
+ Metadata metadata = mock(Metadata.class);
+
+ MockTime mockTime = new MockTime();
+
+ when(metadata.fetch()).thenReturn(onePartitionCluster, onePartitionCluster, threePartitionCluster);
+
+ KafkaProducer<String, String> producer = new KafkaProducer<String, String>(configs, new StringSerializer(),
+ new StringSerializer(), metadata, new MockClient(Time.SYSTEM, metadata), null, mockTime) {
+ @Override
+ Sender newSender(LogContext logContext, KafkaClient kafkaClient, Metadata metadata) {
+ // give Sender its own Metadata instance so that we can isolate Metadata calls from KafkaProducer
+ return super.newSender(logContext, kafkaClient, new Metadata(0, 100_000, true));
+ }
+ };
// One request update if metadata is available but outdated for the given record
- producer.send(extendedRecord);
- verify(metadata, times(6)).requestUpdate();
- verify(metadata, times(6)).awaitUpdate(anyInt(), anyLong());
- verify(metadata, times(10)).fetch();
+ producer.send(record);
+ verify(metadata, times(2)).requestUpdate();
+ verify(metadata, times(2)).awaitUpdate(anyInt(), anyLong());
+ verify(metadata, times(3)).fetch();
producer.close(0, TimeUnit.MILLISECONDS);
}
@Test
+ public void testMetadataTimeoutWithPartitionOutOfRange() throws Exception {
+ Map<String, Object> configs = new HashMap<>();
+ configs.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9999");
+ configs.put(ProducerConfig.MAX_BLOCK_MS_CONFIG, 60000);
+
+ // Create a record with a partition higher than the initial (outdated) partition range
+ ProducerRecord<String, String> record = new ProducerRecord<>(topic, 2, null, "value");
+ Metadata metadata = mock(Metadata.class);
+
+ MockTime mockTime = new MockTime();
+ AtomicInteger invocationCount = new AtomicInteger(0);
+ when(metadata.fetch()).then(invocation -> {
+ invocationCount.incrementAndGet();
+ if (invocationCount.get() == 5) {
+ mockTime.setCurrentTimeMs(mockTime.milliseconds() + 70000);
+ }
+
+ return onePartitionCluster;
+ });
+
+ KafkaProducer<String, String> producer = new KafkaProducer<String, String>(configs, new StringSerializer(),
+ new StringSerializer(), metadata, new MockClient(Time.SYSTEM, metadata), null, mockTime) {
+ @Override
+ Sender newSender(LogContext logContext, KafkaClient kafkaClient, Metadata metadata) {
+ // give Sender its own Metadata instance so that we can isolate Metadata calls from KafkaProducer
+ return super.newSender(logContext, kafkaClient, new Metadata(0, 100_000, true));
+ }
+ };
+
+ // Four request updates where the requested partition is out of range, at which point the timeout expires
+ // and a TimeoutException is thrown
+ Future future = producer.send(record);
+ verify(metadata, times(4)).requestUpdate();
+ verify(metadata, times(4)).awaitUpdate(anyInt(), anyLong());
+ verify(metadata, times(5)).fetch();
+ try {
+ future.get();
+ } catch (ExecutionException e) {
+ assertTrue(e.getCause() instanceof TimeoutException);
+ } finally {
+ producer.close(0, TimeUnit.MILLISECONDS);
+ }
+ }
+
+ @Test
public void testTopicRefreshInMetadata() throws InterruptedException {
Map<String, Object> configs = new HashMap<>();
configs.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9999");
diff --git a/core/src/test/scala/integration/kafka/api/BaseProducerSendTest.scala b/core/src/test/scala/integration/kafka/api/BaseProducerSendTest.scala
index 7a2394e..8799bac 100644
--- a/core/src/test/scala/integration/kafka/api/BaseProducerSendTest.scala
+++ b/core/src/test/scala/integration/kafka/api/BaseProducerSendTest.scala
@@ -21,20 +21,21 @@ import java.nio.charset.StandardCharsets
import java.util.Properties
import java.util.concurrent.TimeUnit
-import collection.JavaConverters._
import kafka.integration.KafkaServerTestHarness
import kafka.log.LogConfig
import kafka.server.KafkaConfig
import kafka.utils.TestUtils
import org.apache.kafka.clients.consumer.{ConsumerRecord, KafkaConsumer}
import org.apache.kafka.clients.producer._
-import org.apache.kafka.common.{KafkaException, TopicPartition}
+import org.apache.kafka.common.errors.TimeoutException
import org.apache.kafka.common.record.TimestampType
import org.apache.kafka.common.security.auth.SecurityProtocol
+import org.apache.kafka.common.{KafkaException, TopicPartition}
import org.junit.Assert._
import org.junit.{After, Before, Test}
import scala.collection.mutable.{ArrayBuffer, Buffer}
+import scala.collection.JavaConverters._
import scala.concurrent.ExecutionException
abstract class BaseProducerSendTest extends KafkaServerTestHarness {
@@ -71,13 +72,15 @@ abstract class BaseProducerSendTest extends KafkaServerTestHarness {
protected def createProducer(brokerList: String,
lingerMs: Int = 0,
batchSize: Int = 16384,
- compressionType: String = "none"): KafkaProducer[Array[Byte],Array[Byte]] = {
+ compressionType: String = "none",
+ maxBlockMs: Long = 60 * 1000L): KafkaProducer[Array[Byte],Array[Byte]] = {
val producer = TestUtils.createProducer(brokerList,
compressionType = compressionType,
securityProtocol = securityProtocol,
trustStoreFile = trustStoreFile,
saslProperties = clientSaslProperties,
- lingerMs = lingerMs)
+ lingerMs = lingerMs,
+ maxBlockMs = maxBlockMs)
registerProducer(producer)
}
@@ -353,7 +356,7 @@ abstract class BaseProducerSendTest extends KafkaServerTestHarness {
*/
@Test
def testSendBeforeAndAfterPartitionExpansion() {
- val producer = createProducer(brokerList)
+ val producer = createProducer(brokerList, maxBlockMs = 5 * 1000L)
// create topic
createTopic(topic, 1, 2)
@@ -373,10 +376,13 @@ abstract class BaseProducerSendTest extends KafkaServerTestHarness {
// Trying to send a record to a partition beyond topic's partition range before adding the partition should fail.
val partition1 = 1
try {
- producer.send(new ProducerRecord(topic, partition1, null, "value".getBytes(StandardCharsets.UTF_8)))
+ producer.send(new ProducerRecord(topic, partition1, null, "value".getBytes(StandardCharsets.UTF_8))).get()
fail("Should not allow sending a record to a partition not present in the metadata")
} catch {
- case _: KafkaException => // this is ok
+ case e: ExecutionException => e.getCause match {
+ case _: TimeoutException => // this is ok
+ case ex => throw new Exception("Sending to a partition not present in the metadata should result in a TimeoutException", ex)
+ }
}
val existingAssignment = zkClient.getReplicaAssignmentForTopics(Set(topic)).map {
diff --git a/core/src/test/scala/integration/kafka/api/ProducerFailureHandlingTest.scala b/core/src/test/scala/integration/kafka/api/ProducerFailureHandlingTest.scala
index b7d3ecb..17d68d1 100644
--- a/core/src/test/scala/integration/kafka/api/ProducerFailureHandlingTest.scala
+++ b/core/src/test/scala/integration/kafka/api/ProducerFailureHandlingTest.scala
@@ -17,7 +17,7 @@
package kafka.api
-import java.util.concurrent.{ExecutionException, TimeoutException}
+import java.util.concurrent.ExecutionException
import java.util.Properties
import kafka.integration.KafkaServerTestHarness
@@ -25,7 +25,6 @@ import kafka.log.LogConfig
import kafka.server.KafkaConfig
import kafka.utils.TestUtils
import org.apache.kafka.clients.producer._
-import org.apache.kafka.common.KafkaException
import org.apache.kafka.common.errors._
import org.apache.kafka.common.internals.Topic
import org.apache.kafka.common.record.{DefaultRecord, DefaultRecordBatch}
@@ -182,8 +181,8 @@ class ProducerFailureHandlingTest extends KafkaServerTestHarness {
}
/**
- * Send with invalid partition id should throw KafkaException when partition is higher than the upper bound of
- * partitions.
+ * Send with invalid partition id should return ExecutionException caused by TimeoutException
+ * when partition is higher than the upper bound of partitions.
*/
@Test
def testInvalidPartition() {
@@ -192,8 +191,11 @@ class ProducerFailureHandlingTest extends KafkaServerTestHarness {
// create a record with incorrect partition id (higher than the number of partitions), send should fail
val higherRecord = new ProducerRecord(topic1, 1, "key".getBytes, "value".getBytes)
- intercept[KafkaException] {
- producer1.send(higherRecord)
+ intercept[ExecutionException] {
+ producer1.send(higherRecord).get
+ }.getCause match {
+ case _: TimeoutException => // this is ok
+ case ex => throw new Exception("Sending to a partition not present in the metadata should result in a TimeoutException", ex)
}
}