[BEAM-9869] adding self-contained Kafka service jar for testing

sdks/java/testing/kafka-service/build.gradle

@@ -0,0 +1,61 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * License); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an AS IS BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+import com.github.jengelman.gradle.plugins.shadow.tasks.ShadowJar
+
+plugins { id 'org.apache.beam.module' }
+applyJavaNature(automaticModuleName: 'org.apache.beam.sdk.testing.kafka')
+
+description = "Apache Beam :: SDKs :: Java :: Test Kafka Service"
+ext.summary = """Self-contained Kafka service for testing IO transforms."""
+
+
+dependencies {
+ testCompile library.java.kafka
+ testCompile "org.apache.zookeeper:zookeeper:3.5.6"
+ testRuntimeOnly library.java.slf4j_log4j12
+}
+
+task runTestKafkaService (type: JavaExec) {
+ main = "org.apache.beam.sdk.testing.kafka.LocalKafka"
+ classpath = sourceSets.test.runtimeClasspath
+ args = [
+ project.findProperty("kafka.port") ?: "9092",
+ project.findProperty("zookeeper.port") ?: "2181"
+ ]
+}
+
+task buildTestKafkaServiceJar(type: ShadowJar) {
+ appendix = "testKafkaService"
+ // Use zip64 mode to avoid "Archive contains more than 65535 entries".
+ zip64 = true
+ mergeServiceFiles()
+ manifest {
+ attributes(
+ 'Main-Class': 'org.apache.beam.sdk.testing.kafka.LocalKafka'
+ )
+ }
+ exclude "META-INF/INDEX.LIST"
+ exclude "META-INF/*.SF"
+ exclude "META-INF/*.DSA"
+ exclude "META-INF/*.RSA"
+ configurations = [
+ project.configurations.testRuntime
+ ]
+ from sourceSets.main.output
+ from sourceSets.test.output
+}

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

@@ -0,0 +1,59 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.beam.sdk.testing.kafka;
+
+import java.nio.file.Files;
+import java.util.Properties;
+import kafka.server.KafkaConfig;
+import kafka.server.KafkaServerStartable;
+
+public class LocalKafka {
+ private final KafkaServerStartable 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));
+ }
+
+ public void start() {
+ server.startup();
+ }
+
+ public void stop() {
+ server.shutdown();
+ }
+
+ public void awaitTermination() {
+ server.awaitShutdown();
+ }
+
+ public static void main(String[] args) throws Exception {
+ int kafkaPort = Integer.parseInt(args[0]);
+ int zookeeperPort = Integer.parseInt(args[1]);
+ LocalZookeeper zookeeper = new LocalZookeeper(zookeeperPort);
+ LocalKafka kafka = new LocalKafka(kafkaPort, zookeeperPort);
+ zookeeper.start();
+ Thread.sleep(5000);
+ kafka.start();
+ kafka.awaitTermination();
+ }
+}

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

@@ -0,0 +1,56 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.beam.sdk.testing.kafka;
+
+import java.nio.file.Files;
+import java.util.Properties;
+import java.util.concurrent.Executor;
+import java.util.concurrent.Executors;
+import org.apache.zookeeper.server.ServerConfig;
+import org.apache.zookeeper.server.ZooKeeperServerMain;
+import org.apache.zookeeper.server.quorum.QuorumPeerConfig;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class LocalZookeeper {
+ private static final Logger log = LoggerFactory.getLogger(LocalZookeeper.class);
+ private final ServerConfig serverConfig;
+ private final Executor executor;
+
+ LocalZookeeper(int port) throws Exception {
+ Properties localProperties = new Properties();
+ localProperties.setProperty("clientPort", String.valueOf(port));
+ localProperties.setProperty("dataDir", Files.createTempDirectory("zookeeper-").toString());
+ QuorumPeerConfig config = new QuorumPeerConfig();
+ config.parseProperties(localProperties);
+ serverConfig = new ServerConfig();
+ serverConfig.readFrom(config);
+ executor = Executors.newSingleThreadExecutor();
+ }
+
+ public void start() {
+ executor.execute(
+ () -> {
+ try {
+ new ZooKeeperServerMain().runFromConfig(serverConfig);
+ } catch (Exception e) {
+ log.error("local zookeeper failure.", e);
+ }
+ });
+ }
+}

sdks/java/testing/kafka-service/src/test/resources/log4j.properties

@@ -0,0 +1,23 @@
+################################################################################
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# https://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+################################################################################
+
+log4j.rootLogger=ERROR,console
+log4j.appender.console=org.apache.log4j.ConsoleAppender
+log4j.appender.console.target=System.err
+log4j.appender.console.layout=org.apache.log4j.PatternLayout
+log4j.appender.console.layout.ConversionPattern=%d{yy/MM/dd HH:mm:ss} %p %c{2}: %m%n

settings.gradle

@@ -131,6 +131,7 @@ include ":sdks:java:maven-archetypes:examples"
 include ":sdks:java:maven-archetypes:starter"
 include ":sdks:java:testing:nexmark"
 include ":sdks:java:testing:expansion-service"
+include ":sdks:java:testing:kafka-service"
 include ":sdks:python"
 include ":sdks:python:apache_beam:testing:load_tests"
 include ":sdks:python:container"