从 Java 创建 `KafkaServer`

Question

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

具体来说，如何将Scala的this line翻译成一行Java？

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

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

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

Apache Kafka 版本 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();