root
/
kafka
mirror of https://github.com/apache/kafka.git
1
0
Fork 0
Code Issues Actions 3 Packages Projects Releases Wiki Activity

MINOR: Remove duplicate definition of transactional.id.expiration.ms config (#7245) 

Reviewers: Bob Barrett <bob.barrett@outlook.com>, Apurva Mehta <apurva@confluent.io>, Ismael Juma <ismael@juma.me.uk>
This commit is contained in:
Stanislav Kozlovski 2019-09-17 22:28:22 +01:00 committed by Ismael Juma
parent 4962c8193e
commit dcfd31c552
2 changed files with 4 additions and 10 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
core/src/main/scala/kafka/log/LogManager.scala
View File

@ -1036,7 +1036,7 @@ object LogManager {
flushRecoveryOffsetCheckpointMs = config.logFlushOffsetCheckpointIntervalMs,
flushStartOffsetCheckpointMs = config.logFlushStartOffsetCheckpointIntervalMs,
retentionCheckMs = config.logCleanupIntervalMs,
maxPidExpirationMs = config.transactionIdExpirationMs,
maxPidExpirationMs = config.transactionalIdExpirationMs,
scheduler = kafkaScheduler,
brokerState = brokerState,
brokerTopicStats = brokerTopicStats,

12
core/src/main/scala/kafka/server/KafkaConfig.scala
View File

@ -731,7 +731,9 @@ object KafkaConfig {
"or this timeout is reached. This is similar to the producer request timeout."
val OffsetCommitRequiredAcksDoc = "The required acks before the commit can be accepted. In general, the default (-1) should not be overridden"
/** ********* Transaction management configuration ***********/
val TransactionalIdExpirationMsDoc = "The maximum amount of time in ms that the transaction coordinator will wait before proactively expire a producer's transactional id without receiving any transaction status updates from it."
val TransactionalIdExpirationMsDoc = "The time in ms that the transaction coordinator will wait without receiving any transaction status updates " +
"for the current transaction before expiring its transactional id. This setting also influences producer id expiration - producer ids are expired " +
"once this time has elapsed after the last write with the given producer id. Note that producer ids may expire sooner if the last write from the producer id is deleted due to the topic's retention settings."
val TransactionsMaxTimeoutMsDoc = "The maximum allowed timeout for transactions. " +
"If a client’s requested transaction time exceed this, then the broker will return an error in InitProducerIdRequest. This prevents a client from too large of a timeout, which can stall consumers reading from topics included in the transaction."
val TransactionsTopicMinISRDoc = "Overridden " + MinInSyncReplicasProp + " config for the transaction topic."
@ -761,11 +763,6 @@ object KafkaConfig {
"which is used to determine quota limits applied to client requests. By default, <user, client-id>, <user> or <client-id> " +
"quotas stored in ZooKeeper are applied. For any given request, the most specific quota that matches the user principal " +
"of the session and the client-id of the request is applied."
/** ********* Transaction Configuration ***********/
val TransactionIdExpirationMsDoc = "The maximum time of inactivity before a transactional id is expired by the " +
"transaction coordinator. Note that this also influences producer id expiration: Producer ids are guaranteed to expire " +
"after expiration of this timeout from the last write by the producer id (they may expire sooner if the last write " +
"from the producer id is deleted due to the topic's retention settings)."
val DeleteTopicEnableDoc = "Enables delete topic. Delete topic through the admin tool will have no effect if this config is turned off"
val CompressionTypeDoc = "Specify the final compression type for a given topic. This configuration accepts the standard compression codecs " +
@ -1361,9 +1358,6 @@ class KafkaConfig(val props: java.util.Map[_, _], doLog: Boolean, dynamicConfigO
val numAlterLogDirsReplicationQuotaSamples = getInt(KafkaConfig.NumAlterLogDirsReplicationQuotaSamplesProp)
val alterLogDirsReplicationQuotaWindowSizeSeconds = getInt(KafkaConfig.AlterLogDirsReplicationQuotaWindowSizeSecondsProp)
/** ********* Transaction Configuration **************/
val transactionIdExpirationMs = getInt(KafkaConfig.TransactionalIdExpirationMsProp)
/** ********* Fetch Session Configuration **************/
val maxIncrementalFetchSessionCacheSlots = getInt(KafkaConfig.MaxIncrementalFetchSessionCacheSlots)