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 3 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
18 changes: 3 additions & 15 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,25 +76,14 @@ object AlterPartitionManager {
def apply(
config: KafkaConfig,
metadataCache: MetadataCache,
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 alterPartition is redundant here. Could we call it channelManager?

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,
controllerChannelManager = alterPartitionChannelManager,
scheduler = scheduler,
time = time,
brokerId = config.brokerId,
Expand Down
47 changes: 35 additions & 12 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 @@ -34,11 +34,14 @@ import kafka.server.KafkaRaftServer.ControllerRole
import kafka.server.metadata.BrokerServerMetrics
import kafka.server.metadata.{BrokerMetadataListener, BrokerMetadataPublisher, BrokerMetadataSnapshotter, ClientQuotaMetadataManager, KRaftMetadataCache, SnapshotWriterBuilder}
import kafka.utils.{CoreUtils, KafkaScheduler}
import org.apache.kafka.clients.NodeApiVersions
import org.apache.kafka.common.feature.SupportedVersionRange
import org.apache.kafka.common.message.ApiMessageType.ListenerType
import org.apache.kafka.common.message.BrokerRegistrationRequestData.{Listener, ListenerCollection}
import org.apache.kafka.common.metrics.Metrics
import org.apache.kafka.common.network.ListenerName
import org.apache.kafka.common.protocol.ApiKeys
import org.apache.kafka.common.requests.ApiVersionsResponse
import org.apache.kafka.common.security.auth.SecurityProtocol
import org.apache.kafka.common.security.scram.internals.ScramMechanism
import org.apache.kafka.common.security.token.delegation.internals.DelegationTokenCache
Expand Down Expand Up @@ -129,7 +132,7 @@ class BrokerServer(

var forwardingManager: ForwardingManager = null

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

var autoTopicCreationManager: AutoTopicCreationManager = null

Expand Down Expand Up @@ -212,6 +215,11 @@ class BrokerServer(

val controllerNodes = RaftConfig.voterConnectionsToNodes(controllerQuorumVotersFuture.get()).asScala
val controllerNodeProvider = RaftControllerNodeProvider(raftManager, config, controllerNodes)
val currentNodeControllerApiVersions = if (config.isKRaftCoResidentMode) {
Some(NodeApiVersions.create(ApiKeys.controllerApis().asScala.map(ApiVersionsResponse.toApiVersion).asJava))
} else {
None

@jolshan jolshan Feb 22, 2022

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

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

Unless I'm misreading the code, it seems like this was not None before. Are there implications for adding the None case?

@dengziming dengziming Feb 23, 2022 •
edited
Loading

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.

The NodeApiVersions is used in BrokerToControllerChannelManager to get controller ApiVersion only if the current node is both the broker and the controller(which only happen in KRaft co-resident mode or zk mode), otherwise it will connect the controller and send ApiversionRequest to get it. in other cases, it is not used so I just set it to None, and it will make no difference if I change it to Some(controllerApiVersion)

}

clientToControllerChannelManager = BrokerToControllerChannelManager(
controllerNodeProvider,
Expand All @@ -220,7 +228,8 @@ class BrokerServer(
config,
channelName = "forwarding",
threadNamePrefix,
retryTimeoutMs = 60000
retryTimeoutMs = 60000,
currentNodeControllerApiVersions
)
clientToControllerChannelManager.start()
forwardingManager = new ForwardingManagerImpl(clientToControllerChannelManager)
Expand Down Expand Up @@ -248,17 +257,18 @@ class BrokerServer(
config,
channelName = "alterIsr",
threadNamePrefix,
retryTimeoutMs = Long.MaxValue
retryTimeoutMs = Long.MaxValue,
currentNodeControllerApiVersions
)
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 +279,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 +353,23 @@ 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,
currentNodeControllerApiVersions
)
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 +567,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
12 changes: 7 additions & 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 @@ -122,7 +122,8 @@ object BrokerToControllerChannelManager {
config: KafkaConfig,
channelName: String,
threadNamePrefix: Option[String],
retryTimeoutMs: Long
retryTimeoutMs: Long,
currentNodeControllerApiVersions: Option[NodeApiVersions]
): BrokerToControllerChannelManager = {
new BrokerToControllerChannelManagerImpl(
controllerNodeProvider,
Expand All @@ -131,7 +132,8 @@ object BrokerToControllerChannelManager {
config,
channelName,
threadNamePrefix,
retryTimeoutMs
retryTimeoutMs,
currentNodeControllerApiVersions
)
}
}
Expand Down Expand Up @@ -160,12 +162,12 @@ class BrokerToControllerChannelManagerImpl(
config: KafkaConfig,
channelName: String,
threadNamePrefix: Option[String],
retryTimeoutMs: Long
retryTimeoutMs: Long,
currentNodeControllerApiVersions: Option[NodeApiVersions]
) extends BrokerToControllerChannelManager with Logging {
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 @@ -254,7 +256,7 @@ class BrokerToControllerChannelManagerImpl(
def controllerApiVersions(): Option[NodeApiVersions] = {
requestThread.activeControllerAddress().flatMap { activeController =>
if (activeController.id == config.brokerId)
Some(currentNodeApiVersions)
currentNodeControllerApiVersions

@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.

I wonder if this optimization is worth preserving. Any harm using the same negotiation even when the controller is co-resident?

@dengziming dengziming Jun 30, 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 didn't find any harm in using the same negotiation so I removed this optimization here. Yet we should notice one thing here, the NetworkClient will only send ApiversionRequest to a broker when connected to it and will not refresh it before reconnecting, this assumption is no longer absolutely right since the broker will intersect controller apiVersion and controller apiVersion would change.

@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.

If I understand correctly, you are saying that the range of supported api versions may change while a client is still connected to the broker. In general it's only a problem if the range of supported versions is reduced (e.g. in a downgrade). The way the code is supposed to handle this is by catching the unsupported version in the Envelope response and disconnecting the client in order to ensure that it reconnects and gets the latest range of api version support. Check KafkaApis.handleInvalidVersionsDuringForwarding and see if that logic makes sense. I do not know if we have testing for this scenario.

@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.

Yes, there may be a similar problem with finalized features since they will also change, I will check KafkaApis.handleInvalidVersionsDuringForwarding to inspect it.

else
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
40 changes: 27 additions & 13 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 @@ -34,14 +34,14 @@ import kafka.security.CredentialProvider
import kafka.server.metadata.{ZkConfigRepository, ZkMetadataCache}
import kafka.utils._
import kafka.zk.{AdminZkClient, BrokerInfo, KafkaZkClient}
import org.apache.kafka.clients.{ApiVersions, ManualMetadataUpdater, NetworkClient, NetworkClientUtils}
import org.apache.kafka.clients.{ApiVersions, ManualMetadataUpdater, NetworkClient, NetworkClientUtils, NodeApiVersions}
import org.apache.kafka.common.internals.Topic
import org.apache.kafka.common.message.ApiMessageType.ListenerType
import org.apache.kafka.common.message.ControlledShutdownRequestData
import org.apache.kafka.common.metrics.Metrics
import org.apache.kafka.common.network._
import org.apache.kafka.common.protocol.Errors
import org.apache.kafka.common.requests.{ControlledShutdownRequest, ControlledShutdownResponse}
import org.apache.kafka.common.protocol.{ApiKeys, Errors}
import org.apache.kafka.common.requests.{ApiVersionsResponse, ControlledShutdownRequest, ControlledShutdownResponse}
import org.apache.kafka.common.security.scram.internals.ScramMechanism
import org.apache.kafka.common.security.token.delegation.internals.DelegationTokenCache
import org.apache.kafka.common.security.{JaasContext, JaasUtils}
Expand Down 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,9 @@ class KafkaServer(
logContext = new LogContext(s"[KafkaServer id=${config.brokerId}] ")
this.logIdent = logContext.logPrefix

val controllerNodeProvider = MetadataCacheControllerNodeProvider(config, metadataCache)
val currentNodeControllerApiVersions = Some(NodeApiVersions.create(ApiKeys.zkBrokerApis.asScala.map(ApiVersionsResponse.toApiVersion).asJava))

// 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 +279,15 @@ 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,
currentNodeControllerApiVersions
)
clientToControllerChannelManager.start()

/* start forwarding manager */
Expand All @@ -309,20 +314,29 @@ 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,
currentNodeControllerApiVersions
)
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 +492,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 +769,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