Add the field to avoid creating missing topic.

writer.go

@@ -185,6 +185,9 @@ type Writer struct {
 	// If nil, DefaultTransport is used.
 	Transport RoundTripper
 
+	// AllowAutoTopicCreation notifies writer to create topic is missing.
+	AllowAutoTopicCreation bool
+
 	// Manages the current set of partition-topic writers.
 	group   sync.WaitGroup
 	mutex   sync.Mutex
@@ -733,7 +736,7 @@ func (w *Writer) partitions(ctx context.Context, topic string) (int, error) {
 	// caching recent results (the kafka.Transport types does).
 	r, err := client.transport().RoundTrip(ctx, client.Addr, &metadataAPI.Request{
 		TopicNames:             []string{topic},
-		AllowAutoTopicCreation: true,
+		AllowAutoTopicCreation: w.AllowAutoTopicCreation,
 	})
 	if err != nil {
 		return 0, err
@@ -941,7 +944,7 @@ func newBatchQueue(initialSize int) batchQueue {
 	bq := batchQueue{
 		queue: make([]*writeBatch, 0, initialSize),
 		mutex: &sync.Mutex{},
-		cond: &sync.Cond{},
+		cond:  &sync.Cond{},
 	}
 
 	bq.cond.L = bq.mutex

writer_test.go

@@ -160,6 +160,10 @@ func TestWriter(t *testing.T) {
 			scenario: "writing a message to a non-existant topic creates the topic",
 			function: testWriterAutoCreateTopic,
 		},
+		{
+			scenario: "terminates on an attempts to write a message to a non-existant topic",
+			function: testWriterTerminateMissingTopic,
+		},
 	}
 
 	for _, test := range tests {
@@ -737,6 +741,30 @@ func testWriterAutoCreateTopic(t *testing.T) {
 	}
 }
 
+func testWriterTerminateMissingTopic(t *testing.T) {
+	topic := makeTopic()
+
+	transport := &Transport{}
+	defer transport.CloseIdleConnections()
+
+	writer := &Writer{
+		Addr:                   TCP("localhost:9092"),
+		Topic:                  topic,
+		Balancer:               &RoundRobin{},
+		RequiredAcks:           RequireNone,
+		AllowAutoTopicCreation: false,
+		Transport:              transport,
+	}
+	defer writer.Close()
+
+	msg := Message{Value: []byte("FooBar")}
+
+	if err := writer.WriteMessages(context.Background(), msg); err == nil {
+		t.Fatal("Kafka error [3] UNKNOWN_TOPIC_OR_PARTITION is expected")
+		return
+	}
+}
+
 type staticBalancer struct {
 	partition int
 }