Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
Expand Up @@ -56,7 +56,7 @@ object TransactionCoordinator {
val txnMarkerPurgatory = DelayedOperationPurgatory[DelayedTxnMarker]("txn-marker-purgatory", config.brokerId,
reaperEnabled = false, timerEnabled = false)
val txnStateManager = new TransactionStateManager(config.brokerId, zkClient, scheduler, replicaManager, txnConfig,
time, metrics, config.interBrokerProtocolVersion)
time, metrics)

val logContext = new LogContext(s"[TransactionCoordinator id=${config.brokerId}] ")
val txnMarkerChannelManager = TransactionMarkerChannelManager(config, metrics, metadataCache, txnStateManager,
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -22,7 +22,6 @@ import java.util.concurrent.TimeUnit
import java.util.concurrent.atomic.AtomicBoolean
import java.util.concurrent.locks.ReentrantReadWriteLock

import kafka.api.ApiVersion
import kafka.log.{AppendOrigin, LogConfig}
import kafka.message.UncompressedCodec
import kafka.server.{Defaults, FetchLogEnd, ReplicaManager}
Expand Down Expand Up @@ -74,8 +73,7 @@ class TransactionStateManager(brokerId: Int,
replicaManager: ReplicaManager,
config: TransactionConfig,
time: Time,
metrics: Metrics,
interBrokerProtocolVersion: ApiVersion) extends Logging {
metrics: Metrics) extends Logging {

this.logIdent = "[Transaction State Manager " + brokerId + "]: "

Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -18,7 +18,6 @@ package kafka.coordinator.transaction

import java.nio.ByteBuffer

import kafka.api.KAFKA_2_4_IV1
import kafka.coordinator.AbstractCoordinatorConcurrencyTest
import kafka.coordinator.AbstractCoordinatorConcurrencyTest._
import kafka.coordinator.transaction.TransactionCoordinatorConcurrencyTest._
Expand Down Expand Up @@ -73,7 +72,7 @@ class TransactionCoordinatorConcurrencyTest extends AbstractCoordinatorConcurren
EasyMock.replay(zkClient)

txnStateManager = new TransactionStateManager(0, zkClient, scheduler, replicaManager, txnConfig, time,
new Metrics(), KAFKA_2_4_IV1)
new Metrics())
for (i <- 0 until numPartitions)
txnStateManager.addLoadedTransactionsToCache(i, coordinatorEpoch, new Pool[String, TransactionMetadata]())

Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -67,7 +67,7 @@ class TransactionStateManagerTest {

val txnConfig = TransactionConfig()
val transactionManager: TransactionStateManager = new TransactionStateManager(0, zkClient, scheduler,
replicaManager, txnConfig, time, metrics, KAFKA_2_4_IV1)
replicaManager, txnConfig, time, metrics)

val transactionalId1: String = "one"
val transactionalId2: String = "two"
Expand Down