smjn commented on code in PR #17270:
URL: https://github.com/apache/kafka/pull/17270#discussion_r1792209784
##########
core/src/main/scala/kafka/server/BrokerServer.scala:
##########
@@ -345,7 +347,29 @@ class BrokerServer(
tokenManager = new DelegationTokenManager(config, tokenCache, time)
tokenManager.startup()
- shareCoordinator = createShareCoordinator()
+ /* get persister instance */
+ persister = if
(config.shareGroupConfig.shareGroupPersisterClassName.nonEmpty) {
+ val klass =
Utils.loadClass(config.shareGroupConfig.shareGroupPersisterClassName,
classOf[Object]).asInstanceOf[Class[Persister]]
+ klass.getDeclaredMethod("getInstance")
Review Comment:
After some discussion we have concluded that the persister will be Factory
pluggable. The persister class name will be an internal config which will not
be exposed to the end user.
The implementations will be internal and thus can be non-general like the
current one.
I have however moved away from the singleton structure approach as that was
causing issues in the integ tests but the behavior of the persister will still
be singleton as it will be created only once in the BrokerServer and then
passed onto other entities which need it.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
