Skip to content

KAFKA-12388: Share broker channel between alterIsrManager and lifecycleManager #10255

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
dengziming wants to merge 1 commit into from
Closed

KAFKA-12388: Share broker channel between alterIsrManager and lifecycleManager #10255

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
36 changes: 9 additions & 27 deletions core/src/main/scala/kafka/server/AlterIsrManager.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,20 +17,20 @@
package kafka.server

import java.util
import java.util.concurrent.atomic.AtomicBoolean
import java.util.concurrent.{ConcurrentHashMap, TimeUnit}

import kafka.api.LeaderAndIsr
import kafka.metrics.KafkaMetricsGroup
import kafka.utils.{KafkaScheduler, Logging, Scheduler}
import kafka.utils.{Logging, Scheduler}
import kafka.zk.KafkaZkClient
import org.apache.kafka.clients.ClientResponse
import org.apache.kafka.common.TopicPartition
import org.apache.kafka.common.message.{AlterIsrRequestData, AlterIsrResponseData}
import org.apache.kafka.common.metrics.Metrics
import org.apache.kafka.common.protocol.Errors
import org.apache.kafka.common.requests.{AlterIsrRequest, AlterIsrResponse}
import org.apache.kafka.common.utils.Time

import java.util.concurrent.atomic.AtomicBoolean
import scala.collection.mutable
import scala.collection.mutable.ListBuffer
import scala.jdk.CollectionConverters._
Expand Down Expand Up @@ -64,26 +64,12 @@ object AlterIsrManager {
* Factory to AlterIsr based implementation, used when IBP >= 2.7-IV2
*/
def apply(
config: KafkaConfig,
metadataCache: MetadataCache,
scheduler: KafkaScheduler,
channelManager: BrokerToControllerChannelManager,
scheduler: Scheduler,
time: Time,
metrics: Metrics,
threadNamePrefix: Option[String],
brokerEpochSupplier: () => Long,
brokerId: Int
brokerId: Int,
brokerEpochSupplier: () => Long
): AlterIsrManager = {
val nodeProvider = MetadataCacheControllerNodeProvider(config, metadataCache)

val channelManager = BrokerToControllerChannelManager(
controllerNodeProvider = nodeProvider,
time = time,
metrics = metrics,
config = config,
channelName = "alterIsr",
threadNamePrefix = threadNamePrefix,
retryTimeoutMs = Long.MaxValue
)
new DefaultAlterIsrManager(
controllerChannelManager = channelManager,
scheduler = scheduler,
Expand Down Expand Up @@ -120,13 +106,9 @@ class DefaultAlterIsrManager(
// Used to allow only one in-flight request at a time
private val inflightRequest: AtomicBoolean = new AtomicBoolean(false)

override def start(): Unit = {
controllerChannelManager.start()
}
override def start(): Unit = {}

override def shutdown(): Unit = {
controllerChannelManager.shutdown()
}
override def shutdown(): Unit = {}

override def submit(alterIsrItem: AlterIsrItem): Boolean = {
val enqueued = unsentIsrUpdates.putIfAbsent(alterIsrItem.topicPartition, alterIsrItem) == null
Expand Down
2 changes: 0 additions & 2 deletions core/src/main/scala/kafka/server/BrokerLifecycleManager.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -251,7 +251,6 @@ class BrokerLifecycleManager(val config: KafkaConfig,
override def run(): Unit = {
_highestMetadataOffsetProvider = highestMetadataOffsetProvider
_channelManager = channelManager
_channelManager.start()
_state = BrokerState.STARTING
_clusterId = clusterId
_advertisedListeners = advertisedListeners.duplicate()
Expand Down Expand Up @@ -473,7 +472,6 @@ class BrokerLifecycleManager(val config: KafkaConfig,
controlledShutdownFuture.complete(null)
initialCatchUpFuture.cancel(false)
if (_channelManager != null) {
_channelManager.shutdown()
_channelManager = null
}
}
Expand Down
22 changes: 14 additions & 8 deletions core/src/main/scala/kafka/server/BrokerServer.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -106,6 +106,8 @@ class BrokerServer(

var forwardingManager: ForwardingManager = null

var brokerToControllerChannelManager: BrokerToControllerChannelManager = null

var alterIsrManager: AlterIsrManager = null

var autoTopicCreationManager: AutoTopicCreationManager = null
Expand Down Expand Up @@ -186,7 +188,7 @@ class BrokerServer(
time,
metrics,
config,
channelName = "forwarding",
channelName = "clientForwarding",
threadNamePrefix,
retryTimeoutMs = 60000
)
Expand All @@ -207,20 +209,22 @@ class BrokerServer(
socketServer = new SocketServer(config, metrics, time, credentialProvider, apiVersionManager)
socketServer.startup(startProcessingRequests = false)

val alterIsrChannelManager = BrokerToControllerChannelManager(
brokerToControllerChannelManager = BrokerToControllerChannelManager(
controllerNodeProvider,
time,
metrics,
config,
channelName = "alterIsr",
channelName = "brokerForwarding",
threadNamePrefix,
retryTimeoutMs = Long.MaxValue
)
alterIsrManager = new DefaultAlterIsrManager(
controllerChannelManager = alterIsrChannelManager,
brokerToControllerChannelManager.start()

alterIsrManager = AlterIsrManager(
brokerToControllerChannelManager,
scheduler = kafkaScheduler,
time = time,
brokerId = config.nodeId,
brokerId = config.brokerId,
brokerEpochSupplier = () => lifecycleManager.brokerEpoch()
)
alterIsrManager.start()
Expand Down Expand Up @@ -277,8 +281,7 @@ class BrokerServer(
setSecurityProtocol(ep.securityProtocol.id))
}
lifecycleManager.start(() => brokerMetadataListener.highestMetadataOffset(),
BrokerToControllerChannelManager(controllerNodeProvider, time, metrics, config,
"heartbeat", threadNamePrefix, config.brokerSessionTimeoutMs.toLong),
brokerToControllerChannelManager,
metaProps.clusterId, networkListeners, supportedFeatures)

// Register a listener with the Raft layer to receive metadata event notifications
Expand Down Expand Up @@ -450,6 +453,9 @@ class BrokerServer(
if (clientToControllerChannelManager != null)
CoreUtils.swallow(clientToControllerChannelManager.shutdown(), this)

if (brokerToControllerChannelManager != null)
CoreUtils.swallow(brokerToControllerChannelManager.shutdown(), this)

if (logManager != null)
CoreUtils.swallow(logManager.shutdown(), this)

Expand Down
30 changes: 24 additions & 6 deletions core/src/main/scala/kafka/server/KafkaServer.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -135,8 +135,12 @@ class KafkaServer(

var autoTopicCreationManager: AutoTopicCreationManager = null

/** Channel to send client request to controller, such as auto topic creation and forwarding */
var clientToControllerChannelManager: Option[BrokerToControllerChannelManager] = None

/** Channel to send broker request to controller, such as AlterIsr(KIP-497) and AlterReplicaState(KIP-589) */
var brokerToControllerChannelManager: Option[BrokerToControllerChannelManager] = None

var alterIsrManager: AlterIsrManager = null

var kafkaScheduler: KafkaScheduler = null
Expand Down Expand Up @@ -265,7 +269,7 @@ class KafkaServer(
time = time,
metrics = metrics,
config = config,
channelName = "forwarding",
channelName = "clientForwarding",
threadNamePrefix = threadNamePrefix,
retryTimeoutMs = config.requestTimeoutMs.longValue)
brokerToControllerManager.start()
Expand All @@ -290,17 +294,29 @@ class KafkaServer(
socketServer = new SocketServer(config, metrics, time, credentialProvider, apiVersionManager)
socketServer.startup(startProcessingRequests = false)

// Currently we only support alterIsr, maybe alterReplicaState in the future
if (config.interBrokerProtocolVersion.isAlterIsrSupported) {
val nodeProvider = MetadataCacheControllerNodeProvider(config, metadataCache)
val channelManager = BrokerToControllerChannelManager(
controllerNodeProvider = nodeProvider,
time = time,
metrics = metrics,
config = config,
channelName = "brokerToControllerForwardingChannel",
threadNamePrefix = threadNamePrefix,
retryTimeoutMs = Long.MaxValue)
channelManager.start()
brokerToControllerChannelManager = Some(channelManager)
}

/* start replica manager */
alterIsrManager = if (config.interBrokerProtocolVersion.isAlterIsrSupported) {
AlterIsrManager(
config = config,
metadataCache = metadataCache,
channelManager = brokerToControllerChannelManager.get,
scheduler = kafkaScheduler,
time = time,
metrics = metrics,
threadNamePrefix = threadNamePrefix,
brokerEpochSupplier = () => kafkaController.brokerEpoch,
config.brokerId
brokerId = config.brokerId
)
} else {
AlterIsrManager(kafkaScheduler, time, zkClient)
Expand Down Expand Up @@ -707,6 +723,8 @@ class KafkaServer(

CoreUtils.swallow(clientToControllerChannelManager.foreach(_.shutdown()), this)

CoreUtils.swallow(brokerToControllerChannelManager.foreach(_.shutdown()), this)

if (logManager != null)
CoreUtils.swallow(logManager.shutdown(), this)

Expand Down
21 changes: 7 additions & 14 deletions core/src/test/scala/unit/kafka/server/AlterIsrManagerTest.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -56,12 +56,11 @@ class AlterIsrManagerTest {

@Test
def testBasic(): Unit = {
EasyMock.expect(brokerToController.start())
EasyMock.expect(brokerToController.sendRequest(EasyMock.anyObject(), EasyMock.anyObject())).once()
EasyMock.replay(brokerToController)

val scheduler = new MockScheduler(time)
val alterIsrManager = new DefaultAlterIsrManager(brokerToController, scheduler, time, brokerId, () => 2)
val alterIsrManager = AlterIsrManager(brokerToController, scheduler, time, brokerId, () => 2)
alterIsrManager.start()
alterIsrManager.submit(AlterIsrItem(tp0, new LeaderAndIsr(1, 1, List(1,2,3), 10), _ => {}, 0))
EasyMock.verify(brokerToController)
Expand All @@ -70,12 +69,11 @@ class AlterIsrManagerTest {
@Test
def testOverwriteWithinBatch(): Unit = {
val capture = EasyMock.newCapture[AbstractRequest.Builder[AlterIsrRequest]]()
EasyMock.expect(brokerToController.start())
EasyMock.expect(brokerToController.sendRequest(EasyMock.capture(capture), EasyMock.anyObject())).once()
EasyMock.replay(brokerToController)

val scheduler = new MockScheduler(time)
val alterIsrManager = new DefaultAlterIsrManager(brokerToController, scheduler, time, brokerId, () => 2)
val alterIsrManager = AlterIsrManager(brokerToController, scheduler, time, brokerId, () => 2)
alterIsrManager.start()

// Only send one ISR update for a given topic+partition
Expand All @@ -93,12 +91,11 @@ class AlterIsrManagerTest {
val capture = EasyMock.newCapture[AbstractRequest.Builder[AlterIsrRequest]]()
val callbackCapture = EasyMock.newCapture[ControllerRequestCompletionHandler]()

EasyMock.expect(brokerToController.start())
EasyMock.expect(brokerToController.sendRequest(EasyMock.capture(capture), EasyMock.capture(callbackCapture))).times(2)
EasyMock.replay(brokerToController)

val scheduler = new MockScheduler(time)
val alterIsrManager = new DefaultAlterIsrManager(brokerToController, scheduler, time, brokerId, () => 2)
val alterIsrManager = AlterIsrManager(brokerToController, scheduler, time, brokerId, () => 2)
alterIsrManager.start()

// First request will send batch of one
Expand Down Expand Up @@ -166,7 +163,6 @@ class AlterIsrManagerTest {
val isrs = Seq(AlterIsrItem(tp0, leaderAndIsr, _ => { }, 0))
val callbackCapture = EasyMock.newCapture[ControllerRequestCompletionHandler]()

EasyMock.expect(brokerToController.start())
EasyMock.expect(brokerToController.sendRequest(EasyMock.anyObject(), EasyMock.capture(callbackCapture))).times(1)
EasyMock.replay(brokerToController)

Expand Down Expand Up @@ -225,12 +221,11 @@ class AlterIsrManagerTest {
private def testPartitionError(tp: TopicPartition, error: Errors): AlterIsrManager = {
val callbackCapture = EasyMock.newCapture[ControllerRequestCompletionHandler]()
EasyMock.reset(brokerToController)
EasyMock.expect(brokerToController.start())
EasyMock.expect(brokerToController.sendRequest(EasyMock.anyObject(), EasyMock.capture(callbackCapture))).once()
EasyMock.replay(brokerToController)

val scheduler = new MockScheduler(time)
val alterIsrManager = new DefaultAlterIsrManager(brokerToController, scheduler, time, brokerId, () => 2)
val alterIsrManager = AlterIsrManager(brokerToController, scheduler, time, brokerId, () => 2)
alterIsrManager.start()

var capturedError: Option[Errors] = None
Expand Down Expand Up @@ -259,12 +254,11 @@ class AlterIsrManagerTest {
def testOneInFlight(): Unit = {
val callbackCapture = EasyMock.newCapture[ControllerRequestCompletionHandler]()
EasyMock.reset(brokerToController)
EasyMock.expect(brokerToController.start())
EasyMock.expect(brokerToController.sendRequest(EasyMock.anyObject(), EasyMock.capture(callbackCapture))).once()
EasyMock.replay(brokerToController)

val scheduler = new MockScheduler(time)
val alterIsrManager = new DefaultAlterIsrManager(brokerToController, scheduler, time, brokerId, () => 2)
val alterIsrManager = AlterIsrManager(brokerToController, scheduler, time, brokerId, () => 2)
alterIsrManager.start()

// First submit will send the request
Expand All @@ -291,16 +285,15 @@ class AlterIsrManagerTest {
def testPartitionMissingInResponse(): Unit = {
val callbackCapture = EasyMock.newCapture[ControllerRequestCompletionHandler]()
EasyMock.reset(brokerToController)
EasyMock.expect(brokerToController.start())
EasyMock.expect(brokerToController.sendRequest(EasyMock.anyObject(), EasyMock.capture(callbackCapture))).once()
EasyMock.replay(brokerToController)

val scheduler = new MockScheduler(time)
val alterIsrManager = new DefaultAlterIsrManager(brokerToController, scheduler, time, brokerId, () => 2)
val alterIsrManager = AlterIsrManager(brokerToController, scheduler, time, brokerId, () => 2)
alterIsrManager.start()

val count = new AtomicInteger(0)
val callback = (result: Either[Errors, LeaderAndIsr]) => {
val callback = (_: Either[Errors, LeaderAndIsr]) => {
count.incrementAndGet()
return
}
Expand Down