Skip to content

KAFKA-13228; Ensure ApiVersionRequest is properly handled KRaft co-resident mode #11784

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 6 commits into from
Jul 5, 2022
Merged

KAFKA-13228; Ensure ApiVersionRequest is properly handled KRaft co-resident mode #11784

Show file tree
Hide file tree
Changes from 4 commits
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
8f795a4
Support co-resident in Test Kit
dengziming Feb 17, 2022
0e76ad9
KAFKA-13228: Ensure ApiVersionRequest is properly handled in co-resid…
dengziming Aug 26, 2021
ac07a78
Enable Kraft co-resident mode in ApiVersionTest
dengziming May 25, 2022
d4b9a04
resolve comments
dengziming Jun 30, 2022
c631dae
resolve comments
dengziming Jul 1, 2022
58af486
Create controller node provider after metadata cache
hachikuji Jul 1, 2022
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
16 changes: 2 additions & 14 deletions core/src/main/scala/kafka/server/AlterPartitionManager.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -29,7 +29,6 @@ import org.apache.kafka.common.TopicPartition
import org.apache.kafka.common.Uuid
import org.apache.kafka.common.errors.OperationNotAttemptedException
import org.apache.kafka.common.message.AlterPartitionRequestData
import org.apache.kafka.common.metrics.Metrics
import org.apache.kafka.common.protocol.Errors
import org.apache.kafka.common.requests.RequestHeader
import org.apache.kafka.common.requests.{AlterPartitionRequest, AlterPartitionResponse}
Expand Down Expand Up @@ -77,23 +76,12 @@ object AlterPartitionManager {
def apply(
config: KafkaConfig,
metadataCache: MetadataCache,
channelManager: BrokerToControllerChannelManager,
scheduler: KafkaScheduler,
time: Time,
metrics: Metrics,
threadNamePrefix: Option[String],
brokerEpochSupplier: () => Long,
brokerEpochSupplier: () => Long
): AlterPartitionManager = {
val nodeProvider = MetadataCacheControllerNodeProvider(config, metadataCache)

val channelManager = BrokerToControllerChannelManager(
controllerNodeProvider = nodeProvider,
time = time,
metrics = metrics,
config = config,
channelName = "alterPartition",
threadNamePrefix = threadNamePrefix,
retryTimeoutMs = Long.MaxValue
)
new DefaultAlterPartitionManager(
controllerChannelManager = channelManager,
scheduler = scheduler,
Expand Down
32 changes: 22 additions & 10 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 @@ -129,7 +129,7 @@ class BrokerServer(

var forwardingManager: ForwardingManager = null

var alterIsrManager: AlterPartitionManager = null
var alterPartitionManager: AlterPartitionManager = null

var autoTopicCreationManager: AutoTopicCreationManager = null

Expand Down Expand Up @@ -250,15 +250,15 @@ class BrokerServer(
threadNamePrefix,
retryTimeoutMs = Long.MaxValue
)
alterIsrManager = new DefaultAlterPartitionManager(
alterPartitionManager = new DefaultAlterPartitionManager(
controllerChannelManager = alterIsrChannelManager,
scheduler = kafkaScheduler,
time = time,
brokerId = config.nodeId,
brokerEpochSupplier = () => lifecycleManager.brokerEpoch,
metadataVersionSupplier = () => metadataCache.metadataVersion()
)
alterIsrManager.start()
alterPartitionManager.start()

this._replicaManager = new ReplicaManager(
config = config,
Expand All @@ -269,7 +269,7 @@ class BrokerServer(
quotaManagers = quotaManagers,
metadataCache = metadataCache,
logDirFailureChannel = logDirFailureChannel,
alterPartitionManager = alterIsrManager,
alterPartitionManager = alterPartitionManager,
brokerTopicStats = brokerTopicStats,
isShuttingDown = isShuttingDown,
zkClient = None,
Expand Down Expand Up @@ -343,10 +343,22 @@ class BrokerServer(
k -> VersionRange.of(v.min, v.max)
}.asJava

lifecycleManager.start(() => metadataListener.highestMetadataOffset,
BrokerToControllerChannelManager(controllerNodeProvider, time, metrics, config,
"heartbeat", threadNamePrefix, config.brokerSessionTimeoutMs.toLong),
metaProps.clusterId, networkListeners, featuresRemapped)
val brokerLifecycleChannelManager = BrokerToControllerChannelManager(
controllerNodeProvider,
time,
metrics,
config,
"heartbeat",
threadNamePrefix,
config.brokerSessionTimeoutMs.toLong
)
lifecycleManager.start(
() => metadataListener.highestMetadataOffset,
brokerLifecycleChannelManager,
metaProps.clusterId,
networkListeners,
featuresRemapped
)

// Register a listener with the Raft layer to receive metadata event notifications
raftManager.register(metadataListener)
Expand Down Expand Up @@ -544,8 +556,8 @@ class BrokerServer(
if (replicaManager != null)
CoreUtils.swallow(replicaManager.shutdown(), this)

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

if (clientToControllerChannelManager != null)
CoreUtils.swallow(clientToControllerChannelManager.shutdown(), this)
Expand Down
6 changes: 1 addition & 5 deletions core/src/main/scala/kafka/server/BrokerToControllerChannelManager.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -165,7 +165,6 @@ class BrokerToControllerChannelManagerImpl(
private val logContext = new LogContext(s"[BrokerToControllerChannelManager broker=${config.brokerId} name=$channelName] ")
private val manualMetadataUpdater = new ManualMetadataUpdater()
private val apiVersions = new ApiVersions()
private val currentNodeApiVersions = NodeApiVersions.create()

@dengziming dengziming Feb 17, 2022

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

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

This is the source of the bug, we always assume currentNodeApiVersions=ApiKeys.zkBrokerApis

private val requestThread = newRequestThread

def start(): Unit = {
Expand Down Expand Up @@ -253,10 +252,7 @@ class BrokerToControllerChannelManagerImpl(

def controllerApiVersions(): Option[NodeApiVersions] = {
requestThread.activeControllerAddress().flatMap { activeController =>
if (activeController.id == config.brokerId)
Some(currentNodeApiVersions)
else
Option(apiVersions.get(activeController.idString))
Option(apiVersions.get(activeController.idString))
}
}
}
Expand Down
6 changes: 5 additions & 1 deletion core/src/main/scala/kafka/server/KafkaConfig.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1637,6 +1637,10 @@ class KafkaConfig private(doLog: Boolean, val props: java.util.Map[_, _], dynami
distinctRoles
}

def isKRaftCoResidentMode: Boolean = {
processRoles == Set(BrokerRole, ControllerRole)
}

def metadataLogDir: String = {
Option(getString(KafkaConfig.MetadataLogDirProp)) match {
case Some(dir) => dir
Expand Down Expand Up @@ -2164,7 +2168,7 @@ class KafkaConfig private(doLog: Boolean, val props: java.util.Map[_, _], dynami
validateControllerQuorumVotersMustContainNodeIdForKRaftController()
validateControllerListenerExistsForKRaftController()
validateControllerListenerNamesMustAppearInListenersForKRaftController()
} else if (processRoles == Set(BrokerRole, ControllerRole)) {
} else if (isKRaftCoResidentMode) {
// KRaft colocated broker and controller
validateNonEmptyQuorumVotersForKRaft()
validateControlPlaneListenerEmptyForKRaft()
Expand Down
31 changes: 21 additions & 10 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 @@ -140,7 +140,7 @@ class KafkaServer(

var clientToControllerChannelManager: BrokerToControllerChannelManager = null

var alterIsrManager: AlterPartitionManager = null
var alterPartitionManager: AlterPartitionManager = null

var kafkaScheduler: KafkaScheduler = null

Expand Down Expand Up @@ -228,6 +228,8 @@ class KafkaServer(
logContext = new LogContext(s"[KafkaServer id=${config.brokerId}] ")
this.logIdent = logContext.logPrefix

val controllerNodeProvider = MetadataCacheControllerNodeProvider(config, metadataCache)

// initialize dynamic broker configs from ZooKeeper. Any updates made after this will be
// applied after ZkConfigManager starts.
config.dynamicConfig.initialize(Some(zkClient))
Expand Down Expand Up @@ -276,13 +278,14 @@ class KafkaServer(
credentialProvider = new CredentialProvider(ScramMechanism.mechanismNames, tokenCache)

clientToControllerChannelManager = BrokerToControllerChannelManager(
controllerNodeProvider = MetadataCacheControllerNodeProvider(config, metadataCache),
controllerNodeProvider = controllerNodeProvider,
time = time,
metrics = metrics,
config = config,
channelName = "forwarding",
threadNamePrefix = threadNamePrefix,
retryTimeoutMs = config.requestTimeoutMs.longValue)
retryTimeoutMs = config.requestTimeoutMs.longValue
)
clientToControllerChannelManager.start()

/* start forwarding manager */
Expand All @@ -309,20 +312,28 @@ class KafkaServer(
socketServer = new SocketServer(config, metrics, time, credentialProvider, apiVersionManager)

// Start alter partition manager based on the IBP version
alterIsrManager = if (config.interBrokerProtocolVersion.isAlterPartitionSupported) {
alterPartitionManager = if (config.interBrokerProtocolVersion.isAlterPartitionSupported) {
val alterPartitionChannelManager = BrokerToControllerChannelManager(

@hachikuji hachikuji Jun 30, 2022

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.

nit: I think I favor creating the BrokerToControllerChannelManager in AlterPartitionManager.apply. The main thing is that it makes the ownership clearer. Currently DefaultAlterPartitionManager is responsible for starting and stopping the channel manager. We can revise BrokerServer to use the same apply method (I am not sure why we didn't do that).

@dengziming dengziming Jul 1, 2022

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

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

I guess we create BrokerToControllerChannelManager in BrokerServer because we want to consolidate these channel to use only one, #10135 (comment)
but it's it's still early to consolidate channel now so I moved it to AlterPartitionManager.apply as you suggested.

@hachikuji hachikuji Jul 1, 2022

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.

Consolidation is a little difficult because of head-of-line blocking (sigh). It might be reasonable to combine the alter partition and broker lifecycle channel managers though. Anyway, this is definitely a separate patch.

controllerNodeProvider,
time = time,
metrics = metrics,
config = config,
channelName = "alterIsr",
threadNamePrefix = threadNamePrefix,
retryTimeoutMs = Long.MaxValue
)
AlterPartitionManager(
config = config,
metadataCache = metadataCache,
alterPartitionChannelManager,
scheduler = kafkaScheduler,
time = time,
metrics = metrics,
threadNamePrefix = threadNamePrefix,
brokerEpochSupplier = () => kafkaController.brokerEpoch
)
} else {
AlterPartitionManager(kafkaScheduler, time, zkClient)
}
alterIsrManager.start()
alterPartitionManager.start()

// Start replica manager
_replicaManager = createReplicaManager(isShuttingDown)
Expand Down Expand Up @@ -478,7 +489,7 @@ class KafkaServer(
quotaManagers = quotaManagers,
metadataCache = metadataCache,
logDirFailureChannel = logDirFailureChannel,
alterPartitionManager = alterIsrManager,
alterPartitionManager = alterPartitionManager,
brokerTopicStats = brokerTopicStats,
isShuttingDown = isShuttingDown,
zkClient = Some(zkClient),
Expand Down Expand Up @@ -755,8 +766,8 @@ class KafkaServer(
if (replicaManager != null)
CoreUtils.swallow(replicaManager.shutdown(), this)

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

if (clientToControllerChannelManager != null)
CoreUtils.swallow(clientToControllerChannelManager.shutdown(), this)
Expand Down
4 changes: 4 additions & 0 deletions core/src/test/java/kafka/test/ClusterTestExtensionsTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -90,6 +90,10 @@ public void testClusterTemplate() {
@ClusterTest(name = "cluster-tests-2", clusterType = Type.KRAFT, serverProperties = {
@ClusterConfigProperty(key = "foo", value = "baz"),
@ClusterConfigProperty(key = "spam", value = "eggz")
}),
@ClusterTest(name = "cluster-tests-3", clusterType = Type.CO_KRAFT, serverProperties = {
@ClusterConfigProperty(key = "foo", value = "baz"),
@ClusterConfigProperty(key = "spam", value = "eggz")
})
})
public void testClusterTests() {
Expand Down
13 changes: 10 additions & 3 deletions core/src/test/java/kafka/test/annotation/Type.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -31,7 +31,13 @@ public enum Type {
KRAFT {
@Override
public void invocationContexts(ClusterConfig config, Consumer<TestTemplateInvocationContext> invocationConsumer) {
invocationConsumer.accept(new RaftClusterInvocationContext(config.copyOf()));
invocationConsumer.accept(new RaftClusterInvocationContext(config.copyOf(), false));
}
},
CO_KRAFT {
@Override
public void invocationContexts(ClusterConfig config, Consumer<TestTemplateInvocationContext> invocationConsumer) {
invocationConsumer.accept(new RaftClusterInvocationContext(config.copyOf(), true));
}
},
ZK {
Expand All @@ -40,10 +46,11 @@ public void invocationContexts(ClusterConfig config, Consumer<TestTemplateInvoca
invocationConsumer.accept(new ZkClusterInvocationContext(config.copyOf()));
}
},
BOTH {
ALL {
@Override
public void invocationContexts(ClusterConfig config, Consumer<TestTemplateInvocationContext> invocationConsumer) {
invocationConsumer.accept(new RaftClusterInvocationContext(config.copyOf()));
invocationConsumer.accept(new RaftClusterInvocationContext(config.copyOf(), false));
invocationConsumer.accept(new RaftClusterInvocationContext(config.copyOf(), true));
invocationConsumer.accept(new ZkClusterInvocationContext(config.copyOf()));
}
},
Expand Down
11 changes: 7 additions & 4 deletions core/src/test/java/kafka/test/junit/RaftClusterInvocationContext.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -65,18 +65,20 @@ public class RaftClusterInvocationContext implements TestTemplateInvocationConte

private final ClusterConfig clusterConfig;
private final AtomicReference<KafkaClusterTestKit> clusterReference;
private final boolean isCoResident;

public RaftClusterInvocationContext(ClusterConfig clusterConfig) {
public RaftClusterInvocationContext(ClusterConfig clusterConfig, boolean isCoResident) {
this.clusterConfig = clusterConfig;
this.clusterReference = new AtomicReference<>();
this.isCoResident = isCoResident;
}

@Override
public String getDisplayName(int invocationIndex) {
String clusterDesc = clusterConfig.nameTags().entrySet().stream()
.map(Object::toString)
.collect(Collectors.joining(", "));
return String.format("[%d] Type=Raft, %s", invocationIndex, clusterDesc);
.map(Object::toString)
.collect(Collectors.joining(", "));
return String.format("[%d] Type=Raft-%s, %s", invocationIndex, isCoResident ? "CoReside" : "Distributed", clusterDesc);
}

@Override
Expand All @@ -86,6 +88,7 @@ public List<Extension> getAdditionalExtensions() {
(BeforeTestExecutionCallback) context -> {
TestKitNodes nodes = new TestKitNodes.Builder().
setBootstrapMetadataVersion(clusterConfig.metadataVersion()).
setCoResident(isCoResident).
setNumBrokerNodes(clusterConfig.numBrokers()).
setNumControllerNodes(clusterConfig.numControllers()).build();
nodes.brokerNodes().forEach((brokerId, brokerNode) -> {
Expand Down
Loading