[controller][test] Message header triggered new KME schema registrati…

[lluwm]

…on in child controller

As part of the solution to remove deployment order for KafkaMessageEnvelope (KME) value schema, for controllers, when they find an unknown KME schema from the message header, they need to (talk to the system cluster leader controller to) register the new schema into local region system store. It helps controllers to deal with unknown KME during admin message consumption.

This change enables the child controllers to register unknown KME schemas when discovering them from the message's header. It mainly leverages the existing functionalities from the ControllerClientBackedSystemSchemaInitializer class for the implementation. A new config is introduced to enable this feature and a new integration test is added to verify the correctness of the new feature.

How was this PR tested?

Added a new integration test.
Passed CI tests.

Does this PR introduce any user-facing changes?

• No. You can skip the rest of this section.
• Yes. Make sure to explain your proposed changes and call out the behavior change.

[shuhui-liu]

Thanks for making this change. I took a look at all code changes except for the new test class (will review the test after meetings).

Great work overall! Left two comments. Also, regarding the following commit message:

Otherwise, as KME is not embedded in every Kafka message, if servers restart and resume the consumption from a non-SOS message with new KME (for example, lor1 servers could consume TS sent by lva1 controller), servers can not find the right schema to deserialize it and keep failing.

I think this PR could not help with servers that encounter unknown KME after restart because this PR does not include any server change. It actually helps controllers which may encounter unknown KME after restarting and resuming admin message consumption. You may need to update the commit message to avoid confusion.

[lluwm]

Thanks for making this change. I took a look at all code changes except for the new test class (will review the test after meetings).

Great work overall! Left two comments. Also, regarding the following commit message:

Otherwise, as KME is not embedded in every Kafka message, if servers restart and resume the consumption from a non-SOS message with new KME (for example, lor1 servers could consume TS sent by lva1 controller), servers can not find the right schema to deserialize it and keep failing.

I think this PR could not help with servers that encounter unknown KME after restart because this PR does not include any server change. It actually helps controllers which may encounter unknown KME after restarting and resuming admin message consumption. You may need to update the commit message to avoid confusion.

Thanks for review, good catch, will fix the description.