[fix][consumer] Use resolved partition topic for single-partition zero queue consumer

pulsar/consumer_impl.go

@@ -284,7 +284,7 @@ func newInternalConsumer(client *client, options ConsumerOptions, topic string,
 	}
 
 	if len(partitions) == 1 && options.EnableZeroQueueConsumer {
-		return newZeroConsumer(client, options, topic, messageCh, dlq, rlq, disableForceTopicCreation)
+		return newZeroConsumer(client, options, partitions[0], messageCh, dlq, rlq, disableForceTopicCreation)
 	}
 
 	consumer := &consumer{

pulsar/consumer_zero_queue_test.go

@@ -654,6 +654,36 @@ func TestSpecifiedPartitionZeroQueueConsumer(t *testing.T) {
 	assert.Nil(t, err)
 }
 
+func TestZeroQueueConsumerUsesOnlyPartitionForSinglePartitionTopic(t *testing.T) {
+	client, err := NewClient(ClientOptions{
+		URL: lookupURL,
+	})
+
+	require.NoError(t, err)
+	defer client.Close()
+
+	topic := newTopicName()
+	err = createPartitionedTopic(topic, 1)
+	require.NoError(t, err)
+	topics, err := client.TopicPartitions(topic)
+	require.NoError(t, err)
+	require.Len(t, topics, 1)
+	topicName, err := internal.ParseTopicName(topics[0])
+	require.NoError(t, err)
+
+	consumer, err := client.Subscribe(ConsumerOptions{
+		Topic:                   topic,
+		SubscriptionName:        "my-sub",
+		EnableZeroQueueConsumer: true,
+	})
+	require.NoError(t, err)
+	defer consumer.Close()
+
+	zeroConsumer, ok := consumer.(*zeroQueueConsumer)
+	require.True(t, ok)
+	assert.Equal(t, topicName.Name, zeroConsumer.pc.topic)
+}
+
 func TestZeroQueueConsumerGetLastMessageIDs(t *testing.T) {
 	client, err := NewClient(ClientOptions{
 		URL: lookupURL,