java - 从 Java 创建 `KafkaServer`

Question

我正在尝试从 Java 启动 Kafka 服务器

具体来说，我该如何翻译this line Scala 的代码转换为 Java 代码？

  private val server = new KafkaServer(serverConfig, kafkaMetricsReporters = reporters)

我可以轻松创建 serverConfig，但我似乎无法创建 kafkaMetricsReporters 参数。

注意:我可以创建一个 KafkaServerStartable，但我想创建一个普通的 KafkaServer 以避免 JVM 在出现错误时退出。

Apache 卡夫卡版本0.11.0.1

Answer 1

kafkaMetricsReporters参数是一个 scala Seq 。

您可以:

1. 创建 Java 集合并将其转换为 Seq:

   您需要导入scala.collection.JavaConverters :

   List<KafkaMetricsReporter> reportersList = new ArrayList<>();
   ...
   Seq<KafkaMetricsReporter> reportersSeq = JavaConverters.asScalaBufferConverter(reportersList).asScala();
   

2. 使用KafkaMetricsReporter.startReporters()从您的配置中为您创建它们的方法:

   如KafkaMetricsReporter是单例，需要使用MODULE使用它的符号:

   Seq<KafkaMetricsReporter> reporters = KafkaMetricsReporter$.MODULE$.startReporters(new VerifiableProperties(props));
   

还有KafkaServer构造函数还有 2 个从 Java 调用时需要的其他参数:

• time可以使用 new org.apache.kafka.common.utils.SystemTime() 轻松创建
• threadNamePrefix是一个选项。如果您导入scala.Option ，您可以调用Option.apply("prefix")

把它们放在一起:

Properties props = new Properties();
props.put(...);
KafkaConfig config = KafkaConfig.fromProps(props);
Seq<KafkaMetricsReporter> reporters = KafkaMetricsReporter$.MODULE$.startReporters(new VerifiableProperties(props));
KafkaServer server = new KafkaServer(config, new SystemTime(), Option.apply("prefix"), reporters);
server.startup();