Skip to content

KAFKA-9839: Broker should accept control requests with newer broker epoch #8509

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

Merged
hachikuji merged 4 commits into from
Apr 27, 2020
Merged

KAFKA-9839: Broker should accept control requests with newer broker epoch #8509

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
0c9d015
KAFKA-9839: Broker should accept control requests with newer broker e…
apovzner Apr 18, 2020
d8604ed
added KafkaApisTest unit tests
apovzner Apr 22, 2020
062c787
fixed testStopReplicaRequestXXX unit tests and added a comment
apovzner Apr 22, 2020
d952854
adjusted comment alignment
apovzner Apr 23, 2020
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
7 changes: 1 addition & 6 deletions core/src/main/scala/kafka/server/KafkaApis.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -3084,12 +3084,7 @@ class KafkaApis(val requestChannel: RequestChannel,
// Broker epoch in LeaderAndIsr/UpdateMetadata/StopReplica request is unknown
// if the controller hasn't been upgraded to use KIP-380
if (brokerEpochInRequest == AbstractControlRequest.UNKNOWN_BROKER_EPOCH) false
else {
val curBrokerEpoch = controller.brokerEpoch
if (brokerEpochInRequest < curBrokerEpoch) true
else if (brokerEpochInRequest == curBrokerEpoch) false
else throw new IllegalStateException(s"Epoch $brokerEpochInRequest larger than current broker epoch $curBrokerEpoch")
}
else brokerEpochInRequest < controller.brokerEpoch

@hachikuji hachikuji Apr 22, 2020

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Short comment here may be helpful about the case where the controller sees the epoch bump first.

}

}
26 changes: 18 additions & 8 deletions core/src/test/scala/unit/kafka/server/BrokerEpochIntegrationTest.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -95,15 +95,20 @@ class BrokerEpochIntegrationTest extends ZooKeeperTestHarness {

@Test
def testControlRequestWithCorrectBrokerEpoch(): Unit = {
testControlRequestWithBrokerEpoch(false)
testControlRequestWithBrokerEpoch(0)
}

@Test
def testControlRequestWithStaleBrokerEpoch(): Unit = {
testControlRequestWithBrokerEpoch(true)
testControlRequestWithBrokerEpoch(-1)
}

private def testControlRequestWithBrokerEpoch(isEpochInRequestStale: Boolean): Unit = {
@Test
def testControlRequestWithNewerBrokerEpoch(): Unit = {
testControlRequestWithBrokerEpoch(1)
}

private def testControlRequestWithBrokerEpoch(epochInRequestDiffFromCurrentEpoch: Long): Unit = {
val tp = new TopicPartition("new-topic", 0)

// create topic with 1 partition, 2 replicas, one on each broker
Expand All @@ -128,8 +133,7 @@ class BrokerEpochIntegrationTest extends ZooKeeperTestHarness {
controllerChannelManager.startup()

val broker2 = servers(brokerId2)
val epochInRequest =
if (isEpochInRequestStale) broker2.kafkaController.brokerEpoch - 1 else broker2.kafkaController.brokerEpoch
val epochInRequest = broker2.kafkaController.brokerEpoch + epochInRequestDiffFromCurrentEpoch

try {
// Send LeaderAndIsr request with correct broker epoch
Expand All @@ -151,10 +155,12 @@ class BrokerEpochIntegrationTest extends ZooKeeperTestHarness {
epochInRequest,
partitionStates.asJava, nodes.toSet.asJava)

if (isEpochInRequestStale) {
if (epochInRequestDiffFromCurrentEpoch < 0) {
// stale broker epoch in LEADER_AND_ISR
sendAndVerifyStaleBrokerEpochInResponse(controllerChannelManager, requestBuilder)
}
else {
// broker epoch in LEADER_AND_ISR >= current broker epoch
sendAndVerifySuccessfulResponse(controllerChannelManager, requestBuilder)
TestUtils.waitUntilLeaderIsKnown(Seq(broker2), tp, 10000)
}
Expand Down Expand Up @@ -191,10 +197,12 @@ class BrokerEpochIntegrationTest extends ZooKeeperTestHarness {
epochInRequest,
partitionStates.asJava, liveBrokers.asJava)

if (isEpochInRequestStale) {
if (epochInRequestDiffFromCurrentEpoch < 0) {
// stale broker epoch in UPDATE_METADATA
sendAndVerifyStaleBrokerEpochInResponse(controllerChannelManager, requestBuilder)
}
else {
// broker epoch in UPDATE_METADATA >= current broker epoch
sendAndVerifySuccessfulResponse(controllerChannelManager, requestBuilder)
TestUtils.waitUntilMetadataIsPropagated(Seq(broker2), tp.topic, tp.partition, 10000)
assertEquals(brokerId2,
Expand All @@ -217,9 +225,11 @@ class BrokerEpochIntegrationTest extends ZooKeeperTestHarness {
epochInRequest, // Correct broker epoch
false, topicStates)

if (isEpochInRequestStale) {
if (epochInRequestDiffFromCurrentEpoch < 0) {
// stale broker epoch in STOP_REPLICA
sendAndVerifyStaleBrokerEpochInResponse(controllerChannelManager, requestBuilder)
} else {
// broker epoch in STOP_REPLICA >= current broker epoch
sendAndVerifySuccessfulResponse(controllerChannelManager, requestBuilder)
assertEquals(HostedPartition.None, broker2.replicaManager.getPartition(tp))
}
Expand Down