chia7712 commented on code in PR #15670:
URL: https://github.com/apache/kafka/pull/15670#discussion_r1554282432
##########
core/src/main/scala/kafka/server/DynamicBrokerConfig.scala:
##########
@@ -1136,18 +1137,18 @@ class DynamicListenerConfig(server: KafkaBroker)
extends BrokerReconfigurable wi
class DynamicProducerStateManagerConfig(val producerStateManagerConfig:
ProducerStateManagerConfig) extends BrokerReconfigurable with Logging {
def reconfigure(oldConfig: KafkaConfig, newConfig: KafkaConfig): Unit = {
if (producerStateManagerConfig.producerIdExpirationMs !=
newConfig.producerIdExpirationMs) {
- info(s"Reconfigure ${KafkaConfig.ProducerIdExpirationMsProp} from
${producerStateManagerConfig.producerIdExpirationMs} to
${newConfig.producerIdExpirationMs}")
+ info(s"Reconfigure
${TransactionLogConfig.PRODUCER_ID_EXPIRATION_MS_CONFIG} from
${producerStateManagerConfig.producerIdExpirationMs} to
${newConfig.producerIdExpirationMs}")
producerStateManagerConfig.setProducerIdExpirationMs(newConfig.producerIdExpirationMs)
}
if (producerStateManagerConfig.transactionVerificationEnabled !=
newConfig.transactionPartitionVerificationEnable) {
- info(s"Reconfigure
${KafkaConfig.TransactionPartitionVerificationEnableProp} from
${producerStateManagerConfig.transactionVerificationEnabled} to
${newConfig.transactionPartitionVerificationEnable}")
+ info(s"Reconfigure
${TransactionLogConfig.TRANSACTION_PARTITION_VERIFICATION_ENABLE_CONFIG} from
${producerStateManagerConfig.transactionVerificationEnabled} to
${newConfig.transactionPartitionVerificationEnable}")
producerStateManagerConfig.setTransactionVerificationEnabled(newConfig.transactionPartitionVerificationEnable)
}
}
def validateReconfiguration(newConfig: KafkaConfig): Unit = {
if (newConfig.producerIdExpirationMs < 0)
- throw new ConfigException(s"${KafkaConfig.ProducerIdExpirationMsProp}
cannot be less than 0, current value is
${producerStateManagerConfig.producerIdExpirationMs}, and new value is
${newConfig.producerIdExpirationMs}")
+ throw new
ConfigException(s"${TransactionLogConfig.PRODUCER_ID_EXPIRATION_MS_CONFIG}
cannot be less than 0, current value is
${producerStateManagerConfig.producerIdExpirationMs}, and new value is
${newConfig.producerIdExpirationMs}")
Review Comment:
maybe `ProducerStateManagerConfig.PRODUCER_ID_EXPIRATION_MS` is more
suitable in this case because the code is used for logging `Producer` state.
##########
core/src/main/scala/kafka/server/KafkaConfig.scala:
##########
@@ -1062,21 +1027,21 @@ object KafkaConfig {
.define(CompressionTypeProp, STRING, LogConfig.DEFAULT_COMPRESSION_TYPE,
in(BrokerCompressionType.names.asScala.toSeq:_*), HIGH, CompressionTypeDoc)
/** ********* Transaction management configuration ***********/
- .define(TransactionalIdExpirationMsProp, INT,
Defaults.TRANSACTIONAL_ID_EXPIRATION_MS, atLeast(1), HIGH,
TransactionalIdExpirationMsDoc)
- .define(TransactionsMaxTimeoutMsProp, INT,
Defaults.TRANSACTIONS_MAX_TIMEOUT_MS, atLeast(1), HIGH,
TransactionsMaxTimeoutMsDoc)
- .define(TransactionsTopicMinISRProp, INT,
Defaults.TRANSACTIONS_TOPIC_MIN_ISR, atLeast(1), HIGH,
TransactionsTopicMinISRDoc)
- .define(TransactionsLoadBufferSizeProp, INT,
Defaults.TRANSACTIONS_LOAD_BUFFER_SIZE, atLeast(1), HIGH,
TransactionsLoadBufferSizeDoc)
- .define(TransactionsTopicReplicationFactorProp, SHORT,
Defaults.TRANSACTIONS_TOPIC_REPLICATION_FACTOR, atLeast(1), HIGH,
TransactionsTopicReplicationFactorDoc)
- .define(TransactionsTopicPartitionsProp, INT,
Defaults.TRANSACTIONS_TOPIC_PARTITIONS, atLeast(1), HIGH,
TransactionsTopicPartitionsDoc)
- .define(TransactionsTopicSegmentBytesProp, INT,
Defaults.TRANSACTIONS_TOPIC_SEGMENT_BYTES, atLeast(1), HIGH,
TransactionsTopicSegmentBytesDoc)
- .define(TransactionsAbortTimedOutTransactionCleanupIntervalMsProp, INT,
Defaults.TRANSACTIONS_ABORT_TIMED_OUT_CLEANUP_INTERVAL_MS, atLeast(1), LOW,
TransactionsAbortTimedOutTransactionsIntervalMsDoc)
- .define(TransactionsRemoveExpiredTransactionalIdCleanupIntervalMsProp,
INT, Defaults.TRANSACTIONS_REMOVE_EXPIRED_CLEANUP_INTERVAL_MS, atLeast(1), LOW,
TransactionsRemoveExpiredTransactionsIntervalMsDoc)
-
- .define(TransactionPartitionVerificationEnableProp, BOOLEAN,
Defaults.TRANSACTION_PARTITION_VERIFICATION_ENABLE, LOW,
TransactionPartitionVerificationEnableDoc)
-
- .define(ProducerIdExpirationMsProp, INT,
Defaults.PRODUCER_ID_EXPIRATION_MS, atLeast(1), LOW, ProducerIdExpirationMsDoc)
+
.define(TransactionStateManagerConfig.TRANSACTIONAL_ID_EXPIRATION_MS_CONFIG,
INT, Defaults.TRANSACTIONAL_ID_EXPIRATION_MS, atLeast(1), HIGH,
TransactionStateManagerConfig.TRANSACTIONAL_ID_EXPIRATION_MS_DOC)
Review Comment:
Should we do similar refactor for `Defaults`?
--
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
