Skip to content
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

Jump to bottom

KAFKA-15036: Add a test case for controller failover #13777

Closed
wants to merge 3 commits into from
Closed

KAFKA-15036: Add a test case for controller failover #13777

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
91efc0c
KAFKA-15036: UnknownServerError on any leader failover
dengziming May 30, 2023
0c20060
Add a test case for restarting controller node
dengziming May 25, 2023
59d5c1e
comments
dengziming May 31, 2023
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Jump to
Jump to file
Failed to load files.
Diff view
Diff view
4 changes: 3 additions & 1 deletion core/src/main/scala/kafka/server/ControllerServer.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -32,6 +32,7 @@ import kafka.utils.{CoreUtils, Logging, PasswordEncoder}
import kafka.zk.{KafkaZkClient, ZkMigrationClient}
import org.apache.kafka.common.config.ConfigException
import org.apache.kafka.common.message.ApiMessageType.ListenerType
import org.apache.kafka.common.metrics.Metrics
import org.apache.kafka.common.security.scram.internals.ScramMechanism
import org.apache.kafka.common.security.token.delegation.internals.DelegationTokenCache
import org.apache.kafka.common.utils.LogContext
Expand Down Expand Up @@ -182,7 +183,8 @@ class ControllerServer(
tokenCache = new DelegationTokenCache(ScramMechanism.mechanismNames)
credentialProvider = new CredentialProvider(ScramMechanism.mechanismNames, tokenCache)
socketServer = new SocketServer(config,
metrics,
// metrics will be null when restarting a controller, this will only happen in test.
if (metrics == null) new Metrics() else metrics,
time,
credentialProvider,
apiVersionManager)
Expand Down
28 changes: 27 additions & 1 deletion core/src/test/scala/integration/kafka/server/KRaftClusterTest.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -80,7 +80,7 @@ class KRaftClusterTest {
}

@Test
def testCreateClusterAndRestartNode(): Unit = {
def testCreateClusterAndRestartBrokerNode(): Unit = {
val cluster = new KafkaClusterTestKit.Builder(
new TestKitNodes.Builder().
setNumBrokerNodes(1).
Expand All @@ -96,6 +96,32 @@ class KRaftClusterTest {
}
}

@Test
def testCreateClusterAndRestartControllerNode(): Unit = {
Copy link
Member Author

Choose a reason for hiding this comment

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

@soarez Yes, we shutdown the active controller, then restart it to make it send ApiVersionRequest to a standby controller.

val cluster = new KafkaClusterTestKit.Builder(
new TestKitNodes.Builder().
setNumBrokerNodes(1).
setNumControllerNodes(2).build()).build()
try {
cluster.format()
cluster.startup()
val controller = cluster.controllers().values().iterator().asScala.filter(_.controller.isActive).next()
val port = controller.socketServer.boundPort(controller.config.controllerListeners.head.listenerName)

// shutdown active controller
controller.shutdown()
// Rewrite The `listeners` config to avoid controller socket server init using different port
val config = controller.sharedServer.controllerConfig.props
config.asInstanceOf[java.util.HashMap[String,String]].put(KafkaConfig.ListenersProp, s"CONTROLLER://localhost:$port")
controller.sharedServer.controllerConfig.updateCurrentConfig(new KafkaConfig(config))

// restart controller
controller.startup()
} finally {
cluster.close()
}
}

@Test
def testCreateClusterAndWaitForBrokerInRunningState(): Unit = {
val cluster = new KafkaClusterTestKit.Builder(
Expand Down
5 changes: 4 additions & 1 deletion metadata/src/main/java/org/apache/kafka/controller/QuorumController.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2028,8 +2028,11 @@ public CompletableFuture<FinalizedControllerFeatures> finalizedFeatures(
if (lastCommittedOffset == -1) {
return CompletableFuture.completedFuture(new FinalizedControllerFeatures(Collections.emptyMap(), -1));
}
// It's possible for a standby controller to receive ApiVersionRequest and we do not have any timeline snapshot
// in a standby controller, in this case we use SnapshotRegistry.LATEST_EPOCH.
long epoch = isActive() ? lastCommittedOffset : SnapshotRegistry.LATEST_EPOCH;
return appendReadEvent("getFinalizedFeatures", context.deadlineNs(),
() -> featureControl.finalizedFeatures(lastCommittedOffset));
() -> featureControl.finalizedFeatures(epoch));
}

@Override
Expand Down