You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "smjn (via GitHub)" <gi...@apache.org> on 2026/05/15 10:22:31 UTC
[PR] MINOR: Prevent this leak while DefaultStatePersister construction. [kafka]
smjn opened a new pull request, #22292:
URL: https://github.com/apache/kafka/pull/22292
* Currently `DefaultStatePersister` instantiates `PersisterStateManager` in its constructor. This implies that the `this` creation is not yet complete and initializing another component.
* To remedy this the public API constructor is replaced with static factory method.
* Tests and `BrokerServer` code has been updated to reflect the change.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] MINOR: Prevent this leak while DefaultStatePersister construction. [kafka]
Posted by "smjn (via GitHub)" <gi...@apache.org>.
smjn commented on code in PR #22292:
URL: https://github.com/apache/kafka/pull/22292#discussion_r3253435858
##########
server-common/src/main/java/org/apache/kafka/server/share/persister/DefaultStatePersister.java:
##########
@@ -45,8 +45,17 @@ public class DefaultStatePersister implements Persister {
private static final Logger log = LoggerFactory.getLogger(DefaultStatePersister.class);
- public DefaultStatePersister(PersisterStateManager stateManager) {
+ public static DefaultStatePersister instance(PersisterStateManager stateManager) {
Review Comment:
@chia7712 Thanks for the feedback.
Incorporated change. Left some default visibility methods for ease of testing.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] MINOR: Prevent this leak while DefaultStatePersister construction. [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #22292:
URL: https://github.com/apache/kafka/pull/22292#discussion_r3255598961
##########
core/src/main/scala/kafka/server/BrokerServer.scala:
##########
@@ -724,17 +725,14 @@ class BrokerServer(
private def createShareStatePersister(): Persister = {
if (config.shareGroupConfig.shareGroupPersisterClassName.nonEmpty) {
val klass = Utils.loadClass(config.shareGroupConfig.shareGroupPersisterClassName, classOf[Object]).asInstanceOf[Class[Persister]]
-
if (klass.getName.equals(classOf[DefaultStatePersister].getName)) {
- klass.getConstructor(classOf[PersisterStateManager])
- .newInstance(
- new PersisterStateManager(
- NetworkUtils.buildNetworkClient("Persister", config, metrics, Time.SYSTEM, new LogContext(s"[Persister broker=${config.brokerId}]")),
- new ShareCoordinatorMetadataCacheHelperImpl(metadataCache, key => shareCoordinator.partitionFor(key), config.interBrokerListenerName, groupConfigManager),
- Time.SYSTEM,
- shareGroupTimer
- )
- )
+ klass.getDeclaredMethod("instance", classOf[KafkaClient], classOf[ShareCoordinatorMetadataCacheHelper], classOf[Time], classOf[Timer])
Review Comment:
Out of curiosity, why use a dynamic call here if we already know it is a `DefaultStatePersister`
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] MINOR: Prevent this leak while DefaultStatePersister construction. [kafka]
Posted by "smjn (via GitHub)" <gi...@apache.org>.
smjn commented on PR #22292:
URL: https://github.com/apache/kafka/pull/22292#issuecomment-4475687204
@chia7712 Thanks for the comments, incorportated.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] MINOR: Prevent this leak while DefaultStatePersister construction. [kafka]
Posted by "smjn (via GitHub)" <gi...@apache.org>.
smjn commented on code in PR #22292:
URL: https://github.com/apache/kafka/pull/22292#discussion_r3255173398
##########
server-common/src/main/java/org/apache/kafka/server/share/persister/DefaultStatePersister.java:
##########
@@ -45,8 +48,29 @@ public class DefaultStatePersister implements Persister {
private static final Logger log = LoggerFactory.getLogger(DefaultStatePersister.class);
- public DefaultStatePersister(PersisterStateManager stateManager) {
+ public static DefaultStatePersister instance(KafkaClient client, ShareCoordinatorMetadataCacheHelper cacheHelper, Time time, Timer timer) {
+ DefaultStatePersister instance = new DefaultStatePersister(client, cacheHelper, time, timer);
+ instance.start();
+ return instance;
+ }
+
+ // Visibility for tests
+ static DefaultStatePersister instance(PersisterStateManager stateManager) {
Review Comment:
@chia7712 Thanks for the suggestion, incorported.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] MINOR: Prevent this leak while DefaultStatePersister construction. [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #22292:
URL: https://github.com/apache/kafka/pull/22292#discussion_r3251758585
##########
server-common/src/main/java/org/apache/kafka/server/share/persister/DefaultStatePersister.java:
##########
@@ -45,8 +45,17 @@ public class DefaultStatePersister implements Persister {
private static final Logger log = LoggerFactory.getLogger(DefaultStatePersister.class);
- public DefaultStatePersister(PersisterStateManager stateManager) {
+ public static DefaultStatePersister instance(PersisterStateManager stateManager) {
Review Comment:
If we want `DefaultStatePersister` to manage the lifecycle of `PersisterStateManager`, should we let it instantiate the manager itself?
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] MINOR: Prevent this leak while DefaultStatePersister construction. [kafka]
Posted by "smjn (via GitHub)" <gi...@apache.org>.
smjn commented on code in PR #22292:
URL: https://github.com/apache/kafka/pull/22292#discussion_r3257281502
##########
core/src/main/scala/kafka/server/BrokerServer.scala:
##########
@@ -724,17 +725,14 @@ class BrokerServer(
private def createShareStatePersister(): Persister = {
if (config.shareGroupConfig.shareGroupPersisterClassName.nonEmpty) {
val klass = Utils.loadClass(config.shareGroupConfig.shareGroupPersisterClassName, classOf[Object]).asInstanceOf[Class[Persister]]
-
if (klass.getName.equals(classOf[DefaultStatePersister].getName)) {
- klass.getConstructor(classOf[PersisterStateManager])
- .newInstance(
- new PersisterStateManager(
- NetworkUtils.buildNetworkClient("Persister", config, metrics, Time.SYSTEM, new LogContext(s"[Persister broker=${config.brokerId}]")),
- new ShareCoordinatorMetadataCacheHelperImpl(metadataCache, key => shareCoordinator.partitionFor(key), config.interBrokerListenerName, groupConfigManager),
- Time.SYSTEM,
- shareGroupTimer
- )
- )
+ klass.getDeclaredMethod("instance", classOf[KafkaClient], classOf[ShareCoordinatorMetadataCacheHelper], classOf[Time], classOf[Timer])
Review Comment:
Hi,
initially the persister was planned to be a pluggable component and code was written in that way but it was subsequently decided to make it factory pluggable. Hence, the if condition. But yes with current state of code, this is unnecessary, will rectify
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] MINOR: Prevent this leak while DefaultStatePersister construction. [kafka]
Posted by "smjn (via GitHub)" <gi...@apache.org>.
smjn commented on code in PR #22292:
URL: https://github.com/apache/kafka/pull/22292#discussion_r3257281502
##########
core/src/main/scala/kafka/server/BrokerServer.scala:
##########
@@ -724,17 +725,14 @@ class BrokerServer(
private def createShareStatePersister(): Persister = {
if (config.shareGroupConfig.shareGroupPersisterClassName.nonEmpty) {
val klass = Utils.loadClass(config.shareGroupConfig.shareGroupPersisterClassName, classOf[Object]).asInstanceOf[Class[Persister]]
-
if (klass.getName.equals(classOf[DefaultStatePersister].getName)) {
- klass.getConstructor(classOf[PersisterStateManager])
- .newInstance(
- new PersisterStateManager(
- NetworkUtils.buildNetworkClient("Persister", config, metrics, Time.SYSTEM, new LogContext(s"[Persister broker=${config.brokerId}]")),
- new ShareCoordinatorMetadataCacheHelperImpl(metadataCache, key => shareCoordinator.partitionFor(key), config.interBrokerListenerName, groupConfigManager),
- Time.SYSTEM,
- shareGroupTimer
- )
- )
+ klass.getDeclaredMethod("instance", classOf[KafkaClient], classOf[ShareCoordinatorMetadataCacheHelper], classOf[Time], classOf[Timer])
Review Comment:
Hi,
initially the persister was planned to be a pluggable component and code was written in that way but it was subsequently decided to make it factory pluggable, hence the if condition. But yes with current state of code, this is unnecessary, will rectify
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] MINOR: Prevent this leak while DefaultStatePersister construction. [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 merged PR #22292:
URL: https://github.com/apache/kafka/pull/22292
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] MINOR: Prevent this leak while DefaultStatePersister construction. [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #22292:
URL: https://github.com/apache/kafka/pull/22292#discussion_r3254178393
##########
server-common/src/main/java/org/apache/kafka/server/share/persister/DefaultStatePersister.java:
##########
@@ -45,8 +48,29 @@ public class DefaultStatePersister implements Persister {
private static final Logger log = LoggerFactory.getLogger(DefaultStatePersister.class);
- public DefaultStatePersister(PersisterStateManager stateManager) {
+ public static DefaultStatePersister instance(KafkaClient client, ShareCoordinatorMetadataCacheHelper cacheHelper, Time time, Timer timer) {
+ DefaultStatePersister instance = new DefaultStatePersister(client, cacheHelper, time, timer);
+ instance.start();
+ return instance;
+ }
+
+ // Visibility for tests
+ static DefaultStatePersister instance(PersisterStateManager stateManager) {
Review Comment:
It appears all the passed `stateManager` instances are mocked. Maybe we could remove this helper and have the tests use the package-private constructor instead?
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org