Replace obsolete class KafkaServerStartable with KafkaServer

Author: sjvanrossum

buildSrc/src/main/groovy/org/apache/beam/gradle/BeamModulePlugin.groovy

@@ -842,6 +842,7 @@ class BeamModulePlugin implements Plugin<Project> {
         kafka_scala_2_12                            : "org.apache.kafka:kafka_2.12:$kafka_version",
         kafka_scala_2_13                            : "org.apache.kafka:kafka_2.13:$kafka_version",
         kafka_clients                               : "org.apache.kafka:kafka-clients:$kafka_version",
+        kafka_server                                : "org.apache.kafka:kafka-server:$kafka_version",
         log4j                                       : "log4j:log4j:1.2.17",
         log4j_over_slf4j                            : "org.slf4j:log4j-over-slf4j:$slf4j_version",
         log4j2_api                                  : "org.apache.logging.log4j:log4j-api:$log4j2_version",

runners/spark/spark_runner.gradle

@@ -289,6 +289,7 @@ dependencies {
   testImplementation library.java.avro
   testImplementation (spark_scala_version == '2.13' ? library.java.kafka_scala_2_13 : library.java.kafka_scala_2_12)
   testImplementation library.java.kafka_clients
+  testImplementation library.java.kafka_server
   testImplementation library.java.junit
   testImplementation library.java.mockito_core
   testImplementation "org.assertj:assertj-core:3.11.1"

runners/spark/src/test/java/org/apache/beam/runners/spark/translation/streaming/utils/EmbeddedKafkaCluster.java

@@ -29,7 +29,7 @@
 import java.util.Properties;
 import java.util.Random;
 import kafka.server.KafkaConfig;
-import kafka.server.KafkaServerStartable;
+import kafka.server.KafkaServer;
 import org.apache.zookeeper.server.NIOServerCnxnFactory;
 import org.apache.zookeeper.server.ServerCnxnFactory;
 import org.apache.zookeeper.server.ZooKeeperServer;
@@ -47,7 +47,7 @@ public class EmbeddedKafkaCluster {
 
   private final String brokerList;
 
-  private final List<KafkaServerStartable> brokers;
+  private final List<KafkaServer> brokers;
   private final List<File> logDirs;
 
   private EmbeddedKafkaCluster(String zkConnection) {
@@ -114,15 +114,20 @@ public void startup() {
       properties.setProperty("offsets.topic.replication.factor", "1");
       properties.setProperty("log.flush.interval.messages", String.valueOf(1));
 
-      KafkaServerStartable broker = startBroker(properties);
+      KafkaServer broker = startBroker(properties);
 
       brokers.add(broker);
       logDirs.add(logDir);
     }
   }
 
-  private static KafkaServerStartable startBroker(Properties props) {
-    KafkaServerStartable server = new KafkaServerStartable(new KafkaConfig(props));
+  private static KafkaServer startBroker(Properties props) {
+    KafkaServer server =
+        new KafkaServer(
+            new KafkaConfig(props),
+            KafkaServer.$lessinit$greater$default$2(),
+            KafkaServer.$lessinit$greater$default$3(),
+            KafkaServer.$lessinit$greater$default$4());
     server.startup();
     return server;
   }
@@ -148,7 +153,7 @@ public String getZkConnection() {
 
   @SuppressWarnings("Slf4jDoNotLogMessageOfExceptionExplicitly")
   public void shutdown() {
-    for (KafkaServerStartable broker : brokers) {
+    for (KafkaServer broker : brokers) {
       try {
         broker.shutdown();
       } catch (Exception e) {

sdks/java/testing/kafka-service/src/test/java/org/apache/beam/sdk/testing/kafka/LocalKafka.java

@@ -20,18 +20,23 @@
 import java.nio.file.Files;
 import java.util.Properties;
 import kafka.server.KafkaConfig;
-import kafka.server.KafkaServerStartable;
+import kafka.server.KafkaServer;
 
 public class LocalKafka {
-  private final KafkaServerStartable server;
+  private final KafkaServer server;
 
   LocalKafka(int kafkaPort, int zookeeperPort) throws Exception {
     Properties kafkaProperties = new Properties();
     kafkaProperties.setProperty("port", String.valueOf(kafkaPort));
     kafkaProperties.setProperty("zookeeper.connect", String.format("localhost:%s", zookeeperPort));
     kafkaProperties.setProperty("offsets.topic.replication.factor", "1");
     kafkaProperties.setProperty("log.dir", Files.createTempDirectory("kafka-log-").toString());
-    server = new KafkaServerStartable(KafkaConfig.fromProps(kafkaProperties));
+    server =
+        new KafkaServer(
+            KafkaConfig.fromProps(kafkaProperties),
+            KafkaServer.$lessinit$greater$default$2(),
+            KafkaServer.$lessinit$greater$default$3(),
+            KafkaServer.$lessinit$greater$default$4());
   }
 
   public void start() {