[improve][broker]Part-3 of PIP-433: always allow replicator to register a new compatible schema

[poorbarcode]

Motivation

This is part-3 of PIP-433, see the goal 1 in PIP-433

Modifications

always allow the replicator to register a new compatible schema, users can turn off this function

Documentation

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

[Denovo1998]

@poorbarcode @codelipenghui
Seems like there are some issues?

When geo-replicator creates a remote producer in the broker, it indeed carries the replicator's producer name and uses Schema.AUTO_PRODUCE_BYTES().

But the key point of AUTO_PRODUCE_BYTES is: when establishing a producer connection, it doesn't send the "specific schema of this message" all at once. The actual specific schema registration happens later during the message sending process, when encountering a new schema, it then triggers GET_OR_CREATE_SCHEMA. This call chain looks like this locally:

1. When replicator acts as a producer to establish a connection, the ProducerImpl.connectionOpened(...) method sends the PRODUCER command, which includes the producerName. But later, when it actually needs to register a specific schema, it follows the path tryRegisterSchema(...) -> getOrCreateSchemaAsync(...).

2. The method finally calls Commands' newGetOrCreateSchema(requestId, topic, schemaInfo).

3. This command builder only wrote requestId/topic/schema, without writing producerName.

4. So the CommandGetOrCreateSchema sent from the sender to the broker always has the field producerName absent.