KAFKA-17367: Share coordinator impl. Broker side code. [2/N]

[smjn]

• Added impl for ShareCoordinatorService and ShareCoordinatorShard
• Moved group-coordinator: GroupCoordinatorRuntimeMetrics -> coordinator-common: CoordinatorRuntimeMetricsImpl. The new impl class can be inherited and used in both group and share coordinators.
• Added tests wherever applicable
• Added plumbing in BrokerServer and BrokerMetadataPublisher to create share coordinator and start/stop it.
• Added ShareCoordinatorConfig class to house various coordinator related configs.
• Added code to create share state topic in AutoTopicCreationManager.scala

[smjn]

Thanks for the patch @smjn! This is a pretty big PR, so I've just reviewed part of it for now.

Regarding the leader epoch, it seems like we are checking the MetadataImage for some things (like if a topic exists), but using the given leader epoch in the RPC to see if the leader has changed. I wonder if we should be checking the given leader epoch against the MetadataImage.

Where are we handling topic deletions and re-creations? Maybe this is coming in a future PR?

For EA only read and write RPCs will be provided.

[smjn]

Will this logic be replaced by the init RPC once it is added? In general, we should not be updating our log based in-memory state on reads. It's probably safe in this case just based >on the nature of leader epoch, but it is definitely atypical.

No it will stay. The initialize rpc does not include any information about the leader epoch which we can store and reference. We are going by epoch definition.

[smjn]

@junrao Thanks for the review, incorporated comments.