When topic is deleted forced, don't to tigger reconnect logic

[wolfstudy]

Expected behavior

When topic is deleted forced, under normal circumstances, we should force the producer to be deleted. What the client feels is that the producer has been closed, instead of triggering the logic of reconnection.

Actual behavior

When topic is deleted forced, the request triggers the logic of reconnection.

Steps to reproduce

1. Start standalone (e.g: 2.8.0) and set allowAutoTopicCreation=false in standalone.conf
2. Run the demo of producer:

func main() {
	client, err := pulsar.NewClient(pulsar.ClientOptions{
		URL: "pulsar://localhost:6650",
	})

	if err != nil {
		log.Fatal(err)
	}

	defer client.Close()

	producer, err := client.CreateProducer(pulsar.ProducerOptions{
		Topic: "topic-1",
	})
	if err != nil {
		log.Fatal(err)
	}

	defer producer.Close()

	ctx := context.Background()

	for i := 0; i < 1000000; i++ {
		if msgId, err := producer.Send(ctx, &pulsar.ProducerMessage{
			Payload: []byte(fmt.Sprintf("hello-%d", i)),
		}); err != nil {
			log.Fatal(err)
		} else {
			time.Sleep(1 * time.Second)
			log.Println("Published message: ", msgId)
		}
	}
}

3. Run the cmd to force delete topic-1

 bin/pulsar-admin topics delete topic-1 -f

4. Observing the behavior of this producer, we will find that we have triggered the reconnection logic.

The same code logic, we can see the following error message in java, this is the behavior we expect:

System configuration

Pulsar version: x.y

[wolfstudy]

And the java demo as follows:

public static void main(String[] args) throws PulsarClientException {
        PulsarClient client = PulsarClient.builder()
                .serviceUrl("pulsar://localhost:6650")
                .build();

        final String topic = "test-topic-002";
        final int msgCount = 100000;

        Producer<byte[]> producer = client.newProducer()
                .topic(topic)
                .create();
        for (int i = 0; i < msgCount; i++) {
            String value = "new msg No." + i;
            MessageId msgId = producer.newMessage()
                    .value(value.getBytes())
                    .send();
            try {
                Thread.sleep(1000);
            } catch (InterruptedException exception) {
                exception.printStackTrace();
            }
            System.out.println("===>>> produce msg id:" + new String(msgId.toString()) + ", value:" + value);
        }
        producer.flush();
}