Repository: kafka
Updated Branches:
  refs/heads/trunk 65f6a7964 -> 1a67739c2


KAFKA-4254; Update producer's metadata before failing on non-existent partitions

Author: Konstantine Karantasis <konstant...@confluent.io>

Reviewers: Ismael Juma <ism...@juma.me.uk>, Jason Gustafson <ja...@confluent.io>

Closes #1995 from 
kkonstantine/KAFKA-4254-Update-producers-metadata-before-failing-on-non-existent-partition


Project: http://git-wip-us.apache.org/repos/asf/kafka/repo
Commit: http://git-wip-us.apache.org/repos/asf/kafka/commit/1a67739c
Tree: http://git-wip-us.apache.org/repos/asf/kafka/tree/1a67739c
Diff: http://git-wip-us.apache.org/repos/asf/kafka/diff/1a67739c

Branch: refs/heads/trunk
Commit: 1a67739c2f3d0418a8f2d8f7b15d5af2ed3a324a
Parents: 65f6a79
Author: Konstantine Karantasis <konstant...@confluent.io>
Authored: Thu Oct 13 20:17:43 2016 -0700
Committer: Jason Gustafson <ja...@confluent.io>
Committed: Thu Oct 13 20:17:57 2016 -0700

----------------------------------------------------------------------
 .../kafka/clients/producer/KafkaProducer.java   | 82 +++++++++--------
 .../kafka/clients/producer/ProducerRecord.java  |  8 +-
 .../clients/producer/KafkaProducerTest.java     | 92 ++++++++++++++++++--
 .../clients/producer/ProducerRecordTest.java    | 40 +++++++--
 .../kafka/api/BaseProducerSendTest.scala        | 90 +++++++++++++++----
 .../kafka/api/ProducerFailureHandlingTest.scala | 26 +++---
 6 files changed, 255 insertions(+), 83 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/kafka/blob/1a67739c/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 3efc7b5..3632384 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
@@ -12,28 +12,16 @@
  */
 package org.apache.kafka.clients.producer;
 
-import java.net.InetSocketAddress;
-import java.util.Collections;
-import java.util.LinkedHashMap;
-import java.util.List;
-import java.util.Map;
-import java.util.Properties;
-import java.util.concurrent.ExecutionException;
-import java.util.concurrent.Future;
-import java.util.concurrent.TimeUnit;
-import java.util.concurrent.atomic.AtomicInteger;
-import java.util.concurrent.atomic.AtomicReference;
-
 import org.apache.kafka.clients.ClientUtils;
 import org.apache.kafka.clients.Metadata;
 import org.apache.kafka.clients.NetworkClient;
+import org.apache.kafka.clients.producer.internals.ProducerInterceptors;
 import org.apache.kafka.clients.producer.internals.RecordAccumulator;
 import org.apache.kafka.clients.producer.internals.Sender;
-import org.apache.kafka.clients.producer.internals.ProducerInterceptors;
 import org.apache.kafka.common.Cluster;
-import org.apache.kafka.common.internals.ClusterResourceListeners;
 import org.apache.kafka.common.KafkaException;
 import org.apache.kafka.common.Metric;
+import org.apache.kafka.common.MetricName;
 import org.apache.kafka.common.PartitionInfo;
 import org.apache.kafka.common.TopicPartition;
 import org.apache.kafka.common.config.ConfigException;
@@ -43,14 +31,14 @@ import 
org.apache.kafka.common.errors.RecordTooLargeException;
 import org.apache.kafka.common.errors.SerializationException;
 import org.apache.kafka.common.errors.TimeoutException;
 import org.apache.kafka.common.errors.TopicAuthorizationException;
+import org.apache.kafka.common.internals.ClusterResourceListeners;
 import org.apache.kafka.common.metrics.JmxReporter;
 import org.apache.kafka.common.metrics.MetricConfig;
-import org.apache.kafka.common.MetricName;
 import org.apache.kafka.common.metrics.Metrics;
 import org.apache.kafka.common.metrics.MetricsReporter;
 import org.apache.kafka.common.metrics.Sensor;
-import org.apache.kafka.common.network.Selector;
 import org.apache.kafka.common.network.ChannelBuilder;
+import org.apache.kafka.common.network.Selector;
 import org.apache.kafka.common.record.CompressionType;
 import org.apache.kafka.common.record.Record;
 import org.apache.kafka.common.record.Records;
@@ -62,6 +50,17 @@ import org.apache.kafka.common.utils.Time;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import java.net.InetSocketAddress;
+import java.util.Collections;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicReference;
 
 /**
  * A Kafka client that publishes records to the Kafka cluster.
@@ -426,7 +425,8 @@ public class KafkaProducer<K, V> implements Producer<K, V> {
      *
      * @throws InterruptException If the thread is interrupted while blocked
      * @throws SerializationException If the key or value are not valid 
objects given the configured serializers
-     * @throws TimeoutException if the time taken for fetching metadata or 
allocating memory for the record has surpassed <code>max.block.ms</code>.
+     * @throws TimeoutException If the time taken for fetching metadata or 
allocating memory for the record has surpassed <code>max.block.ms</code>.
+     * @throws KafkaException If a Kafka related error occurs that does not 
belong to the public API exceptions.
      *
      */
     @Override
@@ -443,8 +443,8 @@ public class KafkaProducer<K, V> implements Producer<K, V> {
         TopicPartition tp = null;
         try {
             // first make sure the metadata for the topic is available
-            ClusterAndWaitTime clusterAndWaitTime = 
waitOnMetadata(record.topic(), this.maxBlockTimeMs);
-            long remainingWaitMs = Math.max(0, this.maxBlockTimeMs - 
clusterAndWaitTime.waitedOnMetadataMs);
+            ClusterAndWaitTime clusterAndWaitTime = 
waitOnMetadata(record.topic(), record.partition(), maxBlockTimeMs);
+            long remainingWaitMs = Math.max(0, maxBlockTimeMs - 
clusterAndWaitTime.waitedOnMetadataMs);
             Cluster cluster = clusterAndWaitTime.cluster;
             byte[] serializedKey;
             try {
@@ -515,20 +515,28 @@ public class KafkaProducer<K, V> implements Producer<K, 
V> {
     /**
      * Wait for cluster metadata including partitions for the given topic to 
be available.
      * @param topic The topic we want metadata for
+     * @param partition A specific partition expected to exist in metadata, or 
null if there's no preference
      * @param maxWaitMs The maximum time in ms for waiting on the metadata
      * @return The cluster containing topic metadata and the amount of time we 
waited in ms
      */
-    private ClusterAndWaitTime waitOnMetadata(String topic, long maxWaitMs) 
throws InterruptedException {
+    private ClusterAndWaitTime waitOnMetadata(String topic, Integer partition, 
long maxWaitMs) throws InterruptedException {
         // add topic to metadata topic list if it is not there already and 
reset expiry
-        this.metadata.add(topic);
+        metadata.add(topic);
         Cluster cluster = metadata.fetch();
-        if (cluster.partitionsForTopic(topic) != null)
+        Integer partitionsCount = cluster.partitionCountForTopic(topic);
+        // Return cached metadata if we have it, and if the record's partition 
is either undefined
+        // or within the known partition range
+        if (partitionsCount != null && (partition == null || partition < 
partitionsCount))
             return new ClusterAndWaitTime(cluster, 0);
 
         long begin = time.milliseconds();
         long remainingWaitMs = maxWaitMs;
-        long elapsed = 0;
-        while (cluster.partitionsForTopic(topic) == null) {
+        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
+        // is stale and the number of partitions for this topic has increased 
in the meantime.
+        do {
             log.trace("Requesting metadata update for topic {}.", topic);
             int version = metadata.requestUpdate();
             sender.wakeup();
@@ -545,7 +553,14 @@ public class KafkaProducer<K, V> implements Producer<K, V> 
{
             if (cluster.unauthorizedTopics().contains(topic))
                 throw new TopicAuthorizationException(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));
         }
+
         return new ClusterAndWaitTime(cluster, elapsed);
     }
 
@@ -611,13 +626,11 @@ public class KafkaProducer<K, V> implements Producer<K, 
V> {
      */
     @Override
     public List<PartitionInfo> partitionsFor(String topic) {
-        Cluster cluster;
         try {
-            cluster = waitOnMetadata(topic, this.maxBlockTimeMs).cluster;
+            return waitOnMetadata(topic, null, 
maxBlockTimeMs).cluster.partitionsForTopic(topic);
         } catch (InterruptedException e) {
             throw new InterruptException(e);
         }
-        return cluster.partitionsForTopic(topic);
     }
 
     /**
@@ -733,17 +746,10 @@ public class KafkaProducer<K, V> implements Producer<K, 
V> {
      */
     private int partition(ProducerRecord<K, V> record, byte[] serializedKey, 
byte[] serializedValue, Cluster cluster) {
         Integer partition = record.partition();
-        if (partition != null) {
-            List<PartitionInfo> partitions = 
cluster.partitionsForTopic(record.topic());
-            int lastPartition = partitions.size() - 1;
-            // they have given us a partition, use it
-            if (partition < 0 || partition > lastPartition) {
-                throw new IllegalArgumentException(String.format("Invalid 
partition given with record: %d is not in the range [0...%d].", partition, 
lastPartition));
-            }
-            return partition;
-        }
-        return this.partitioner.partition(record.topic(), record.key(), 
serializedKey, record.value(), serializedValue,
-            cluster);
+        return partition != null ?
+                partition :
+                partitioner.partition(
+                        record.topic(), record.key(), serializedKey, 
record.value(), serializedValue, cluster);
     }
 
     private static class ClusterAndWaitTime {

http://git-wip-us.apache.org/repos/asf/kafka/blob/1a67739c/clients/src/main/java/org/apache/kafka/clients/producer/ProducerRecord.java
----------------------------------------------------------------------
diff --git 
a/clients/src/main/java/org/apache/kafka/clients/producer/ProducerRecord.java 
b/clients/src/main/java/org/apache/kafka/clients/producer/ProducerRecord.java
index 85b4d8d..536067e 100644
--- 
a/clients/src/main/java/org/apache/kafka/clients/producer/ProducerRecord.java
+++ 
b/clients/src/main/java/org/apache/kafka/clients/producer/ProducerRecord.java
@@ -55,9 +55,13 @@ public final class ProducerRecord<K, V> {
      */
     public ProducerRecord(String topic, Integer partition, Long timestamp, K 
key, V value) {
         if (topic == null)
-            throw new IllegalArgumentException("Topic cannot be null");
+            throw new IllegalArgumentException("Topic cannot be null.");
         if (timestamp != null && timestamp < 0)
-            throw new IllegalArgumentException("Invalid timestamp " + 
timestamp);
+            throw new IllegalArgumentException(
+                    String.format("Invalid timestamp: %d. Timestamp should 
always be non-negative or null.", timestamp));
+        if (partition != null && partition < 0)
+            throw new IllegalArgumentException(
+                    String.format("Invalid partition: %d. Partition number 
should always be non-negative or null.", partition));
         this.topic = topic;
         this.partition = partition;
         this.key = key;

http://git-wip-us.apache.org/repos/asf/kafka/blob/1a67739c/clients/src/test/java/org/apache/kafka/clients/producer/KafkaProducerTest.java
----------------------------------------------------------------------
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 333fbfd..c82b18b 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
@@ -6,7 +6,7 @@
  * (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
+ * 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,
@@ -16,14 +16,14 @@
  */
 package org.apache.kafka.clients.producer;
 
+import org.apache.kafka.clients.CommonClientConfigs;
+import org.apache.kafka.clients.Metadata;
 import org.apache.kafka.common.Cluster;
 import org.apache.kafka.common.KafkaException;
 import org.apache.kafka.common.Node;
 import org.apache.kafka.common.PartitionInfo;
 import org.apache.kafka.common.network.Selectable;
 import org.apache.kafka.common.serialization.ByteArraySerializer;
-import org.apache.kafka.clients.CommonClientConfigs;
-import org.apache.kafka.clients.Metadata;
 import org.apache.kafka.common.serialization.StringSerializer;
 import org.apache.kafka.test.MockMetricsReporter;
 import org.apache.kafka.test.MockProducerInterceptor;
@@ -38,12 +38,14 @@ import 
org.powermock.core.classloader.annotations.PowerMockIgnore;
 import org.powermock.core.classloader.annotations.PrepareOnlyThisForTest;
 import org.powermock.modules.junit4.PowerMockRunner;
 
-import java.util.Properties;
-import java.util.Map;
 import java.util.Arrays;
 import java.util.Collection;
 import java.util.Collections;
 import java.util.HashMap;
+import java.util.Map;
+import java.util.Properties;
+
+import static org.junit.Assert.fail;
 
 @RunWith(PowerMockRunner.class)
 @PowerMockIgnore("javax.management.*")
@@ -67,7 +69,7 @@ public class KafkaProducerTest {
             Assert.assertEquals("Failed to construct kafka producer", 
e.getMessage());
             return;
         }
-        Assert.fail("should have caught an exception and returned");
+        fail("should have caught an exception and returned");
     }
 
     @Test
@@ -191,4 +193,82 @@ public class KafkaProducerTest {
         producer.partitionsFor(topic);
         PowerMock.verify(metadata);
     }
+
+    @PrepareOnlyThisForTest(Metadata.class)
+    @Test
+    public void testMetadataFetchOnStaleMetadata() throws Exception {
+        Properties props = new Properties();
+        props.setProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, 
"localhost:9999");
+        KafkaProducer<String, String> producer = new KafkaProducer<>(props, 
new StringSerializer(), new StringSerializer());
+        Metadata metadata = PowerMock.createNiceMock(Metadata.class);
+        MemberModifier.field(KafkaProducer.class, "metadata").set(producer, 
metadata);
+
+        String topic = "topic";
+        ProducerRecord<String, String> initialRecord = new 
ProducerRecord<>(topic, "value");
+        // 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.<PartitionInfo>emptySet(),
+                Collections.<String>emptySet(),
+                Collections.<String>emptySet());
+        final Cluster initialCluster = new Cluster(
+                "dummy",
+                Collections.singletonList(new Node(0, "host1", 1000)),
+                Arrays.asList(new PartitionInfo(topic, 0, null, null, null)),
+                Collections.<String>emptySet(),
+                Collections.<String>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.<String>emptySet(),
+                Collections.<String>emptySet());
+
+        // Expect exactly one fetch for each attempt to refresh while topic 
metadata is not available
+        final int refreshAttempts = 5;
+        
EasyMock.expect(metadata.fetch()).andReturn(emptyCluster).times(refreshAttempts 
- 1);
+        EasyMock.expect(metadata.fetch()).andReturn(initialCluster).once();
+        EasyMock.expect(metadata.fetch()).andThrow(new 
IllegalStateException("Unexpected call to metadata.fetch()")).anyTimes();
+        PowerMock.replay(metadata);
+        producer.send(initialRecord);
+        PowerMock.verify(metadata);
+
+        // Expect exactly one fetch if topic metadata is available and records 
are still within range
+        PowerMock.reset(metadata);
+        EasyMock.expect(metadata.fetch()).andReturn(initialCluster).once();
+        EasyMock.expect(metadata.fetch()).andThrow(new 
IllegalStateException("Unexpected call to metadata.fetch()")).anyTimes();
+        PowerMock.replay(metadata);
+        producer.send(initialRecord, null);
+        PowerMock.verify(metadata);
+
+        // Expect exactly two fetches 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).
+        PowerMock.reset(metadata);
+        EasyMock.expect(metadata.fetch()).andReturn(initialCluster).once();
+        EasyMock.expect(metadata.fetch()).andReturn(initialCluster).once();
+        EasyMock.expect(metadata.fetch()).andThrow(new 
IllegalStateException("Unexpected call to metadata.fetch()")).anyTimes();
+        PowerMock.replay(metadata);
+        try {
+            producer.send(extendedRecord, null);
+            fail("Expected KafkaException to be raised");
+        } catch (KafkaException e) {
+            // expected
+        }
+        PowerMock.verify(metadata);
+
+        // Expect exactly two fetches if topic metadata is available but 
outdated for the given record
+        PowerMock.reset(metadata);
+        EasyMock.expect(metadata.fetch()).andReturn(initialCluster).once();
+        EasyMock.expect(metadata.fetch()).andReturn(extendedCluster).once();
+        EasyMock.expect(metadata.fetch()).andThrow(new 
IllegalStateException("Unexpected call to metadata.fetch()")).anyTimes();
+        PowerMock.replay(metadata);
+        producer.send(extendedRecord, null);
+        PowerMock.verify(metadata);
+    }
+
 }

http://git-wip-us.apache.org/repos/asf/kafka/blob/1a67739c/clients/src/test/java/org/apache/kafka/clients/producer/ProducerRecordTest.java
----------------------------------------------------------------------
diff --git 
a/clients/src/test/java/org/apache/kafka/clients/producer/ProducerRecordTest.java
 
b/clients/src/test/java/org/apache/kafka/clients/producer/ProducerRecordTest.java
index f3db098..a844bb0 100644
--- 
a/clients/src/test/java/org/apache/kafka/clients/producer/ProducerRecordTest.java
+++ 
b/clients/src/test/java/org/apache/kafka/clients/producer/ProducerRecordTest.java
@@ -20,33 +20,59 @@ import org.junit.Test;
 
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.fail;
 
 public class ProducerRecordTest {
 
     @Test
     public void testEqualsAndHashCode() {
-        ProducerRecord<String, Integer> producerRecord = new 
ProducerRecord<String, Integer>("test", 1 , "key", 1);
+        ProducerRecord<String, Integer> producerRecord = new 
ProducerRecord<>("test", 1 , "key", 1);
         assertEquals(producerRecord, producerRecord);
         assertEquals(producerRecord.hashCode(), producerRecord.hashCode());
 
-        ProducerRecord<String, Integer> equalRecord = new 
ProducerRecord<String, Integer>("test", 1 , "key", 1);
+        ProducerRecord<String, Integer> equalRecord = new 
ProducerRecord<>("test", 1 , "key", 1);
         assertEquals(producerRecord, equalRecord);
         assertEquals(producerRecord.hashCode(), equalRecord.hashCode());
 
-        ProducerRecord<String, Integer> topicMisMatch = new 
ProducerRecord<String, Integer>("test-1", 1 , "key", 1);
+        ProducerRecord<String, Integer> topicMisMatch = new 
ProducerRecord<>("test-1", 1 , "key", 1);
         assertFalse(producerRecord.equals(topicMisMatch));
 
-        ProducerRecord<String, Integer> partitionMismatch = new 
ProducerRecord<String, Integer>("test", 2 , "key", 1);
+        ProducerRecord<String, Integer> partitionMismatch = new 
ProducerRecord<>("test", 2 , "key", 1);
         assertFalse(producerRecord.equals(partitionMismatch));
 
-        ProducerRecord<String, Integer> keyMisMatch = new 
ProducerRecord<String, Integer>("test", 1 , "key-1", 1);
+        ProducerRecord<String, Integer> keyMisMatch = new 
ProducerRecord<>("test", 1 , "key-1", 1);
         assertFalse(producerRecord.equals(keyMisMatch));
 
-        ProducerRecord<String, Integer> valueMisMatch = new 
ProducerRecord<String, Integer>("test", 1 , "key", 2);
+        ProducerRecord<String, Integer> valueMisMatch = new 
ProducerRecord<>("test", 1 , "key", 2);
         assertFalse(producerRecord.equals(valueMisMatch));
 
-        ProducerRecord<String, Integer> nullFieldsRecord = new 
ProducerRecord<String, Integer>("topic", null, null, null, null);
+        ProducerRecord<String, Integer> nullFieldsRecord = new 
ProducerRecord<>("topic", null, null, null, null);
         assertEquals(nullFieldsRecord, nullFieldsRecord);
         assertEquals(nullFieldsRecord.hashCode(), nullFieldsRecord.hashCode());
     }
+
+    @Test
+    public void testInvalidRecords() {
+        try {
+            new ProducerRecord<>(null, 0, "key", 1);
+            fail("Expected IllegalArgumentException to be raised because topic 
is null");
+        } catch (IllegalArgumentException e) {
+            //expected
+        }
+
+        try {
+            new ProducerRecord<>("test", 0, -1L, "key", 1);
+            fail("Expected IllegalArgumentException to be raised because of 
negative timestamp");
+        } catch (IllegalArgumentException e) {
+            //expected
+        }
+
+        try {
+            new ProducerRecord<>("test", -1, "key", 1);
+            fail("Expected IllegalArgumentException to be raised because of 
negative partition");
+        } catch (IllegalArgumentException e) {
+            //expected
+        }
+    }
+
 }

http://git-wip-us.apache.org/repos/asf/kafka/blob/1a67739c/core/src/test/scala/integration/kafka/api/BaseProducerSendTest.scala
----------------------------------------------------------------------
diff --git 
a/core/src/test/scala/integration/kafka/api/BaseProducerSendTest.scala 
b/core/src/test/scala/integration/kafka/api/BaseProducerSendTest.scala
index 5bb0438..816f36a 100644
--- a/core/src/test/scala/integration/kafka/api/BaseProducerSendTest.scala
+++ b/core/src/test/scala/integration/kafka/api/BaseProducerSendTest.scala
@@ -20,13 +20,16 @@ package kafka.api
 import java.util.Properties
 import java.util.concurrent.TimeUnit
 
+import kafka.admin.AdminUtils
 import kafka.consumer.SimpleConsumer
 import kafka.integration.KafkaServerTestHarness
 import kafka.log.LogConfig
 import kafka.message.Message
 import kafka.server.KafkaConfig
 import kafka.utils.TestUtils
+import kafka.utils.TestUtils._
 import org.apache.kafka.clients.producer._
+import org.apache.kafka.common.KafkaException
 import org.apache.kafka.common.record.TimestampType
 import org.junit.Assert._
 import org.junit.{After, Before, Test}
@@ -262,26 +265,19 @@ abstract class BaseProducerSendTest extends 
KafkaServerTestHarness {
       val leaders = TestUtils.createTopic(zkUtils, topic, 2, 2, servers)
       val partition = 1
 
-      // make sure leaders exist
-      val leader1 = leaders(partition)
-      assertTrue("Leader for topic \"topic\" partition 1 should exist", 
leader1.isDefined)
-
       val now = System.currentTimeMillis()
-      val responses =
-        for (i <- 1 to numRecords)
-        yield producer.send(new ProducerRecord[Array[Byte], 
Array[Byte]](topic, partition, now, null, ("value" + i).getBytes))
-      val futures = responses.toList
-      futures.foreach(_.get)
-      for (future <- futures)
-        assertTrue("Request should have completed", future.isDone)
+      val futures = (1 to numRecords).map { i =>
+        producer.send(new ProducerRecord(topic, partition, now, null, ("value" 
+ i).getBytes))
+      }.map(_.get(30, TimeUnit.SECONDS))
 
       // make sure all of them end up in the same partition with increasing 
offset values
-      for ((future, offset) <- futures zip (0 until numRecords)) {
-        assertEquals(offset.toLong, future.get.offset)
-        assertEquals(topic, future.get.topic)
-        assertEquals(partition, future.get.partition)
+      for ((recordMetadata, offset) <- futures zip (0 until numRecords)) {
+        assertEquals(offset.toLong, recordMetadata.offset)
+        assertEquals(topic, recordMetadata.topic)
+        assertEquals(partition, recordMetadata.partition)
       }
 
+      val leader1 = leaders(partition)
       // make sure the fetched messages also respect the partitioning and 
ordering
       val fetchResponse1 = if (leader1.get == configs.head.brokerId) {
         consumer1.fetch(new FetchRequestBuilder().addFetch(topic, partition, 
0, Int.MaxValue).build())
@@ -302,6 +298,70 @@ abstract class BaseProducerSendTest extends 
KafkaServerTestHarness {
   }
 
   /**
+    * Checks partitioning behavior before and after partitions are added
+    *
+    * Producer will attempt to send messages to the partition specified in 
each record, and should
+    * succeed as long as the partition is included in the metadata.
+    */
+  @Test
+  def testSendBeforeAndAfterPartitionExpansion() {
+    val producer = createProducer(brokerList)
+
+    // create topic
+    TestUtils.createTopic(zkUtils, topic, 1, 2, servers)
+    val partition0 = 0
+
+    var futures0 = (1 to numRecords).map { i =>
+      producer.send(new ProducerRecord(topic, partition0, null, ("value" + 
i).getBytes))
+    }.map(_.get(30, TimeUnit.SECONDS))
+
+    // make sure all of them end up in the same partition with increasing 
offset values
+    for ((recordMetadata, offset) <- futures0 zip (0 until numRecords)) {
+      assertEquals(offset.toLong, recordMetadata.offset)
+      assertEquals(topic, recordMetadata.topic)
+      assertEquals(partition0, recordMetadata.partition)
+    }
+
+    // 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))
+      fail("Should not allow sending a record to a partition not present in 
the metadata")
+    } catch {
+      case ke: KafkaException => // this is ok
+      case e: Throwable => fail("Only expecting KafkaException", e)
+    }
+
+    AdminUtils.addPartitions(zkUtils, topic, 2)
+    // read metadata from a broker and verify the new topic partitions exist
+    TestUtils.waitUntilMetadataIsPropagated(servers, topic, 0)
+    TestUtils.waitUntilMetadataIsPropagated(servers, topic, 1)
+
+    // send records to the newly added partition after confirming that 
metadata have been updated.
+    val futures1 = (1 to numRecords).map { i =>
+      producer.send(new ProducerRecord(topic, partition1, null, ("value" + 
i).getBytes))
+    }.map(_.get(30, TimeUnit.SECONDS))
+
+    // make sure all of them end up in the same partition with increasing 
offset values
+    for ((recordMetadata, offset) <- futures1 zip (0 until numRecords)) {
+      assertEquals(offset.toLong, recordMetadata.offset)
+      assertEquals(topic, recordMetadata.topic)
+      assertEquals(partition1, recordMetadata.partition)
+    }
+
+    futures0 = (1 to numRecords).map { i =>
+      producer.send(new ProducerRecord(topic, partition0, null, ("value" + 
i).getBytes))
+    }.map(_.get(30, TimeUnit.SECONDS))
+
+    // make sure all of them end up in the same partition with increasing 
offset values starting where previous
+    for ((recordMetadata, offset) <- futures0 zip (numRecords until 2 * 
numRecords)) {
+      assertEquals(offset.toLong, recordMetadata.offset)
+      assertEquals(topic, recordMetadata.topic)
+      assertEquals(partition0, recordMetadata.partition)
+    }
+  }
+
+  /**
    * Test that flush immediately sends all accumulated requests.
    */
   @Test

http://git-wip-us.apache.org/repos/asf/kafka/blob/1a67739c/core/src/test/scala/integration/kafka/api/ProducerFailureHandlingTest.scala
----------------------------------------------------------------------
diff --git 
a/core/src/test/scala/integration/kafka/api/ProducerFailureHandlingTest.scala 
b/core/src/test/scala/integration/kafka/api/ProducerFailureHandlingTest.scala
index 457a909..5385bbe 100644
--- 
a/core/src/test/scala/integration/kafka/api/ProducerFailureHandlingTest.scala
+++ 
b/core/src/test/scala/integration/kafka/api/ProducerFailureHandlingTest.scala
@@ -24,9 +24,10 @@ import kafka.common.Topic
 import kafka.integration.KafkaServerTestHarness
 import kafka.log.LogConfig
 import kafka.server.KafkaConfig
-import kafka.utils.{TestUtils}
+import kafka.utils.TestUtils
 import org.apache.kafka.clients.producer._
-import org.apache.kafka.common.errors.{InvalidTopicException, 
NotEnoughReplicasAfterAppendException, NotEnoughReplicasException}
+import org.apache.kafka.common.KafkaException
+import org.apache.kafka.common.errors._
 import org.junit.Assert._
 import org.junit.{After, Before, Test}
 
@@ -175,23 +176,18 @@ class ProducerFailureHandlingTest extends 
KafkaServerTestHarness {
   }
 
   /**
-   *  The send call with invalid partition id should throw KafkaException 
caused by IllegalArgumentException
-   */
+    * Send with invalid partition id should throw KafkaException when 
partition is higher than the
+    * upper bound of partitions and IllegalArgumentException when partition is 
negative
+    */
   @Test
   def testInvalidPartition() {
-    // create topic
+    // create topic with a single partition
     TestUtils.createTopic(zkUtils, topic1, 1, numServers, servers)
 
-    // create a record with incorrect partition id, send should fail
-    val record = new ProducerRecord[Array[Byte],Array[Byte]](topic1, new 
Integer(1), "key".getBytes, "value".getBytes)
-    intercept[IllegalArgumentException] {
-      producer1.send(record)
-    }
-    intercept[IllegalArgumentException] {
-      producer2.send(record)
-    }
-    intercept[IllegalArgumentException] {
-      producer3.send(record)
+    // create a record with incorrect partition id (higher than the number of 
partitions), send should fail
+    val higherRecord = new ProducerRecord[Array[Byte], Array[Byte]](topic1, 1, 
"key".getBytes, "value".getBytes)
+    intercept[KafkaException] {
+      producer1.send(higherRecord)
     }
   }
 

Reply via email to