[client] Cleanup consumer on multitopic subscribe failure

[addisonj]

Motivation

Currently, when a multi-topic subscribe fails (via a set of topics or a
regex) we can leave consumers connected, as the multitopic consumer
doesn't close any of the topics.

This means we rely on the client to call closeAsync, otherwise, the
consumer is left in partially open state.

Modifications

This fix changes that, and ensures we call close in the case of an
exception

Verifying this change

This change added tests that ensure that the expected methods are called in the event of subscribe failure

[zymap]

Failing tests record:

integration sql:

Error:  Tests run: 6, Failures: 2, Errors: 0, Skipped: 2, Time elapsed: 349.226 s <<< FAILURE! - in TestSuite
Error:  pulsar-sql-test-suite(org.apache.pulsar.tests.integration.presto.TestPrestoQueryTieredStorage)  Time elapsed: 29.664 s  <<< FAILURE!
org.apache.pulsar.tests.integration.docker.ContainerExecException: /bin/bash -c /pulsar/bin/pulsar sql --execute 'select * from pulsar."presto/ts"."stocks_ts_nons_yjxio" order by entryid;' failed on 3fa284a519c77b9a14f64469696bdbac3f79644c4c0d96913462773fefabc3da with error code 1
	at org.apache.pulsar.tests.integration.utils.DockerUtils$2.onComplete(DockerUtils.java:259)
	at org.testcontainers.shaded.com.github.dockerjava.core.exec.AbstrAsyncDockerCmdExec$1.onComplete(AbstrAsyncDockerCmdExec.java:51)
	at org.testcontainers.shaded.com.github.dockerjava.core.DefaultInvocationBuilder.lambda$executeAndStream$1(DefaultInvocationBuilder.java:276)
	at java.lang.Thread.run(Thread.java:748)

Error:  pulsar-sql-test-suite(org.apache.pulsar.tests.integration.presto.TestPrestoQueryTieredStorage)  Time elapsed: 36.393 s  <<< FAILURE!
org.apache.pulsar.tests.integration.docker.ContainerExecException: /bin/bash -c /pulsar/bin/pulsar sql --execute 'select * from pulsar."presto/ts"."stocks_ts_ns_xbnhw" order by entryid;' failed on 3fa284a519c77b9a14f64469696bdbac3f79644c4c0d96913462773fefabc3da with error code 1
	at org.apache.pulsar.tests.integration.utils.DockerUtils$2.onComplete(DockerUtils.java:259)
	at org.testcontainers.shaded.com.github.dockerjava.core.exec.AbstrAsyncDockerCmdExec$1.onComplete(AbstrAsyncDockerCmdExec.java:51)
	at org.testcontainers.shaded.com.github.dockerjava.core.DefaultInvocationBuilder.lambda$executeAndStream$1(DefaultInvocationBuilder.java:276)
	at java.lang.Thread.run(Thread.java:748)

[INFO] 
[INFO] Results:
[INFO] 
Error:  Failures: 
Error:  org.apache.pulsar.tests.integration.presto.TestPrestoQueryTieredStorage.pulsar-sql-test-suite(org.apache.pulsar.tests.integration.presto.TestPrestoQueryTieredStorage)
[INFO]   Run 1: PASS
Error:    Run 2: TestPrestoQueryTieredStorage.testQueryTieredStorage1 » ContainerExec /bin/bash...
[INFO]   Run 3: PASS
Error:    Run 4: TestPrestoQueryTieredStorage.testQueryTieredStorage2 » ContainerExec /bin/bash...

broker group 1:

Error:  Tests run: 43, Failures: 1, Errors: 0, Skipped: 2, Time elapsed: 1,222.846 s <<< FAILURE! - in org.apache.pulsar.broker.admin.TopicPoliciesTest
Error:  setup(org.apache.pulsar.broker.admin.TopicPoliciesTest)  Time elapsed: 942.58 s  <<< FAILURE!
java.lang.OutOfMemoryError: Java heap space

[INFO] 
[INFO] Results:
[INFO] 
Error:  Failures: 
Error:    TopicPoliciesTest.setup ? OutOfMemory Java heap space
[INFO] 
Error:  Tests run: 507, Failures: 1, Errors: 0, Skipped: 1

[zymap]

/pulsarbot run-failure-checks