jolshan commented on code in PR #12643:
URL: https://github.com/apache/kafka/pull/12643#discussion_r975585267
##########
core/src/main/scala/kafka/log/ProducerStateManager.scala:
##########
@@ -939,3 +940,24 @@ object SnapshotFile {
SnapshotFile(file, offset)
}
}
+
+class ProducerStateManagerConfig(@volatile var producerIdExpirationMs: Int)
extends Logging with BrokerReconfigurable {
+
+ override def reconfigurableConfigs: Set[String] =
Set(KafkaConfig.ProducerIdExpirationMsProp)
+
+ override def reconfigure(oldConfig: KafkaConfig, newConfig: KafkaConfig):
Unit = {
+ if (producerIdExpirationMs != newConfig.producerIdExpirationMs) {
+ info(s"Reconfigure ${KafkaConfig.ProducerIdExpirationMsProp} from
$producerIdExpirationMs to ${newConfig.producerIdExpirationMs}")
+ producerIdExpirationMs = newConfig.producerIdExpirationMs
+ }
+ }
+
+ override def validateReconfiguration(newConfig: KafkaConfig): Unit = {
+ if (newConfig.producerIdExpirationMs < 0)
+ throw new ConfigException(s"${KafkaConfig.ProducerIdExpirationMsProp}
cannot be less than 0, current value is $producerIdExpirationMs")
+ }
+}
+
+object ProducerStateManagerConfig {
+ val ReconfigurableConfigs = Set(KafkaConfig.ProducerIdExpirationMsProp)
Review Comment:
This is required to override the class.
--
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-unsubscr...@kafka.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
