Build the partitionConsumer into the topicConsumer

tools/README.md

@@ -4,7 +4,6 @@ This folder contains applications that are useful for exploration of your Kafka
 Some of these tools mirror tools that ship with Kafka, but these tools won't require installing the JVM to function.
 
 - [kafka-console-producer](./kafka-console-producer): a command line tool to produce a single message to your Kafka custer.
-- [kafka-console-partitionconsumer](./kafka-console-partitionconsumer): a command line tool to consume a single partition of a topic on your Kafka cluster.
-- [kafka-console-topicconsumer](./kafka-console-topicconsumer): a command line tool to consume all partition of a topic on your Kafka cluster.
+- [kafka-console-topicconsumer](./kafka-console-topicconsumer): a command line tool to consume partitions of a topic on your Kafka cluster.
 
 To install all tools, run `go get github.com/Shopify/sarama/tools/...`

tools/kafka-console-topicconsumer/README.md

@@ -1,6 +1,6 @@
 # kafka-console-topicconsumer
 
-A simple command line tool to consume all partitions a topic and print the messages
+A simple command line tool to consume partitions of a topic and print the messages
 on the standard output.
 
 ### Installation
@@ -21,5 +21,9 @@ on the standard output.
     kafka-console-topicconsumer -topic=test -offset=oldest
     kafka-console-topicconsumer -topic=test -offset=newest
 
+    # You can specify the partition(s) you want to consume as a comma-separated
+    # list. The default is `all`.
+    kafka-console-topicconsumer -topic=test -partitions=1,2,3
+
     # Display all command line options
     kafka-console-topicconsumer -help

tools/kafka-console-topicconsumer/kafka-console-topicconsumer.go

@@ -6,6 +6,7 @@ import (
 	"log"
 	"os"
 	"os/signal"
+	"strconv"
 	"strings"
 	"sync"
 
@@ -15,6 +16,7 @@ import (
 var (
 	brokerList = flag.String("brokers", os.Getenv("KAFKA_PEERS"), "The comma separated list of brokers in the Kafka cluster")
 	topic      = flag.String("topic", "", "REQUIRED: the topic to consume")
+	partitions = flag.String("partitions", "all", "The partitions to consume, can be 'all' or comma-separated numbers")
 	offset     = flag.String("offset", "newest", "The offset to start with. Can be `oldest`, `newest`")
 	verbose    = flag.Bool("verbose", false, "Whether to turn on sarama logging")
 	bufferSize = flag.Int("buffer-size", 256, "The buffer size of the message channel.")
@@ -52,7 +54,7 @@ func main() {
 		printErrorAndExit(69, "Failed to start consumer: %s", err)
 	}
 
-	partitions, err := c.Partitions(*topic)
+	partitionList, err := getPartitions(c)
 	if err != nil {
 		printErrorAndExit(69, "Failed to get the list of partitions: %s", err)
 	}
@@ -71,7 +73,7 @@ func main() {
 		close(closing)
 	}()
 
-	for _, partition := range partitions {
+	for _, partition := range partitionList {
 		pc, err := c.ConsumePartition(*topic, partition, initialOffset)
 		if err != nil {
 			printErrorAndExit(69, "Failed to start consumer for partition %d: %s", partition, err)
@@ -110,6 +112,24 @@ func main() {
 	}
 }
 
+func getPartitions(c sarama.Consumer) ([]int32, error) {
+	if *partitions == "all" {
+		return c.Partitions(*topic)
+	}
+
+	tmp := strings.Split(*partitions, ",")
+	var pList []int32
+	for i := range tmp {
+		val, err := strconv.ParseInt(tmp[i], 10, 32)
+		if err != nil {
+			return nil, err
+		}
+		pList = append(pList, int32(val))
+	}
+
+	return pList, nil
+}
+
 func printErrorAndExit(code int, format string, values ...interface{}) {
 	fmt.Fprintf(os.Stderr, "ERROR: %s\n", fmt.Sprintf(format, values...))
 	fmt.Fprintln(os.Stderr)