hachikuji commented on code in PR #12837:
URL: https://github.com/apache/kafka/pull/12837#discussion_r1036347804
##########
raft/src/main/java/org/apache/kafka/raft/KafkaRaftClient.java:
##########
@@ -2368,6 +2368,10 @@ public void close() {
if (kafkaRaftMetrics != null) {
kafkaRaftMetrics.close();
}
+ if (memoryPool instanceof BatchMemoryPool) {
Review Comment:
Perhaps we can let `MemoryPool` implement `Closeable`?
By the way, was this a leak? I think the only reference to the pool is in
`KafkaRaftClient`, so does that mean we were leaking `KafkaRaftClient`
references?
##########
core/src/test/java/kafka/testkit/KafkaClusterTestKit.java:
##########
@@ -214,61 +249,30 @@ metaProperties, config, new MetadataRecordSerde(),
metadataPartition, KafkaRaftS
connectFutureManager.registerPort(node.id(), port);
}
});
- raftManagers.put(node.id(), raftManager);
+ jointServers.put(node.id(), jointServer);
}
for (BrokerNode node : nodes.brokerNodes().values()) {
- Map<String, String> props = new HashMap<>(configProps);
- props.put(KafkaConfig$.MODULE$.ProcessRolesProp(),
roles(node.id()));
- props.put(KafkaConfig$.MODULE$.BrokerIdProp(),
- Integer.toString(node.id()));
- props.put(KafkaConfig$.MODULE$.MetadataLogDirProp(),
- node.metadataDirectory());
- props.put(KafkaConfig$.MODULE$.LogDirsProp(),
- String.join(",", node.logDataDirectories()));
-
props.put(KafkaConfig$.MODULE$.ListenerSecurityProtocolMapProp(),
- "EXTERNAL:PLAINTEXT,CONTROLLER:PLAINTEXT");
- props.put(KafkaConfig$.MODULE$.ListenersProp(),
listeners(node.id()));
-
props.put(KafkaConfig$.MODULE$.InterBrokerListenerNameProp(),
- nodes.interBrokerListenerName().value());
-
props.put(KafkaConfig$.MODULE$.ControllerListenerNamesProp(),
- "CONTROLLER");
-
- setupNodeDirectories(baseDirectory,
node.metadataDirectory(),
- node.logDataDirectories());
-
- // Just like above, we set a placeholder voter list here
until we
- // find out what ports the controllers picked.
- props.put(RaftConfig.QUORUM_VOTERS_CONFIG,
uninitializedQuorumVotersString);
- props.putAll(node.propertyOverrides());
- KafkaConfig config = new KafkaConfig(props, false,
Option.empty());
-
- String threadNamePrefix = String.format("broker%d_",
node.id());
- MetaProperties metaProperties =
MetaProperties.apply(nodes.clusterId().toString(), node.id());
- TopicPartition metadataPartition = new
TopicPartition(KafkaRaftServer.MetadataTopic(), 0);
- KafkaRaftManager<ApiMessageAndVersion> raftManager;
- if (raftManagers.containsKey(node.id())) {
- raftManager = raftManagers.get(node.id());
- } else {
- raftManager = new KafkaRaftManager<>(
- metaProperties, config, new MetadataRecordSerde(),
metadataPartition, KafkaRaftServer.MetadataTopicId(),
- Time.SYSTEM, new Metrics(),
Option.apply(threadNamePrefix), connectFutureManager.future);
- raftManagers.put(node.id(), raftManager);
+ JointServer jointServer = jointServers.get(node.id());
+ if (jointServer == null) {
+ jointServer = new JointServer(createNodeConfig(node),
+
MetaProperties.apply(nodes.clusterId().toString(), node.id()),
+ Time.SYSTEM,
+ new Metrics(),
+ Option.apply(String.format("broker%d_",
node.id())),
+ connectFutureManager.future,
+ faultHandlerFactory);
+ jointServers.put(node.id(), jointServer);
+ }
+ BrokerServer broker = null;
+ try {
+ broker = new BrokerServer(
+ jointServer,
+
JavaConverters.asScalaBuffer(Collections.<String>emptyList()).toSeq());
+ } catch (Throwable e) {
+ log.error("Error creating broker {}", node.id(), e);
+ if (broker != null) broker.shutdown();
Review Comment:
nit: similar comment as before. I think JointServer still has resources that
need to be cleaned up even if it doesn't get started (e.g. `Metrics`). Same
thing for the controller above.
##########
core/src/main/scala/kafka/server/BrokerServer.scala:
##########
@@ -72,19 +67,14 @@ class BrokerSnapshotWriterBuilder(raftClient:
RaftClient[ApiMessageAndVersion])
* A Kafka broker that runs in KRaft (Kafka Raft) mode.
*/
class BrokerServer(
- val config: KafkaConfig,
- val metaProps: MetaProperties,
- val raftManager: RaftManager[ApiMessageAndVersion],
- val time: Time,
- val metrics: Metrics,
- val brokerMetrics: BrokerServerMetrics,
- val threadNamePrefix: Option[String],
+ val jointServer: JointServer,
val initialOfflineDirs: Seq[String],
- val controllerQuorumVotersFuture: CompletableFuture[util.Map[Integer,
AddressSpec]],
- val fatalFaultHandler: FaultHandler,
- val metadataLoadingFaultHandler: FaultHandler,
- val metadataPublishingFaultHandler: FaultHandler
) extends KafkaBroker {
+ val threadNamePrefix = jointServer.threadNamePrefix
+ val config = jointServer.config
+ val time = jointServer.time
+ val metrics = jointServer.metrics
+ def raftManager: KafkaRaftManager[ApiMessageAndVersion] =
jointServer.raftManager
Review Comment:
nit: why not make this a `val` also?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscr...@kafka.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
