fix topic while publishing

Author: srikanth-mohan

producer.go

@@ -179,7 +179,7 @@ func (p *Producer) MultiPublish(message [][]byte) (err error) {
 // first unsuccessful attempt to publish the message. It is the responsibility
 // of the caller to retry if necessary.
 func (p *Producer) PublishTo(topic string, message []byte) (err error) {
-	return p.MultiPublishTo(p.topic, [][]byte{message})
+	return p.MultiPublishTo(topic, [][]byte{message})
 }
 
 // MultiPublishTo sends a message batch to a specific topic using the producer p, returning

producer_test.go

@@ -136,3 +136,40 @@ func TestProducerBatch(t *testing.T) {
 		})
 	}
 }
+func TestProducer_PublishTo(t *testing.T) {
+
+	tests := []struct {
+		name          string
+		topic         string
+		expectedError string
+		stopProducer  bool
+	}{
+		{name: "PublishesMessageToSpecifiedTopic", topic: "test-topic", expectedError: ""},
+		{name: "FailsWhenPublishingToEmptyTopic", topic: "", expectedError: "topic cannot be empty"},
+		{name: "FailsWhenProducerIsStopped", topic: "test-topic", expectedError: "producer is stopped", stopProducer: true},
+	}
+
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			p, _ := StartProducer(ProducerConfig{
+				Address:        "localhost:4150",
+				MaxConcurrency: 1,
+			})
+			if test.stopProducer {
+				p.Stop()
+			} else {
+				defer p.Stop()
+			}
+			message := []byte("test-message")
+			err := p.PublishTo(test.topic, message)
+
+			if err != nil && test.expectedError == "" {
+				t.Errorf("expected no error, got %v", err)
+			}
+
+			if err == nil && test.expectedError != "" {
+				t.Errorf("expected error %v, got nil", test.expectedError)
+			}
+		})
+	}
+}