diff --git a/CHANGES.md b/CHANGES.md index ca911e52a7ad..694f8163d03a 100644 --- a/CHANGES.md +++ b/CHANGES.md @@ -71,6 +71,7 @@ ## New Features / Improvements +* (Java) Added a `runners/kafka-streams` Gradle module with portable job server and runner entry points; translation fails fast with an explicit unsupported-URN message until transforms are implemented ([#38465](https://github.com/apache/beam/issues/38465)). * Capability introduces an indicator for aggregations and timers firing during a pipeline drain, allowing users and sinks to recognize and appropriately handle potentially incomplete or partial data ([#36884](https://github.com/apache/beam/issues/36884)). * Added support for setting disk provisioned IOPS and throughput in Dataflow runner via `--diskProvisionedIops` and `--diskProvisionedThroughputMibps` pipeline options (Java/Go/Python) ([#38349](https://github.com/apache/beam/issues/38349)). * TriggerStateMachineRunner changes from BitSetCoder to SentinelBitSetCoder to diff --git a/build.gradle.kts b/build.gradle.kts index 4af8fa3f1ab4..b3b9fdd7fdf0 100644 --- a/build.gradle.kts +++ b/build.gradle.kts @@ -278,6 +278,7 @@ tasks.register("javaPreCommit") { dependsOn(":runners:java-fn-execution:build") dependsOn(":runners:java-job-service:build") dependsOn(":runners:jet:build") + dependsOn(":runners:kafka-streams:build") dependsOn(":runners:local-java:build") dependsOn(":runners:portability:java:build") dependsOn(":runners:prism:java:build") diff --git a/runners/kafka-streams/build.gradle b/runners/kafka-streams/build.gradle new file mode 100644 index 000000000000..54474502ad7b --- /dev/null +++ b/runners/kafka-streams/build.gradle @@ -0,0 +1,62 @@ +/* + * 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 + * + * http://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. + */ + +plugins { id 'org.apache.beam.module' } + +def kafka_version = '3.9.0' + +applyJavaNature( + automaticModuleName: 'org.apache.beam.runners.kafka.streams', +) + +description = "Apache Beam :: Runners :: Kafka Streams" + +evaluationDependsOn(":sdks:java:core") +evaluationDependsOn(":runners:core-java") + +configurations.configureEach { + resolutionStrategy.eachDependency { details -> + if (details.requested.group == "org.apache.kafka") { + details.useVersion(kafka_version) + details.because("Kafka Streams runner is developed against Kafka ${kafka_version}.") + } + } +} + +dependencies { + compileOnly project(":sdks:java:build-tools") + permitUnusedDeclared project(":sdks:java:build-tools") + + implementation project(path: ":sdks:java:core", configuration: "shadow") + implementation project(path: ":model:pipeline", configuration: "shadow") + implementation project(":runners:core-java") + permitUnusedDeclared project(":runners:core-java") + implementation project(":runners:java-fn-execution") + implementation project(":runners:java-job-service") + implementation project(":runners:portability:java") + implementation project(path: ":sdks:java:extensions:google-cloud-platform-core") + implementation library.java.args4j + implementation library.java.joda_time + implementation library.java.slf4j_api + implementation library.java.vendored_grpc_1_69_0 + implementation library.java.vendored_guava_32_1_2_jre + implementation "org.apache.kafka:kafka-clients:$kafka_version" + implementation "org.apache.kafka:kafka-streams:$kafka_version" + permitUnusedDeclared "org.apache.kafka:kafka-clients:$kafka_version" + permitUnusedDeclared "org.apache.kafka:kafka-streams:$kafka_version" +} diff --git a/runners/kafka-streams/src/main/java/org/apache/beam/runners/kafka/streams/KafkaStreamsJobInvoker.java b/runners/kafka-streams/src/main/java/org/apache/beam/runners/kafka/streams/KafkaStreamsJobInvoker.java new file mode 100644 index 000000000000..bd1ed4c1bcf6 --- /dev/null +++ b/runners/kafka-streams/src/main/java/org/apache/beam/runners/kafka/streams/KafkaStreamsJobInvoker.java @@ -0,0 +1,99 @@ +/* + * 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 + * + * http://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.runners.kafka.streams; + +import java.util.UUID; +import org.apache.beam.model.pipeline.v1.RunnerApi; +import org.apache.beam.runners.fnexecution.provisioning.JobInfo; +import org.apache.beam.runners.jobsubmission.JobInvocation; +import org.apache.beam.runners.jobsubmission.JobInvoker; +import org.apache.beam.runners.jobsubmission.PortablePipelineRunner; +import org.apache.beam.sdk.util.construction.PipelineOptionsTranslation; +import org.apache.beam.vendor.grpc.v1p69p0.com.google.protobuf.Struct; +import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.util.concurrent.ListeningExecutorService; +import org.checkerframework.checker.nullness.qual.Nullable; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** Job invoker for the Kafka Streams portable runner. */ +@SuppressWarnings({ + "nullness" // TODO(https://github.com/apache/beam/issues/20497) +}) +public class KafkaStreamsJobInvoker extends JobInvoker { + + private static final Logger LOG = LoggerFactory.getLogger(KafkaStreamsJobInvoker.class); + + public static KafkaStreamsJobInvoker create( + KafkaStreamsJobServerDriver.KafkaStreamsServerConfiguration serverConfig) { + return new KafkaStreamsJobInvoker(serverConfig); + } + + private final KafkaStreamsJobServerDriver.KafkaStreamsServerConfiguration serverConfig; + + protected KafkaStreamsJobInvoker( + KafkaStreamsJobServerDriver.KafkaStreamsServerConfiguration serverConfig) { + super("kafka-streams-runner-job-invoker-%d"); + this.serverConfig = serverConfig; + } + + @Override + protected JobInvocation invokeWithExecutor( + RunnerApi.Pipeline pipeline, + Struct options, + @Nullable String retrievalToken, + ListeningExecutorService executorService) { + + LOG.trace( + "Parsing pipeline options (job server {}:{})", + serverConfig.getHost(), + serverConfig.getPort()); + KafkaStreamsPipelineOptions kafkaStreamsOptions = + PipelineOptionsTranslation.fromProto(options).as(KafkaStreamsPipelineOptions.class); + + String invocationId = + String.format("%s_%s", kafkaStreamsOptions.getJobName(), UUID.randomUUID().toString()); + + PortablePipelineRunner pipelineRunner = new KafkaStreamsPipelineRunner(kafkaStreamsOptions); + kafkaStreamsOptions.setRunner(null); + + LOG.info("Invoking job {} with pipeline runner {}", invocationId, pipelineRunner); + return createJobInvocation( + invocationId, + retrievalToken, + executorService, + pipeline, + kafkaStreamsOptions, + pipelineRunner); + } + + protected JobInvocation createJobInvocation( + String invocationId, + String retrievalToken, + ListeningExecutorService executorService, + RunnerApi.Pipeline pipeline, + KafkaStreamsPipelineOptions kafkaStreamsOptions, + PortablePipelineRunner pipelineRunner) { + JobInfo jobInfo = + JobInfo.create( + invocationId, + kafkaStreamsOptions.getJobName(), + retrievalToken, + PipelineOptionsTranslation.toProto(kafkaStreamsOptions)); + return new JobInvocation(jobInfo, executorService, pipeline, pipelineRunner); + } +} diff --git a/runners/kafka-streams/src/main/java/org/apache/beam/runners/kafka/streams/KafkaStreamsJobServerDriver.java b/runners/kafka-streams/src/main/java/org/apache/beam/runners/kafka/streams/KafkaStreamsJobServerDriver.java new file mode 100644 index 000000000000..ddeac8b7c959 --- /dev/null +++ b/runners/kafka-streams/src/main/java/org/apache/beam/runners/kafka/streams/KafkaStreamsJobServerDriver.java @@ -0,0 +1,109 @@ +/* + * 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 + * + * http://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.runners.kafka.streams; + +import org.apache.beam.runners.jobsubmission.JobServerDriver; +import org.apache.beam.sdk.extensions.gcp.options.GcsOptions; +import org.apache.beam.sdk.fn.server.ServerFactory; +import org.apache.beam.sdk.io.FileSystems; +import org.apache.beam.sdk.options.PipelineOptions; +import org.apache.beam.sdk.options.PipelineOptionsFactory; +import org.kohsuke.args4j.CmdLineException; +import org.kohsuke.args4j.CmdLineParser; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** Driver that starts a Beam job server for the Kafka Streams portable runner. */ +@SuppressWarnings({ + "nullness" // TODO(https://github.com/apache/beam/issues/20497) +}) +public class KafkaStreamsJobServerDriver extends JobServerDriver { + + private static final Logger LOG = LoggerFactory.getLogger(KafkaStreamsJobServerDriver.class); + + /** Runner-specific configuration for the job server process. */ + public static class KafkaStreamsServerConfiguration extends ServerConfiguration {} + + public static void main(String[] args) throws Exception { + PipelineOptions options = PipelineOptionsFactory.create(); + options.as(GcsOptions.class).setGcsUploadBufferSizeBytes(1024 * 1024); + FileSystems.setDefaultPipelineOptions(options); + fromParams(args).run(); + } + + private static void printUsage(CmdLineParser parser) { + System.err.println( + String.format( + "Usage: java %s arguments...", KafkaStreamsJobServerDriver.class.getSimpleName())); + parser.printUsage(System.err); + System.err.println(); + } + + public static KafkaStreamsServerConfiguration parseArgs(String[] args) { + KafkaStreamsServerConfiguration configuration = new KafkaStreamsServerConfiguration(); + CmdLineParser parser = new CmdLineParser(configuration); + try { + parser.parseArgument(args); + } catch (CmdLineException e) { + LOG.error("Unable to parse command line arguments.", e); + printUsage(parser); + throw new IllegalArgumentException("Unable to parse command line arguments.", e); + } + return configuration; + } + + /** Used by tests and tooling to construct a driver from command-line parameters. */ + public static KafkaStreamsJobServerDriver fromParams(String[] args) { + return fromConfig(parseArgs(args)); + } + + public static KafkaStreamsJobServerDriver fromConfig( + KafkaStreamsServerConfiguration configuration) { + return create( + configuration, + createJobServerFactory(configuration), + createArtifactServerFactory(configuration), + () -> KafkaStreamsJobInvoker.create(configuration)); + } + + public static KafkaStreamsJobServerDriver fromConfig( + KafkaStreamsServerConfiguration configuration, JobInvokerFactory jobInvokerFactory) { + return create( + configuration, + createJobServerFactory(configuration), + createArtifactServerFactory(configuration), + jobInvokerFactory); + } + + private static KafkaStreamsJobServerDriver create( + KafkaStreamsServerConfiguration configuration, + ServerFactory jobServerFactory, + ServerFactory artifactServerFactory, + JobInvokerFactory jobInvokerFactory) { + return new KafkaStreamsJobServerDriver( + configuration, jobServerFactory, artifactServerFactory, jobInvokerFactory); + } + + private KafkaStreamsJobServerDriver( + KafkaStreamsServerConfiguration configuration, + ServerFactory jobServerFactory, + ServerFactory artifactServerFactory, + JobInvokerFactory jobInvokerFactory) { + super(configuration, jobServerFactory, artifactServerFactory, jobInvokerFactory); + } +} diff --git a/runners/kafka-streams/src/main/java/org/apache/beam/runners/kafka/streams/KafkaStreamsPipelineOptions.java b/runners/kafka-streams/src/main/java/org/apache/beam/runners/kafka/streams/KafkaStreamsPipelineOptions.java new file mode 100644 index 000000000000..019b37cba770 --- /dev/null +++ b/runners/kafka-streams/src/main/java/org/apache/beam/runners/kafka/streams/KafkaStreamsPipelineOptions.java @@ -0,0 +1,76 @@ +/* + * 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 + * + * http://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.runners.kafka.streams; + +import java.nio.file.Paths; +import org.apache.beam.sdk.options.Default; +import org.apache.beam.sdk.options.DefaultValueFactory; +import org.apache.beam.sdk.options.Description; +import org.apache.beam.sdk.options.PipelineOptions; +import org.apache.beam.sdk.options.PortablePipelineOptions; + +/** Pipeline options for the Kafka Streams runner. */ +public interface KafkaStreamsPipelineOptions extends PortablePipelineOptions { + + @Description("Comma-separated list of host:port Kafka brokers used by the Kafka Streams client.") + @Default.String("localhost:9092") + String getBootstrapServers(); + + void setBootstrapServers(String bootstrapServers); + + @Description( + "Kafka Streams application.id (must be unique for each distinct topology using the same " + + "input topics in a Kafka cluster).") + @Default.String("beam-kafka-streams-runner") + String getApplicationId(); + + void setApplicationId(String applicationId); + + @Description( + "Kafka Streams processing.guarantee setting, for example at_least_once or exactly_once_v2.") + @Default.String("exactly_once_v2") + String getProcessingGuarantee(); + + void setProcessingGuarantee(String processingGuarantee); + + @Description("Soft cap on the number of elements per bundle.") + @Default.Integer(1000) + int getMaxBundleSize(); + + void setMaxBundleSize(int maxBundleSize); + + @Description("Soft cap on bundle wall-clock duration in milliseconds.") + @Default.Integer(1000) + int getMaxBundleTimeMs(); + + void setMaxBundleTimeMs(int maxBundleTimeMs); + + @Description("Directory where Kafka Streams stores local state.") + @Default.InstanceFactory(StateDirDefaultFactory.class) + String getStateDir(); + + void setStateDir(String stateDir); + + /** Default {@link #getStateDir()} under the JVM temp directory. */ + class StateDirDefaultFactory implements DefaultValueFactory { + @Override + public String create(PipelineOptions options) { + return Paths.get(System.getProperty("java.io.tmpdir"), "beam-kafka-streams-state").toString(); + } + } +} diff --git a/runners/kafka-streams/src/main/java/org/apache/beam/runners/kafka/streams/KafkaStreamsPipelineResult.java b/runners/kafka-streams/src/main/java/org/apache/beam/runners/kafka/streams/KafkaStreamsPipelineResult.java new file mode 100644 index 000000000000..776eaa6629f5 --- /dev/null +++ b/runners/kafka-streams/src/main/java/org/apache/beam/runners/kafka/streams/KafkaStreamsPipelineResult.java @@ -0,0 +1,69 @@ +/* + * 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 + * + * http://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.runners.kafka.streams; + +import java.io.IOException; +import org.apache.beam.sdk.PipelineResult; +import org.apache.beam.sdk.metrics.MetricResults; +import org.joda.time.Duration; + +/** + * Forwards {@link PipelineResult} calls to a delegate and stops an embedded job server when the + * pipeline reaches a terminal state. + */ +class KafkaStreamsPipelineResult implements PipelineResult { + + private final PipelineResult delegate; + private final Runnable stopJobServer; + + KafkaStreamsPipelineResult(PipelineResult delegate, Runnable stopJobServer) { + this.delegate = delegate; + this.stopJobServer = stopJobServer; + } + + @Override + public State getState() { + return delegate.getState(); + } + + @Override + public State cancel() throws IOException { + State state = delegate.cancel(); + stopJobServer.run(); + return state; + } + + @Override + public State waitUntilFinish(Duration duration) { + State state = delegate.waitUntilFinish(duration); + stopJobServer.run(); + return state; + } + + @Override + public State waitUntilFinish() { + State state = delegate.waitUntilFinish(); + stopJobServer.run(); + return state; + } + + @Override + public MetricResults metrics() { + return delegate.metrics(); + } +} diff --git a/runners/kafka-streams/src/main/java/org/apache/beam/runners/kafka/streams/KafkaStreamsPipelineRunner.java b/runners/kafka-streams/src/main/java/org/apache/beam/runners/kafka/streams/KafkaStreamsPipelineRunner.java new file mode 100644 index 000000000000..cc7464e22786 --- /dev/null +++ b/runners/kafka-streams/src/main/java/org/apache/beam/runners/kafka/streams/KafkaStreamsPipelineRunner.java @@ -0,0 +1,48 @@ +/* + * 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 + * + * http://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.runners.kafka.streams; + +import org.apache.beam.model.pipeline.v1.RunnerApi; +import org.apache.beam.runners.fnexecution.provisioning.JobInfo; +import org.apache.beam.runners.jobsubmission.PortablePipelineResult; +import org.apache.beam.runners.jobsubmission.PortablePipelineRunner; +import org.apache.beam.runners.kafka.streams.translation.KafkaStreamsPipelineTranslator; +import org.apache.beam.runners.kafka.streams.translation.KafkaStreamsTranslationContext; + +/** Executes a portable pipeline by translating it to Kafka Streams. */ +@SuppressWarnings({ + "nullness" // TODO(https://github.com/apache/beam/issues/20497) +}) +public class KafkaStreamsPipelineRunner implements PortablePipelineRunner { + + private final KafkaStreamsPipelineOptions pipelineOptions; + + public KafkaStreamsPipelineRunner(KafkaStreamsPipelineOptions pipelineOptions) { + this.pipelineOptions = pipelineOptions; + } + + @Override + public PortablePipelineResult run(RunnerApi.Pipeline pipeline, JobInfo jobInfo) throws Exception { + KafkaStreamsPipelineTranslator translator = new KafkaStreamsPipelineTranslator(); + KafkaStreamsTranslationContext context = + translator.createTranslationContext(jobInfo, pipelineOptions); + RunnerApi.Pipeline prepared = translator.prepareForTranslation(pipeline); + translator.translate(context, prepared); + throw new IllegalStateException("Translation unexpectedly completed without an executor"); + } +} diff --git a/runners/kafka-streams/src/main/java/org/apache/beam/runners/kafka/streams/KafkaStreamsRunner.java b/runners/kafka-streams/src/main/java/org/apache/beam/runners/kafka/streams/KafkaStreamsRunner.java new file mode 100644 index 000000000000..ce8f0544b681 --- /dev/null +++ b/runners/kafka-streams/src/main/java/org/apache/beam/runners/kafka/streams/KafkaStreamsRunner.java @@ -0,0 +1,101 @@ +/* + * 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 + * + * http://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.runners.kafka.streams; + +import java.io.IOException; +import java.util.ArrayList; +import java.util.List; +import org.apache.beam.runners.portability.PortableRunner; +import org.apache.beam.sdk.Pipeline; +import org.apache.beam.sdk.PipelineResult; +import org.apache.beam.sdk.PipelineRunner; +import org.apache.beam.sdk.options.ExperimentalOptions; +import org.apache.beam.sdk.options.PipelineOptions; +import org.apache.beam.sdk.util.construction.Environments; +import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.Strings; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * A {@link PipelineRunner} that submits portable jobs to an in-process or external Beam job service + * backed by the Kafka Streams translation path. + */ +@SuppressWarnings({ + "nullness" // TODO(https://github.com/apache/beam/issues/20497) +}) +public class KafkaStreamsRunner extends PipelineRunner { + + private static final Logger LOG = LoggerFactory.getLogger(KafkaStreamsRunner.class); + + private final KafkaStreamsPipelineOptions pipelineOptions; + + public static KafkaStreamsRunner fromOptions(PipelineOptions options) { + return new KafkaStreamsRunner(options.as(KafkaStreamsPipelineOptions.class)); + } + + protected KafkaStreamsRunner(KafkaStreamsPipelineOptions pipelineOptions) { + this.pipelineOptions = pipelineOptions; + } + + @Override + public PipelineResult run(Pipeline pipeline) { + assignPortableDefaults(pipelineOptions); + KafkaStreamsJobServerDriver jobServerDriver = null; + try { + if (Strings.isNullOrEmpty(pipelineOptions.getJobEndpoint())) { + LOG.info("No job endpoint configured; starting an embedded Kafka Streams job server."); + KafkaStreamsJobServerDriver.KafkaStreamsServerConfiguration configuration = + new KafkaStreamsJobServerDriver.KafkaStreamsServerConfiguration(); + configuration.setPort(0); + jobServerDriver = KafkaStreamsJobServerDriver.fromConfig(configuration); + pipelineOptions.setJobEndpoint(jobServerDriver.start()); + } + PortableRunner portableRunner = PortableRunner.fromOptions(pipelineOptions); + PipelineResult result = portableRunner.run(pipeline); + if (jobServerDriver != null) { + return new KafkaStreamsPipelineResult(result, jobServerDriver::stop); + } + return result; + } catch (IOException e) { + if (jobServerDriver != null) { + jobServerDriver.stop(); + } + throw new RuntimeException(e); + } + } + + private static void assignPortableDefaults(KafkaStreamsPipelineOptions pipelineOptions) { + if (Strings.isNullOrEmpty(pipelineOptions.getDefaultEnvironmentType())) { + pipelineOptions.setDefaultEnvironmentType(Environments.ENVIRONMENT_LOOPBACK); + } + ExperimentalOptions experimentalOptions = pipelineOptions.as(ExperimentalOptions.class); + List experiments = + experimentalOptions.getExperiments() == null + ? new ArrayList<>() + : new ArrayList<>(experimentalOptions.getExperiments()); + if (!experiments.contains("beam_fn_api")) { + experiments.add("beam_fn_api"); + experimentalOptions.setExperiments(experiments); + } + } + + @Override + public String toString() { + return "KafkaStreamsRunner#" + hashCode(); + } +} diff --git a/runners/kafka-streams/src/main/java/org/apache/beam/runners/kafka/streams/KafkaStreamsRunnerRegistrar.java b/runners/kafka-streams/src/main/java/org/apache/beam/runners/kafka/streams/KafkaStreamsRunnerRegistrar.java new file mode 100644 index 000000000000..ac3c64b97bb0 --- /dev/null +++ b/runners/kafka-streams/src/main/java/org/apache/beam/runners/kafka/streams/KafkaStreamsRunnerRegistrar.java @@ -0,0 +1,48 @@ +/* + * 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 + * + * http://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.runners.kafka.streams; + +import com.google.auto.service.AutoService; +import org.apache.beam.sdk.PipelineRunner; +import org.apache.beam.sdk.options.PipelineOptions; +import org.apache.beam.sdk.options.PipelineOptionsRegistrar; +import org.apache.beam.sdk.runners.PipelineRunnerRegistrar; +import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.ImmutableList; + +/** {@link com.google.auto.service.AutoService} registrations for the Kafka Streams runner. */ +public class KafkaStreamsRunnerRegistrar { + private KafkaStreamsRunnerRegistrar() {} + + /** Registers {@link KafkaStreamsRunner}. */ + @AutoService(PipelineRunnerRegistrar.class) + public static class Runner implements PipelineRunnerRegistrar { + @Override + public Iterable>> getPipelineRunners() { + return ImmutableList.of(KafkaStreamsRunner.class); + } + } + + /** Registers {@link KafkaStreamsPipelineOptions}. */ + @AutoService(PipelineOptionsRegistrar.class) + public static class Options implements PipelineOptionsRegistrar { + @Override + public Iterable> getPipelineOptions() { + return ImmutableList.of(KafkaStreamsPipelineOptions.class); + } + } +} diff --git a/runners/kafka-streams/src/main/java/org/apache/beam/runners/kafka/streams/package-info.java b/runners/kafka-streams/src/main/java/org/apache/beam/runners/kafka/streams/package-info.java new file mode 100644 index 000000000000..c9def4d1a4d7 --- /dev/null +++ b/runners/kafka-streams/src/main/java/org/apache/beam/runners/kafka/streams/package-info.java @@ -0,0 +1,20 @@ +/* + * 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 + * + * http://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. + */ + +/** Kafka Streams runner: portable pipeline execution backed by Apache Kafka Streams. */ +package org.apache.beam.runners.kafka.streams; diff --git a/runners/kafka-streams/src/main/java/org/apache/beam/runners/kafka/streams/translation/KafkaStreamsPipelineTranslator.java b/runners/kafka-streams/src/main/java/org/apache/beam/runners/kafka/streams/translation/KafkaStreamsPipelineTranslator.java new file mode 100644 index 000000000000..cc915d604b68 --- /dev/null +++ b/runners/kafka-streams/src/main/java/org/apache/beam/runners/kafka/streams/translation/KafkaStreamsPipelineTranslator.java @@ -0,0 +1,70 @@ +/* + * 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 + * + * http://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.runners.kafka.streams.translation; + +import java.util.Map; +import java.util.TreeMap; +import org.apache.beam.model.pipeline.v1.RunnerApi; +import org.apache.beam.runners.fnexecution.provisioning.JobInfo; +import org.apache.beam.runners.kafka.streams.KafkaStreamsPipelineOptions; + +/** + * Translates a portable Beam pipeline into a Kafka Streams {@code Topology}. + * + *

The initial implementation only validates the graph and fails fast with an explicit message + * for transforms that are not yet supported. + */ +public class KafkaStreamsPipelineTranslator { + + public KafkaStreamsTranslationContext createTranslationContext( + JobInfo jobInfo, KafkaStreamsPipelineOptions pipelineOptions) { + return KafkaStreamsTranslationContext.create(jobInfo, pipelineOptions); + } + + /** Returns the pipeline to translate (placeholder for future fusion / expansion steps). */ + public RunnerApi.Pipeline prepareForTranslation(RunnerApi.Pipeline pipeline) { + return pipeline; + } + + /** + * Translates the pipeline. Throws {@link UnsupportedOperationException} with a clear URN message + * for the first unsupported primitive encountered. + */ + public void translate(KafkaStreamsTranslationContext context, RunnerApi.Pipeline pipeline) { + Map transforms = pipeline.getComponents().getTransformsMap(); + TreeMap ordered = new TreeMap<>(transforms); + for (Map.Entry entry : ordered.entrySet()) { + RunnerApi.PTransform transform = entry.getValue(); + if (!transform.hasSpec()) { + continue; + } + String urn = transform.getSpec().getUrn(); + if (urn.isEmpty()) { + continue; + } + throw new UnsupportedOperationException( + "No translator registered for URN " + + urn + + " (jobId=" + + context.getJobInfo().jobId() + + ")"); + } + throw new UnsupportedOperationException( + "No translator registered for pipeline (no transform URNs found)"); + } +} diff --git a/runners/kafka-streams/src/main/java/org/apache/beam/runners/kafka/streams/translation/KafkaStreamsTranslationContext.java b/runners/kafka-streams/src/main/java/org/apache/beam/runners/kafka/streams/translation/KafkaStreamsTranslationContext.java new file mode 100644 index 000000000000..7c6d3d079159 --- /dev/null +++ b/runners/kafka-streams/src/main/java/org/apache/beam/runners/kafka/streams/translation/KafkaStreamsTranslationContext.java @@ -0,0 +1,47 @@ +/* + * 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 + * + * http://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.runners.kafka.streams.translation; + +import org.apache.beam.runners.fnexecution.provisioning.JobInfo; +import org.apache.beam.runners.kafka.streams.KafkaStreamsPipelineOptions; + +/** Mutable state shared while translating a portable pipeline into a Kafka Streams topology. */ +public class KafkaStreamsTranslationContext { + + private final JobInfo jobInfo; + private final KafkaStreamsPipelineOptions pipelineOptions; + + public static KafkaStreamsTranslationContext create( + JobInfo jobInfo, KafkaStreamsPipelineOptions pipelineOptions) { + return new KafkaStreamsTranslationContext(jobInfo, pipelineOptions); + } + + private KafkaStreamsTranslationContext( + JobInfo jobInfo, KafkaStreamsPipelineOptions pipelineOptions) { + this.jobInfo = jobInfo; + this.pipelineOptions = pipelineOptions; + } + + public JobInfo getJobInfo() { + return jobInfo; + } + + public KafkaStreamsPipelineOptions getPipelineOptions() { + return pipelineOptions; + } +} diff --git a/runners/kafka-streams/src/main/java/org/apache/beam/runners/kafka/streams/translation/package-info.java b/runners/kafka-streams/src/main/java/org/apache/beam/runners/kafka/streams/translation/package-info.java new file mode 100644 index 000000000000..9c09b9ceeb0b --- /dev/null +++ b/runners/kafka-streams/src/main/java/org/apache/beam/runners/kafka/streams/translation/package-info.java @@ -0,0 +1,20 @@ +/* + * 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 + * + * http://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. + */ + +/** Portable pipeline translation to Kafka Streams topologies. */ +package org.apache.beam.runners.kafka.streams.translation; diff --git a/settings.gradle.kts b/settings.gradle.kts index fc5f40c23d17..b92b254981fe 100644 --- a/settings.gradle.kts +++ b/settings.gradle.kts @@ -143,6 +143,7 @@ include(":runners:google-cloud-dataflow-java:examples-streaming") include(":runners:java-fn-execution") include(":runners:java-job-service") include(":runners:jet") +include(":runners:kafka-streams") include(":runners:local-java") include(":runners:portability:java") include(":runners:prism")