cmccabe commented on code in PR #12837:
URL: https://github.com/apache/kafka/pull/12837#discussion_r1036532484
##########
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:
I think `Metrics` is the only one. It looks like its constructor starts a
thread and so on.
But yes, we should shut down all that. Fixed.
--
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
