Kafka 2257

checkstyle/import-control.xml

@@ -92,7 +92,7 @@
 		<subpackage name="tools">
 			<allow pkg="org.apache.kafka.clients.producer" />
 			<allow pkg="org.apache.kafka.clients.consumer" />
-            <allow pkg="com.fasterxml.jackson.core" />
+            <allow pkg="com.fasterxml.jackson" />
             <allow pkg="net.sourceforge.argparse4j" />
 		</subpackage>
 	</subpackage>

clients/src/main/java/org/apache/kafka/clients/ClientRequest.java

@@ -23,24 +23,43 @@ public final class ClientRequest {
     private final boolean expectResponse;
     private final RequestSend request;
     private final RequestCompletionHandler callback;
+    private final boolean isInitiatedByNetworkClient;
 
     /**
      * @param createdMs The unix timestamp in milliseconds for the time at which this request was created.
      * @param expectResponse Should we expect a response message or is this request complete once it is sent?
      * @param request The request
      * @param callback A callback to execute when the response has been received (or null if no callback is necessary)
      */
-    public ClientRequest(long createdMs, boolean expectResponse, RequestSend request, RequestCompletionHandler callback) {
+    public ClientRequest(long createdMs, boolean expectResponse, RequestSend request,
+                         RequestCompletionHandler callback) {
+        this(createdMs, expectResponse, request, callback, false);
+    }
+
+    /**
+     * @param createdMs The unix timestamp in milliseconds for the time at which this request was created.
+     * @param expectResponse Should we expect a response message or is this request complete once it is sent?
+     * @param request The request
+     * @param callback A callback to execute when the response has been received (or null if no callback is necessary)
+     * @param isInitiatedByNetworkClient Is request initiated by network client, if yes, its
+     *                                   response will be consumed by network client
+     */
+    public ClientRequest(long createdMs, boolean expectResponse, RequestSend request,
+                         RequestCompletionHandler callback, boolean isInitiatedByNetworkClient) {
         this.createdMs = createdMs;
         this.callback = callback;
         this.request = request;
         this.expectResponse = expectResponse;
+        this.isInitiatedByNetworkClient = isInitiatedByNetworkClient;
     }
 
     @Override
     public String toString() {
-        return "ClientRequest(expectResponse=" + expectResponse + ", callback=" + callback + ", request=" + request
-                + ")";
+        return "ClientRequest(expectResponse=" + expectResponse +
+            ", callback=" + callback +
+            ", request=" + request +
+            (isInitiatedByNetworkClient ? ", isInitiatedByNetworkClient" : "") +
+            ")";
     }
 
     public boolean expectResponse() {
@@ -63,4 +82,8 @@ public long createdTime() {
         return createdMs;
     }
 
+    public boolean isInitiatedByNetworkClient() {
+        return isInitiatedByNetworkClient;
+    }
+
 }

clients/src/main/java/org/apache/kafka/clients/NetworkClient.java

@@ -378,7 +378,7 @@ private void handleCompletedReceives(List<ClientResponse> responses, long now) {
             short apiKey = req.request().header().apiKey();
             Struct body = (Struct) ProtoUtils.currentResponseSchema(apiKey).read(receive.payload());
             correlate(req.request().header(), header);
-            if (apiKey == ApiKeys.METADATA.id) {
+            if (apiKey == ApiKeys.METADATA.id && req.isInitiatedByNetworkClient()) {
                 handleMetadataResponse(req.request().header(), body, now);
             } else {
                 // need to add body/header to response here
@@ -454,7 +454,7 @@ private void correlate(RequestHeader requestHeader, ResponseHeader responseHeade
     private ClientRequest metadataRequest(long now, String node, Set<String> topics) {
         MetadataRequest metadata = new MetadataRequest(new ArrayList<String>(topics));
         RequestSend send = new RequestSend(node, nextRequestHeader(ApiKeys.METADATA), metadata.toStruct());
-        return new ClientRequest(now, true, send, null);
+        return new ClientRequest(now, true, send, null, true);
     }
 
     /**

clients/src/main/java/org/apache/kafka/clients/consumer/Consumer.java

@@ -113,6 +113,11 @@ public interface Consumer<K, V> extends Closeable {
      */
     public List<PartitionInfo> partitionsFor(String topic);
 
+    /**
+     * @see KafkaConsumer#listTopics()
+     */
+    public Map<String, List<PartitionInfo>> listTopics();
+
     /**
      * @see KafkaConsumer#close()
      */

clients/src/main/java/org/apache/kafka/clients/consumer/KafkaConsumer.java

@@ -1024,6 +1024,22 @@ public List<PartitionInfo> partitionsFor(String topic) {
         }
     }
 
+    /**
+     * Get metadata about partitions for all topics. This method will issue a remote call to the
+     * server.
+     *
+     * @return The map of topics and its partitions
+     */
+    @Override
+    public Map<String, List<PartitionInfo>> listTopics() {
+        acquire();
+        try {
+            return fetcher.getAllTopics(requestTimeoutMs);
+        } finally {
+            release();
+        }
+    }
+
     @Override
     public void close() {
         acquire();

clients/src/main/java/org/apache/kafka/clients/consumer/MockConsumer.java

@@ -177,6 +177,12 @@ public synchronized List<PartitionInfo> partitionsFor(String topic) {
             return parts;
     }
 
+    @Override
+    public Map<String, List<PartitionInfo>> listTopics() {
+        ensureNotClosed();
+        return partitions;
+    }
+
     public synchronized void updatePartitions(String topic, List<PartitionInfo> partitions) {
         ensureNotClosed();
         this.partitions.put(topic, partitions);

clients/src/main/java/org/apache/kafka/clients/consumer/internals/Fetcher.java

@@ -39,6 +39,8 @@
 import org.apache.kafka.common.requests.FetchResponse;
 import org.apache.kafka.common.requests.ListOffsetRequest;
 import org.apache.kafka.common.requests.ListOffsetResponse;
+import org.apache.kafka.common.requests.MetadataRequest;
+import org.apache.kafka.common.requests.MetadataResponse;
 import org.apache.kafka.common.serialization.Deserializer;
 import org.apache.kafka.common.utils.Time;
 import org.apache.kafka.common.utils.Utils;
@@ -160,6 +162,48 @@ public void updateFetchPositions(Set<TopicPartition> partitions) {
         }
     }
 
+
+
+    /**
+     * Get metadata for all topics present in Kafka cluster
+     *
+     * @param timeout time for which getting all topics is attempted
+     * @return The map of topics and its partitions
+     */
+    public Map<String, List<PartitionInfo>> getAllTopics(long timeout) {
+        final HashMap<String, List<PartitionInfo>> topicsPartitionInfos = new HashMap<>();
+        long startTime = time.milliseconds();
+
+        while (time.milliseconds() - startTime < timeout) {
+            final Node node = client.leastLoadedNode();
+            if (node != null) {
+                MetadataRequest metadataRequest = new MetadataRequest(Collections.<String>emptyList());
+                final RequestFuture<ClientResponse> requestFuture =
+                    client.send(node, ApiKeys.METADATA, metadataRequest);
+
+                client.poll(requestFuture);
+
+                if (requestFuture.succeeded()) {
+                    MetadataResponse response =
+                        new MetadataResponse(requestFuture.value().responseBody());
+
+                    for (String topic : response.cluster().topics())
+                        topicsPartitionInfos.put(
+                            topic, response.cluster().availablePartitionsForTopic(topic));
+
+                    return topicsPartitionInfos;
+                }
+
+                if (!requestFuture.isRetriable())
+                    throw requestFuture.exception();
+            }
+
+            Utils.sleep(retryBackoffMs);
+        }
+
+        return topicsPartitionInfos;
+    }
+
     /**
      * Reset offsets for the given partition using the offset reset strategy.
      *

clients/src/main/java/org/apache/kafka/clients/consumer/internals/SubscriptionState.java

@@ -15,6 +15,7 @@
 import org.apache.kafka.clients.consumer.OffsetResetStrategy;
 import org.apache.kafka.common.TopicPartition;
 
+import java.util.ArrayList;
 import java.util.HashMap;
 import java.util.HashSet;
 import java.util.List;
@@ -83,7 +84,8 @@ public void unsubscribe(String topic) {
             throw new IllegalStateException("Topic " + topic + " was never subscribed to.");
         this.subscribedTopics.remove(topic);
         this.needsPartitionAssignment = true;
-        for (TopicPartition tp: assignedPartitions())
+        final List<TopicPartition> existingAssignedPartitions = new ArrayList<>(assignedPartitions());
+        for (TopicPartition tp: existingAssignedPartitions)
             if (topic.equals(tp.topic()))
                 clearPartition(tp);
     }

clients/src/test/java/org/apache/kafka/clients/MetadataTest.java

@@ -12,7 +12,7 @@
  */
 package org.apache.kafka.clients;
 
-import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicReference;
 
 import org.apache.kafka.common.Cluster;
 import org.apache.kafka.common.errors.TimeoutException;
@@ -27,11 +27,11 @@ public class MetadataTest {
     private long refreshBackoffMs = 100;
     private long metadataExpireMs = 1000;
     private Metadata metadata = new Metadata(refreshBackoffMs, metadataExpireMs);
-    private AtomicBoolean backgroundError = new AtomicBoolean(false);
+    private AtomicReference<String> backgroundError = new AtomicReference<String>();
 
     @After
     public void tearDown() {
-        assertFalse(backgroundError.get());
+        assertNull("Exception in background thread : " + backgroundError.get(), backgroundError.get());
     }
 
     @Test
@@ -48,7 +48,15 @@ public void testMetadata() throws Exception {
         Thread t2 = asyncFetch(topic);
         assertTrue("Awaiting update", t1.isAlive());
         assertTrue("Awaiting update", t2.isAlive());
-        metadata.update(TestUtils.singletonCluster(topic, 1), time);
+        // Perform metadata update when an update is requested on the async fetch thread
+        // This simulates the metadata update sequence in KafkaProducer
+        while (t1.isAlive() || t2.isAlive()) {
+            if (metadata.timeToNextUpdate(time) == 0) {
+                metadata.update(TestUtils.singletonCluster(topic, 1), time);
+                time += refreshBackoffMs;
+            }
+            Thread.sleep(1);
+        }
         t1.join();
         t2.join();
         assertFalse("No update needed.", metadata.timeToNextUpdate(time) == 0);
@@ -106,7 +114,7 @@ public void run() {
                     try {
                         metadata.awaitUpdate(metadata.requestUpdate(), refreshBackoffMs);
                     } catch (Exception e) {
-                        backgroundError.set(true);
+                        backgroundError.set(e.toString());
                     }
                 }
             }

clients/src/test/java/org/apache/kafka/clients/consumer/internals/FetcherTest.java

@@ -22,13 +22,15 @@
 import org.apache.kafka.clients.consumer.OffsetResetStrategy;
 import org.apache.kafka.common.Cluster;
 import org.apache.kafka.common.Node;
+import org.apache.kafka.common.PartitionInfo;
 import org.apache.kafka.common.TopicPartition;
 import org.apache.kafka.common.metrics.Metrics;
 import org.apache.kafka.common.protocol.Errors;
 import org.apache.kafka.common.protocol.types.Struct;
 import org.apache.kafka.common.record.CompressionType;
 import org.apache.kafka.common.record.MemoryRecords;
 import org.apache.kafka.common.requests.FetchResponse;
+import org.apache.kafka.common.requests.MetadataResponse;
 import org.apache.kafka.common.serialization.ByteArrayDeserializer;
 import org.apache.kafka.common.utils.MockTime;
 import org.apache.kafka.test.TestUtils;
@@ -180,6 +182,17 @@ public void testFetchOutOfRange() {
         assertEquals(null, subscriptions.consumed(tp));
     }
 
+    @Test
+    public void testGetAllTopics() throws InterruptedException {
+        // sending response before request, as getAllTopics is a blocking call
+        client.prepareResponse(
+            new MetadataResponse(cluster, Collections.<String, Errors>emptyMap()).toStruct());
+
+        Map<String, List<PartitionInfo>> allTopics = fetcher.getAllTopics(5000L);
+
+        assertEquals(cluster.topics().size(), allTopics.size());
+    }
+
     private Struct fetchResponse(ByteBuffer buffer, short error, long hw) {
         FetchResponse response = new FetchResponse(Collections.singletonMap(tp, new FetchResponse.PartitionData(error, hw, buffer)));
         return response.toStruct();

clients/src/test/java/org/apache/kafka/clients/consumer/internals/SubscriptionStateTest.java

@@ -73,6 +73,27 @@ public void topicSubscription() {
         assertAllPositions(tp0, null);
         assertEquals(Collections.singleton(tp1), state.assignedPartitions());
     }
+
+    @Test
+    public void topicUnsubscription() {
+        final String topic = "test";
+        state.subscribe(topic);
+        assertEquals(1, state.subscribedTopics().size());
+        assertTrue(state.assignedPartitions().isEmpty());
+        assertTrue(state.partitionsAutoAssigned());
+        state.changePartitionAssignment(asList(tp0));
+        state.committed(tp0, 1);
+        state.fetched(tp0, 1);
+        state.consumed(tp0, 1);
+        assertAllPositions(tp0, 1L);
+        state.changePartitionAssignment(asList(tp1));
+        assertAllPositions(tp0, null);
+        assertEquals(Collections.singleton(tp1), state.assignedPartitions());
+
+        state.unsubscribe(topic);
+        assertEquals(0, state.subscribedTopics().size());
+        assertTrue(state.assignedPartitions().isEmpty());
+    }
 
     @Test(expected = IllegalArgumentException.class)
     public void cantChangeFetchPositionForNonAssignedPartition() {

config/server.properties

@@ -121,3 +121,5 @@ zookeeper.connect=localhost:2181
 
 # Timeout in ms for connecting to zookeeper
 zookeeper.connection.timeout.ms=6000
+
+auto.create.topics.enable=true

core/src/main/scala/kafka/javaapi/consumer/ConsumerConnector.java

@@ -75,6 +75,12 @@ public interface ConsumerConnector {
    */
   public void commitOffsets(Map<TopicAndPartition, OffsetAndMetadata> offsetsToCommit, boolean retryOnFailure);
 
+  /**
+   * Wire in a consumer rebalance listener to be executed when consumer rebalance occurs.
+   * @param listener The consumer rebalance listener to wire in
+   */
+  public void setConsumerRebalanceListener(ConsumerRebalanceListener listener);
+
   /**
    *  Shut down the connector
    */

core/src/main/scala/kafka/tools/ConsumerOffsetChecker.scala

@@ -107,6 +107,8 @@ object ConsumerOffsetChecker extends Logging {
   }
 
   def main(args: Array[String]) {
+    warn("WARNING: ConsumerOffsetChecker is deprecated and will be dropped in releases following 0.9.0. Use ConsumerGroupCommand instead.")
+
     val parser = new OptionParser()
 
     val zkConnectOpt = parser.accepts("zookeeper", "ZooKeeper connect string.").

core/src/test/scala/integration/kafka/api/ConsumerTest.scala

@@ -186,6 +186,24 @@ class ConsumerTest extends IntegrationTestHarness with Logging {
     assertNull(this.consumers(0).partitionsFor("non-exist-topic"))
   }
 
+  def testListTopics() {
+    val numParts = 2
+    val topic1: String = "part-test-topic-1"
+    val topic2: String = "part-test-topic-2"
+    val topic3: String = "part-test-topic-3"
+    TestUtils.createTopic(this.zkClient, topic1, numParts, 1, this.servers)
+    TestUtils.createTopic(this.zkClient, topic2, numParts, 1, this.servers)
+    TestUtils.createTopic(this.zkClient, topic3, numParts, 1, this.servers)
+
+    val topics = this.consumers.head.listTopics()
+    assertNotNull(topics)
+    assertEquals(5, topics.size())
+    assertEquals(5, topics.keySet().size())
+    assertEquals(2, topics.get(topic1).length)
+    assertEquals(2, topics.get(topic2).length)
+    assertEquals(2, topics.get(topic3).length)
+  }
+
   def testPartitionReassignmentCallback() {
     val callback = new TestConsumerReassignmentCallback()
     this.consumerConfig.setProperty(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, "100"); // timeout quickly to avoid slow test
@@ -217,6 +235,25 @@ class ConsumerTest extends IntegrationTestHarness with Logging {
     consumer0.close()
   }
 
+  def testUnsubscribeTopic() {
+    val callback = new TestConsumerReassignmentCallback()
+    this.consumerConfig.setProperty(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, "100"); // timeout quickly to avoid slow test
+    val consumer0 = new KafkaConsumer(this.consumerConfig, callback, new ByteArrayDeserializer(), new ByteArrayDeserializer())
+
+    try {
+      consumer0.subscribe(topic)
+
+      // the initial subscription should cause a callback execution
+      while (callback.callsToAssigned == 0)
+        consumer0.poll(50)
+
+      consumer0.unsubscribe(topic)
+      assertEquals(0, consumer0.subscriptions.size())
+    } finally {
+      consumer0.close()
+    }
+  }
+
   private class TestConsumerReassignmentCallback extends ConsumerRebalanceCallback {
     var callsToAssigned = 0
     var callsToRevoked = 0