[fix][test] SchemaMap in AutoConsumeSchema has been reused

[Denovo1998]

Fixes #22470

Motivation

Related to #17887. After executing the test method once does not clean up the pulsarClient, causing the AutoConsumeSchema to be reused. When the schemaMap in the first AutoConsumeSchema is set to AvroSchema.

pulsar/pulsar-broker/src/test/java/org/apache/pulsar/client/api/SimpleProducerConsumerTest.java

Lines 149 to 169 in 982d94d

	@AfterMethod(alwaysRun = true)
	public void rest() throws Exception {
	pulsar.getConfiguration().setForceDeleteTenantAllowed(true);
	pulsar.getConfiguration().setForceDeleteNamespaceAllowed(true);
	for (String tenant : admin.tenants().getTenants()) {
	for (String namespace : admin.namespaces().getNamespaces(tenant)) {
	deleteNamespaceWithRetry(namespace, true);
	}
	admin.tenants().deleteTenant(tenant, true);
	}
	for (String cluster : admin.clusters().getClusters()) {
	admin.clusters().deleteCluster(cluster);
	}
	pulsar.getConfiguration().setForceDeleteTenantAllowed(false);
	pulsar.getConfiguration().setForceDeleteNamespaceAllowed(false);
	super.producerBaseSetup();
	}

pulsar/pulsar-broker/src/test/java/org/apache/pulsar/client/api/SimpleProducerConsumerTest.java

Lines 202 to 206 in 982d94d

	@AfterClass(alwaysRun = true)
	@Override
	protected void cleanup() throws Exception {
	super.internalCleanup();
	}

The second execution of the test method will also use the AvroSchema instead of the JsonSchema. Because the schema will not be fetched again.

pulsar/pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/AutoConsumeSchema.java

Lines 130 to 136 in af1360f

	@Override
	public GenericRecord decode(ByteBuffer buffer, byte[] schemaVersion) {
	SchemaVersion sv = getSchemaVersion(schemaVersion);
	fetchSchemaIfNeeded(sv);
	ensureSchemaInitialized(sv);
	return adapt(schemaMap.get(sv).decode(buffer, schemaVersion), schemaVersion);
	}

pulsar/pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/AutoConsumeSchema.java

Lines 328 to 338 in af1360f

	/**
	* It may happen that the schema is not loaded but we need it, for instance in order to call getSchemaInfo()
	* We cannot call this method in getSchemaInfo, because getSchemaInfo is called in many
	* places and we will introduce lots of deadlocks.
	*/
	public void fetchSchemaIfNeeded(SchemaVersion schemaVersion) throws SchemaSerializationException {
	if (schemaVersion == null) {
	schemaVersion = BytesSchemaVersion.of(new byte[0]);
	}
	if (!schemaMap.containsKey(schemaVersion)) {
	if (schemaInfoProvider == null) {

Modifications

1. Closes the pulsarClient after throwing an exception.
2. Determine whether the pulsarClient is null before starting, and create a new one if it is null.

Verifying this change

• Make sure that the change passes the CI checks.

(Please pick either of the following options)

This change is a trivial rework / code cleanup without any test coverage.

(or)

This change is already covered by existing tests, such as (please describe tests).

(or)

This change added tests and can be verified as follows:

(example:)

• Added integration tests for end-to-end deployment with large payloads (10MB)
• Extended integration test for recovery after broker failure

Does this pull request potentially affect one of the following parts:

If the box was checked, please highlight the changes

• Dependencies (add or upgrade a dependency)
• The public API
• The schema
• The default values of configurations
• The threading model
• The binary protocol
• The REST endpoints
• The admin CLI options
• The metrics
• Anything that affects deployment

Documentation

• doc
• doc-required
• doc-not-needed
• doc-complete

Matching PR in forked repository

PR in forked repository: Denovo1998#6