diff --git a/.github/trigger_files/beam_PreCommit_Java_Kafka_Streams_Runner.json b/.github/trigger_files/beam_PreCommit_Java_Kafka_Streams_Runner.json
new file mode 100644
index 000000000000..5abe02fc09c7
--- /dev/null
+++ b/.github/trigger_files/beam_PreCommit_Java_Kafka_Streams_Runner.json
@@ -0,0 +1,4 @@
+{
+ "comment": "Modify this file in a trivial way to cause this test suite to run.",
+ "modification": 1
+}
diff --git a/.github/workflows/README.md b/.github/workflows/README.md
index 41287872ee25..4fd8c8d36a61 100644
--- a/.github/workflows/README.md
+++ b/.github/workflows/README.md
@@ -259,6 +259,7 @@ PreCommit Jobs run in a schedule and also get triggered in a PR if relevant sour
| [ PreCommit Java HBase IO Direct ](https://github.com/apache/beam/actions/workflows/beam_PreCommit_Java_HBase_IO_Direct.yml) | N/A |`Run Java_HBase_IO_Direct PreCommit`| [](https://github.com/apache/beam/actions/workflows/beam_PreCommit_Java_HBase_IO_Direct.yml?query=event%3Aschedule) |
| [ PreCommit Java HCatalog IO Direct ](https://github.com/apache/beam/actions/workflows/beam_PreCommit_Java_HCatalog_IO_Direct.yml) | N/A |`Run Java_HCatalog_IO_Direct PreCommit`| [](https://github.com/apache/beam/actions/workflows/beam_PreCommit_Java_HCatalog_IO_Direct.yml?query=event%3Aschedule) |
| [ PreCommit Java Kafka IO Direct ](https://github.com/apache/beam/actions/workflows/beam_PreCommit_Java_Kafka_IO_Direct.yml) | N/A |`Run Java_Kafka_IO_Direct PreCommit`| [](https://github.com/apache/beam/actions/workflows/beam_PreCommit_Java_Kafka_IO_Direct.yml?query=event%3Aschedule) |
+| [ PreCommit Java Kafka Streams Runner ](https://github.com/apache/beam/actions/workflows/beam_PreCommit_Java_Kafka_Streams_Runner.yml) | N/A |`Run Java_Kafka_Streams_Runner PreCommit`| [](https://github.com/apache/beam/actions/workflows/beam_PreCommit_Java_Kafka_Streams_Runner.yml?query=event%3Aschedule) |
| [ PreCommit Java InfluxDb IO Direct ](https://github.com/apache/beam/actions/workflows/beam_PreCommit_Java_InfluxDb_IO_Direct.yml) | N/A |`Run Java_InfluxDb_IO_Direct PreCommit`| [](https://github.com/apache/beam/actions/workflows/beam_PreCommit_Java_InfluxDb_IO_Direct.yml?query=event%3Aschedule) |
| [ PreCommit Java IOs Direct ](https://github.com/apache/beam/actions/workflows/beam_PreCommit_Java_IOs_Direct.yml) | N/A |`Run Java_IOs_Direct PreCommit`| N/A |
| [ PreCommit Java JDBC IO Direct ](https://github.com/apache/beam/actions/workflows/beam_PreCommit_Java_JDBC_IO_Direct.yml) | N/A |`Run Java_JDBC_IO_Direct PreCommit`| [](https://github.com/apache/beam/actions/workflows/beam_PreCommit_Java_JDBC_IO_Direct.yml?query=event%3Aschedule) |
diff --git a/.github/workflows/beam_PreCommit_Java_Kafka_Streams_Runner.yml b/.github/workflows/beam_PreCommit_Java_Kafka_Streams_Runner.yml
new file mode 100644
index 000000000000..005764b4d33c
--- /dev/null
+++ b/.github/workflows/beam_PreCommit_Java_Kafka_Streams_Runner.yml
@@ -0,0 +1,119 @@
+# 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.
+
+name: PreCommit Java Kafka Streams Runner
+
+on:
+ push:
+ tags: ['v*']
+ branches: ['master', 'release-*']
+ paths:
+ - "runners/kafka-streams/**"
+ - ".github/workflows/beam_PreCommit_Java_Kafka_Streams_Runner.yml"
+ pull_request_target:
+ branches: ['master', 'release-*']
+ paths:
+ - "runners/kafka-streams/**"
+ - 'release/trigger_all_tests.json'
+ - '.github/trigger_files/beam_PreCommit_Java_Kafka_Streams_Runner.json'
+ issue_comment:
+ types: [created]
+ schedule:
+ - cron: '15 2/6 * * *'
+ workflow_dispatch:
+
+# Setting explicit permissions for the action to avoid the default permissions which are `write-all` in case of pull_request_target event
+permissions:
+ actions: write
+ pull-requests: write
+ checks: write
+ contents: read
+ deployments: read
+ id-token: none
+ issues: write
+ discussions: read
+ packages: read
+ pages: read
+ repository-projects: read
+ security-events: read
+ statuses: read
+
+# This allows a subsequently queued workflow run to interrupt previous runs
+concurrency:
+ group: '${{ github.workflow }} @ ${{ github.event.pull_request.number || github.event.pull_request.head.label || github.sha || github.head_ref || github.ref }}-${{ github.event.schedule || github.event.comment.id || github.event.sender.login }}'
+ cancel-in-progress: true
+
+env:
+ DEVELOCITY_ACCESS_KEY: ${{ secrets.DEVELOCITY_ACCESS_KEY }}
+ GRADLE_ENTERPRISE_CACHE_USERNAME: ${{ secrets.GE_CACHE_USERNAME }}
+ GRADLE_ENTERPRISE_CACHE_PASSWORD: ${{ secrets.GE_CACHE_PASSWORD }}
+
+jobs:
+ beam_PreCommit_Java_Kafka_Streams_Runner:
+ name: ${{ matrix.job_name }} (${{ matrix.job_phrase }})
+ strategy:
+ matrix:
+ job_name: ["beam_PreCommit_Java_Kafka_Streams_Runner"]
+ job_phrase: ["Run Java_Kafka_Streams_Runner PreCommit"]
+ timeout-minutes: 60
+ if: |
+ github.event_name == 'push' ||
+ github.event_name == 'pull_request_target' ||
+ (github.event_name == 'schedule' && github.repository == 'apache/beam') ||
+ github.event_name == 'workflow_dispatch' ||
+ github.event.comment.body == 'Run Java_Kafka_Streams_Runner PreCommit'
+ runs-on: [self-hosted, ubuntu-24.04, main]
+ steps:
+ - uses: actions/checkout@v6
+ - name: Setup repository
+ uses: ./.github/actions/setup-action
+ with:
+ comment_phrase: ${{ matrix.job_phrase }}
+ github_token: ${{ secrets.GITHUB_TOKEN }}
+ github_job: ${{ matrix.job_name }} (${{ matrix.job_phrase }})
+ - name: Setup environment
+ uses: ./.github/actions/setup-environment-action
+ - name: run Kafka Streams runner build script
+ uses: ./.github/actions/gradle-command-self-hosted-action
+ with:
+ gradle-command: :runners:kafka-streams:build
+ arguments: -Pwith-kafka-streams-runner
+ max-workers: 4
+ - name: Archive JUnit Test Results
+ uses: actions/upload-artifact@v7
+ if: ${{ !success() }}
+ with:
+ name: JUnit Test Results
+ path: "**/build/reports/tests/"
+ - name: Publish JUnit Test Results
+ uses: EnricoMi/publish-unit-test-result-action@v2
+ if: always()
+ with:
+ commit: '${{ env.prsha || env.GITHUB_SHA }}'
+ comment_mode: ${{ github.event_name == 'issue_comment' && 'always' || 'off' }}
+ files: '**/build/test-results/**/*.xml'
+ large_files: true
+ - name: Archive SpotBugs Results
+ uses: actions/upload-artifact@v7
+ if: always()
+ with:
+ name: SpotBugs Results
+ path: '**/build/reports/spotbugs/*.html'
+ - name: Publish SpotBugs Results
+ uses: jwgmeligmeyling/spotbugs-github-action@v1.2
+ if: always()
+ with:
+ name: Publish SpotBugs
+ path: '**/build/reports/spotbugs/*.html'
diff --git a/CHANGES.md b/CHANGES.md
index 73966a48313c..263a2dc5d0b4 100644
--- a/CHANGES.md
+++ b/CHANGES.md
@@ -211,6 +211,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 de278f9f2833..d0cc5fbb32b7 100644
--- a/build.gradle.kts
+++ b/build.gradle.kts
@@ -278,6 +278,10 @@ tasks.register("javaPreCommit") {
dependsOn(":runners:java-fn-execution:build")
dependsOn(":runners:java-job-service:build")
dependsOn(":runners:jet:build")
+ // Only when the opt-in flag put it in the build; see settings.gradle.kts.
+ if (findProject(":runners:kafka-streams") != null) {
+ 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..399550534092
--- /dev/null
+++ b/runners/kafka-streams/build.gradle
@@ -0,0 +1,201 @@
+/*
+ * 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.
+ */
+
+import groovy.json.JsonOutput
+import java.time.Duration
+
+plugins { id 'org.apache.beam.module' }
+
+// An extension property rather than a local, so the job server module can pin the same version
+// instead of repeating it. Both have to pin: applyJavaNature forces every version in library.java,
+// which includes an older kafka-clients, and a module that does not override it links against that
+// one at runtime.
+ext.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 {
+ validatesRunner
+}
+
+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: ":runners:kafka-streams:proto", configuration: "shadow")
+ implementation project(path: ":model:pipeline", configuration: "shadow")
+ implementation project(path: ":model:fn-execution", configuration: "shadow")
+ implementation project(path: ":model:job-management", 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"
+
+ testImplementation project(path: ":sdks:java:core", configuration: "shadowTest")
+ testImplementation project(":sdks:java:harness")
+ testImplementation library.java.hamcrest
+ testImplementation library.java.junit
+ testImplementation library.java.mockito_core
+ testImplementation "org.apache.kafka:kafka-streams-test-utils:$kafka_version"
+ testImplementation library.java.testcontainers_kafka
+
+ // Beam's @ValidatesRunner suite: the test classes come from the SDK core test jar; the runner
+ // (TestKafkaStreamsRunner) and its TopologyTestDriver harness come from this module's test
+ // output and test runtime classpath.
+ validatesRunner project(path: ":sdks:java:core", configuration: "shadowTest")
+ validatesRunner project(project.path)
+ validatesRunner sourceSets.test.output
+ validatesRunner sourceSets.test.runtimeClasspath
+}
+
+
+// Starts the job server a portable pipeline is submitted to. Pass driver arguments with
+// -PjobServerArgs="--job-port=8099,--artifact-port=8098".
+tasks.register("runJobServer", JavaExec) {
+ group = "Application"
+ description = "Runs the Kafka Streams job server."
+ mainClass = "org.apache.beam.runners.kafka.streams.KafkaStreamsJobServerDriver"
+ classpath = sourceSets.main.runtimeClasspath
+ args = project.hasProperty("jobServerArgs") ? project.property("jobServerArgs").split(",") : []
+}
+
+// The broker integration test drives the production runner against a real Kafka in Docker, so it
+// is not part of the default build. Run it with :runners:kafka-streams:brokerIntegrationTest.
+test {
+ filter {
+ excludeTestsMatching 'org.apache.beam.runners.kafka.streams.*IT'
+ }
+}
+
+tasks.register("brokerIntegrationTest", Test) {
+ group = "Verification"
+ description = "Runs the Kafka Streams runner against a real broker (requires Docker)."
+ outputs.upToDateWhen { false }
+ testClassesDirs = sourceSets.test.output.classesDirs
+ classpath = sourceSets.test.runtimeClasspath
+ filter {
+ includeTestsMatching 'org.apache.beam.runners.kafka.streams.*IT'
+ }
+ // A container start plus a streaming run is well past the default per-test expectations.
+ timeout = Duration.ofMinutes(15)
+}
+
+// Known-failing @ValidatesRunner tests, excluded until the feature they need lands.
+def sickbayTests = [
+ // Merging (session) windows are not supported yet: ReduceFnRunner drives them through a merging
+ // window set that moves per-window state as windows merge, which this first windowing pass does
+ // not implement. Non-merging windows (fixed, sliding), the default trigger and timestamp
+ // combiners do work, for both GroupByKey and Combine. Lands with the follow-up windowing PR.
+ 'org.apache.beam.sdk.transforms.GroupByKeyTest$WindowTests.testGroupByKeyMergingWindows',
+ 'org.apache.beam.sdk.transforms.CombineTest$WindowingTests.testSessionsCombine',
+ // A DoFn whose @StartBundle throws never gets to report its error: SdkHarnessClient.newBundle
+ // sends the ProcessBundleRequest and then blocks in GrpcDataService.createOutboundAggregator
+ // waiting for the SDK harness to open its data stream, which a bundle that failed during setup
+ // never does — so the run blocks for the data service's three-minute timeout instead of
+ // surfacing the user's exception. This is shared java-fn-execution behaviour rather than
+ // anything specific to this runner; the Flink runner sickbays all of LifecycleTests and the
+ // Prism runner sickbays each of its three error tests. The @ProcessElement and @FinishBundle
+ // variants do pass here, because by then the data stream is established.
+ // Tracked by https://github.com/apache/beam/issues/39452.
+ 'org.apache.beam.sdk.transforms.ParDoTest$LifecycleTests.testParDoWithErrorInStartBatch',
+]
+
+tasks.register("validatesRunner", Test) {
+ group = "Verification"
+ description = "Runs the subset of Beam's ValidatesRunner suite the Kafka Streams runner supports."
+ // Never consider up-to-date; the suite is the correctness gate.
+ outputs.upToDateWhen { false }
+ systemProperty "beamTestPipelineOptions",
+ JsonOutput.toJson(["--runner=org.apache.beam.runners.kafka.streams.TestKafkaStreamsRunner"])
+ classpath = configurations.validatesRunner
+ testClassesDirs = files(project(":sdks:java:core").sourceSets.test.output.classesDirs)
+ maxParallelForks 2
+ useJUnit {
+ includeCategories 'org.apache.beam.sdk.testing.ValidatesRunner'
+ // Environment / harness features that need a properly configured external environment.
+ excludeCategories 'org.apache.beam.sdk.testing.UsesExternalService'
+ excludeCategories 'org.apache.beam.sdk.testing.UsesSdkHarnessEnvironment'
+ excludeCategories 'org.apache.beam.sdk.testing.UsesBundleFinalizer'
+ excludeCategories 'org.apache.beam.sdk.testing.UsesJavaExpansionService'
+ excludeCategories 'org.apache.beam.sdk.testing.UsesPythonExpansionService'
+ // Features the runner does not support yet.
+ excludeCategories 'org.apache.beam.sdk.testing.UsesSideInputs'
+ excludeCategories 'org.apache.beam.sdk.testing.UsesStatefulParDo'
+ excludeCategories 'org.apache.beam.sdk.testing.UsesTimersInParDo'
+ excludeCategories 'org.apache.beam.sdk.testing.UsesTimerMap'
+ excludeCategories 'org.apache.beam.sdk.testing.UsesLoopingTimer'
+ excludeCategories 'org.apache.beam.sdk.testing.UsesStrictTimerOrdering'
+ excludeCategories 'org.apache.beam.sdk.testing.UsesProcessingTimeTimers'
+ excludeCategories 'org.apache.beam.sdk.testing.UsesOnWindowExpiration'
+ excludeCategories 'org.apache.beam.sdk.testing.UsesTestStream'
+ excludeCategories 'org.apache.beam.sdk.testing.UsesUnboundedPCollections'
+ excludeCategories 'org.apache.beam.sdk.testing.UsesUnboundedSplittableParDo'
+ excludeCategories 'org.apache.beam.sdk.testing.UsesBoundedSplittableParDo'
+ excludeCategories 'org.apache.beam.sdk.testing.UsesCustomWindowMerging'
+ excludeCategories 'org.apache.beam.sdk.testing.UsesMetricsPusher'
+ excludeCategories 'org.apache.beam.sdk.testing.UsesCommittedMetrics'
+ excludeCategories 'org.apache.beam.sdk.testing.UsesSystemMetrics'
+ excludeCategories 'org.apache.beam.sdk.testing.UsesOrderedListState'
+ excludeCategories 'org.apache.beam.sdk.testing.UsesMultimapState'
+ excludeCategories 'org.apache.beam.sdk.testing.UsesMapState'
+ excludeCategories 'org.apache.beam.sdk.testing.UsesSetState'
+ excludeCategories 'org.apache.beam.sdk.testing.FlattenWithHeterogeneousCoders'
+ excludeCategories 'org.apache.beam.sdk.testing.LargeKeys$Above100MB'
+ }
+ filter {
+ // The suites enabled so far, extended class by class as runner support grows. An explicit
+ // include list is needed because feature gaps like windowing beyond the global window have no
+ // JUnit category to exclude.
+ includeTestsMatching 'org.apache.beam.sdk.transforms.CreateTest'
+ includeTestsMatching 'org.apache.beam.sdk.transforms.FlattenTest'
+ includeTestsMatching 'org.apache.beam.sdk.transforms.GroupByKeyTest*'
+ includeTestsMatching 'org.apache.beam.sdk.transforms.ParDoTest*'
+ includeTestsMatching 'org.apache.beam.sdk.transforms.CombineTest*'
+ for (String test : sickbayTests) {
+ excludeTestsMatching test
+ }
+ failOnNoMatchingTests = false
+ }
+}
diff --git a/runners/kafka-streams/job-server/build.gradle b/runners/kafka-streams/job-server/build.gradle
new file mode 100644
index 000000000000..aef2f271b7b1
--- /dev/null
+++ b/runners/kafka-streams/job-server/build.gradle
@@ -0,0 +1,88 @@
+/*
+ * 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 JobServer build file.
+ *
+ * Packages the runner and everything it needs into one jar, so a pipeline from an SDK other than
+ * Java can start a job server without a Beam source tree. The Python KafkaStreamsRunner builds and
+ * launches this jar for the user.
+ */
+
+apply plugin: 'org.apache.beam.module'
+apply plugin: 'application'
+// Must be set before the shadow plugin is applied.
+mainClassName = "org.apache.beam.runners.kafka.streams.KafkaStreamsJobServerDriver"
+
+applyJavaNature(
+ automaticModuleName: 'org.apache.beam.runners.kafka.streams.jobserver',
+ validateShadowJar: false,
+ exportJavadoc: false,
+ shadowClosure: {
+ // Kafka's clients and Streams libraries ship reference.conf-style resources that have to be
+ // concatenated rather than overwritten when everything lands in one jar.
+ append "reference.conf"
+ },
+)
+
+def kafkaStreamsRunnerProject = ":runners:kafka-streams"
+
+description = "Apache Beam :: Runners :: Kafka Streams :: Job Server"
+
+evaluationDependsOn(kafkaStreamsRunnerProject)
+
+// The runner is compiled against this version, so the jar has to carry it. Without this the
+// versions forced by applyJavaNature win, the shaded jar ships an older kafka-clients, and every
+// pipeline fails at translation with a NoSuchMethodError rather than at build time.
+def kafka_version = project(kafkaStreamsRunnerProject).kafka_version
+
+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 {
+ implementation project(kafkaStreamsRunnerProject)
+ permitUnusedDeclared project(kafkaStreamsRunnerProject)
+ // A binding, or the job server starts but logs nothing at all, which is unhelpful for something
+ // a user runs in the foreground and reads to see what their pipeline is doing.
+ runtimeOnly library.java.slf4j_simple
+ runtimeOnly project(":sdks:java:extensions:google-cloud-platform-core")
+}
+
+// The runner's classes only exist in the shadow jar, so the job server has to be started through
+// runShadow rather than the plain run task.
+runShadow {
+ args = []
+ if (project.hasProperty('jobHost'))
+ args += ["--job-host=${project.property('jobHost')}"]
+ if (project.hasProperty('jobPort'))
+ args += ["--job-port=${project.property('jobPort')}"]
+ if (project.hasProperty('artifactPort'))
+ args += ["--artifact-port=${project.property('artifactPort')}"]
+ if (project.hasProperty('expansionPort'))
+ args += ["--expansion-port=${project.property('expansionPort')}"]
+ if (project.hasProperty('artifactsDir'))
+ args += ["--artifacts-dir=${project.property('artifactsDir')}"]
+ if (project.hasProperty('cleanArtifactsPerJob'))
+ args += ["--clean-artifacts-per-job=${project.property('cleanArtifactsPerJob')}"]
+}
diff --git a/runners/kafka-streams/measurement/build.gradle b/runners/kafka-streams/measurement/build.gradle
new file mode 100644
index 000000000000..5e7992a0916e
--- /dev/null
+++ b/runners/kafka-streams/measurement/build.gradle
@@ -0,0 +1,78 @@
+/*
+ * 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.
+ */
+
+/**
+ * An application for measuring the Kafka Streams runner's behaviour when instances come and go.
+ *
+ * Not part of the build's verification: it is something a person runs against a Kafka, several
+ * copies at once, and watches.
+ */
+
+apply plugin: 'org.apache.beam.module'
+apply plugin: 'application'
+mainClassName = "org.apache.beam.runners.kafka.streams.measurement.RescalingMeasurement"
+
+applyJavaNature(
+ automaticModuleName: 'org.apache.beam.runners.kafka.streams.measurement',
+ publish: false,
+ exportJavadoc: false,
+ // This module runs the pipeline in its own process, so the SDK harness and its dependencies are
+ // on the classpath, and SpotBugs reports on those rather than on the four classes here — some
+ // eleven thousand warnings, none of them in this source tree. The same is done in the it/
+ // modules, which are on the classpath of what they exercise for the same reason. Checkstyle,
+ // ErrorProne, spotless and the nullness checker all still run.
+ enableSpotbugs: false,
+)
+
+description = "Apache Beam :: Runners :: Kafka Streams :: Measurement"
+
+def kafkaStreamsRunnerProject = ":runners:kafka-streams"
+
+evaluationDependsOn(kafkaStreamsRunnerProject)
+
+// Same pin as the runner and the job server: applyJavaNature forces the versions in library.java,
+// which includes an older kafka-clients than the runner is compiled against.
+def kafka_version = project(kafkaStreamsRunnerProject).kafka_version
+
+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 {
+ implementation project(kafkaStreamsRunnerProject)
+ implementation project(path: ":sdks:java:core", configuration: "shadow")
+ implementation project(path: ":model:pipeline", configuration: "shadow")
+ implementation project(":runners:java-fn-execution")
+ // On the compile classpath to resolve PortablePipelineRunner, which KafkaStreamsPipelineRunner
+ // implements; no class of it is named here, so the dependency analysis does not see it used.
+ implementation project(":runners:java-job-service")
+ permitUnusedDeclared project(":runners:java-job-service")
+ implementation project(":runners:core-java")
+ permitUnusedDeclared project(":runners:core-java")
+ // The pipeline's own code runs in this process, so the Java SDK harness has to be present.
+ runtimeOnly project(":sdks:java:harness")
+ implementation library.java.joda_time
+ // Without a binding the application starts and says nothing, which is unhelpful for something
+ // whose whole purpose is to be watched while it runs.
+ runtimeOnly library.java.slf4j_simple
+}
diff --git a/runners/kafka-streams/measurement/docker-compose.yml b/runners/kafka-streams/measurement/docker-compose.yml
new file mode 100644
index 000000000000..aa9102b97424
--- /dev/null
+++ b/runners/kafka-streams/measurement/docker-compose.yml
@@ -0,0 +1,44 @@
+#
+# 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.
+#
+
+# One Kafka for the measurement application. One broker is enough: what gets run several times is
+# the runner instance, not the broker.
+#
+# docker compose -f runners/kafka-streams/measurement/docker-compose.yml up -d
+#
+# group.min.session.timeout.ms is lowered because a broker refuses a session timeout below it, and
+# how quickly the group notices a departed instance is the floor on how quickly its work moves. The
+# default of 6s would put a floor under every measurement of recovery.
+services:
+ kafka:
+ image: apache/kafka:4.0.0
+ container_name: ks-measurement-kafka
+ ports:
+ - "9092:9092"
+ environment:
+ KAFKA_NODE_ID: 1
+ KAFKA_PROCESS_ROLES: broker,controller
+ KAFKA_LISTENERS: PLAINTEXT://:9092,CONTROLLER://:9093
+ KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://localhost:9092
+ KAFKA_CONTROLLER_QUORUM_VOTERS: 1@localhost:9093
+ KAFKA_CONTROLLER_LISTENER_NAMES: CONTROLLER
+ KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT
+ KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
+ KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 1
+ KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 1
+ KAFKA_GROUP_MIN_SESSION_TIMEOUT_MS: 1000
+ KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS: 0
diff --git a/runners/kafka-streams/measurement/src/main/java/org/apache/beam/runners/kafka/streams/measurement/RescalingMeasurement.java b/runners/kafka-streams/measurement/src/main/java/org/apache/beam/runners/kafka/streams/measurement/RescalingMeasurement.java
new file mode 100644
index 000000000000..34ef1654345d
--- /dev/null
+++ b/runners/kafka-streams/measurement/src/main/java/org/apache/beam/runners/kafka/streams/measurement/RescalingMeasurement.java
@@ -0,0 +1,246 @@
+/*
+ * 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.measurement;
+
+import org.apache.beam.model.pipeline.v1.RunnerApi;
+import org.apache.beam.runners.fnexecution.provisioning.JobInfo;
+import org.apache.beam.runners.kafka.streams.KafkaStreamsPipelineOptions;
+import org.apache.beam.runners.kafka.streams.KafkaStreamsPipelineRunner;
+import org.apache.beam.runners.kafka.streams.KafkaStreamsRunner;
+import org.apache.beam.sdk.Pipeline;
+import org.apache.beam.sdk.io.GenerateSequence;
+import org.apache.beam.sdk.options.Default;
+import org.apache.beam.sdk.options.Description;
+import org.apache.beam.sdk.options.PipelineOptionsFactory;
+import org.apache.beam.sdk.options.PortablePipelineOptions;
+import org.apache.beam.sdk.transforms.Count;
+import org.apache.beam.sdk.transforms.DoFn;
+import org.apache.beam.sdk.transforms.MapElements;
+import org.apache.beam.sdk.transforms.ParDo;
+import org.apache.beam.sdk.transforms.windowing.BoundedWindow;
+import org.apache.beam.sdk.transforms.windowing.FixedWindows;
+import org.apache.beam.sdk.transforms.windowing.Window;
+import org.apache.beam.sdk.util.construction.Environments;
+import org.apache.beam.sdk.util.construction.PipelineOptionsTranslation;
+import org.apache.beam.sdk.util.construction.PipelineTranslation;
+import org.apache.beam.sdk.util.construction.SplittableParDo;
+import org.apache.beam.sdk.values.KV;
+import org.apache.beam.sdk.values.TypeDescriptors;
+import org.joda.time.Duration;
+
+/**
+ * One instance of a streaming pipeline, run as an ordinary application, for measuring what happens
+ * when instances come and go.
+ *
+ *
Run several against one Kafka. They share an application id, so the consumer group divides the
+ * work between them and stopping one hands its share to the others. It is an application rather
+ * than a test because the numbers only mean something under a realistic load: a grouping over
+ * thousands of keys, fed fast enough that no partition sits idle holding a watermark back.
+ *
+ *
The source runs at a fixed rate over a fixed key space, so a complete window is known before
+ * the run starts — one line per key, the same count on each — which is what makes a shortfall
+ * legible as one.
+ *
+ *
Each instance needs its own {@code --stateDir}; sharing one fails with a {@code
+ * LockException}. Output is one line per key per window, counted by the pipeline itself rather than
+ * beside it, so the tally does not depend on how many instances are running:
+ *
+ *
{@code skew_ms} is the gap between the window's event time and the wall clock when it came
+ * out. A pipeline that cannot keep up should report its groups later and later while still
+ * reporting all of them, so climbing skew with complete windows is congestion and missing groups
+ * are something else. To watch a handover, kill one instance and watch the other; the delay before
+ * it reports the dead instance's share is dominated by {@code --sessionTimeoutMs}.
+ */
+
+public final class RescalingMeasurement {
+
+ private RescalingMeasurement() {}
+
+ /** Whether the command line mentioned an option, so that a default is not applied over it. */
+ private static boolean given(String[] args, String name) {
+ for (String arg : args) {
+ if (arg.equals("--" + name) || arg.startsWith("--" + name + "=")) {
+ return true;
+ }
+ }
+ return false;
+ }
+
+ /**
+ * Applies the defaults this measurement needs, where they differ from the runner's own.
+ *
+ *
The runner's defaults are meant for a pipeline, not for this. Left alone, the source is read
+ * in large enough turns that the read starves the rest of the topology and no groups come out at
+ * all. The parallelism is raised for a related reason: a measurement of work moving between
+ * instances needs more than the single partition the runner defaults to, since with one partition
+ * there is nothing to divide.
+ */
+ private static void applyMeasurementDefaults(String[] args, MeasurementOptions options) {
+ if (!given(args, "readMaxElementsPerPoll")) {
+ options.setReadMaxElementsPerPoll(200);
+ }
+ if (!given(args, "internalParallelism")) {
+ options.setInternalParallelism(3);
+ }
+ }
+
+ /** Options of the measurement itself, on top of the runner's own. */
+ public interface MeasurementOptions extends KafkaStreamsPipelineOptions {
+
+ @Description("Name for this instance in the output, so several can be told apart.")
+ @Default.String("instance")
+ String getInstanceName();
+
+ void setInstanceName(String instanceName);
+
+ @Description(
+ "How many distinct keys the grouping runs over. Thousands, so that every partition of the"
+ + " shuffle has work and no partition sits idle holding a watermark back. With a"
+ + " window long enough to contain them all, this is also how many groups a complete"
+ + " window has.")
+ @Default.Integer(2_000)
+ int getNumKeys();
+
+ void setNumKeys(int numKeys);
+
+ @Description(
+ "How many elements the source produces per second. Fixed rather than as-fast-as-possible so"
+ + " that a window's contents are known in advance and a shortfall is visible.")
+ @Default.Integer(20_000)
+ int getElementsPerSecond();
+
+ void setElementsPerSecond(int elementsPerSecond);
+
+ @Description("Window size in milliseconds; how often the groups are counted and reported.")
+ @Default.Integer(1_000)
+ int getWindowMs();
+
+ void setWindowMs(int windowMs);
+ }
+
+ /**
+ * Logs each group the pipeline produces, with how far behind the wall clock its window was.
+ *
+ *
One line per key per window. With a fixed rate over a fixed key space every window holds the
+ * same groups, so counting the lines for a window says whether the window was complete, and no
+ * counter has to be kept anywhere for that to be true — the count is the pipeline's own output
+ * rather than a tally maintained beside it, which is what makes it independent of how many
+ * instances are running.
+ *
+ *
The skew is the point of the timestamp. A pipeline that cannot keep up should report its
+ * groups later and later rather than stop reporting them, so a skew that climbs while the groups
+ * stay complete is the pipeline falling behind, and groups going missing is something else.
+ */
+ private static class ReportGroupFn extends DoFn, Void> {
+ private final String instanceName;
+
+ ReportGroupFn(String instanceName) {
+ this.instanceName = instanceName;
+ }
+
+ @ProcessElement
+ public void processElement(@Element KV group, BoundedWindow window) {
+ long windowEnd = window.maxTimestamp().getMillis();
+ long now = System.currentTimeMillis();
+ System.out.printf(
+ "%d %s window_end=%d key=%s count=%d skew_ms=%d%n",
+ now, instanceName, windowEnd, group.getKey(), group.getValue(), now - windowEnd);
+ }
+ }
+
+ public static void main(String[] args) throws Exception {
+ PipelineOptionsFactory.register(MeasurementOptions.class);
+ // Deliberately not withValidation(): that enforces the options a pipeline needs when it is
+ // submitted to a job server, and --jobEndpoint above all, which means nothing here because this
+ // application runs the pipeline itself.
+ MeasurementOptions options = PipelineOptionsFactory.fromArgs(args).as(MeasurementOptions.class);
+ if (options.getApplicationId() == null || options.getApplicationId().isEmpty()) {
+ throw new IllegalArgumentException(
+ "--applicationId is required, and every instance of one measurement must share it: it is"
+ + " what puts them in the same consumer group and so divides the work between them.");
+ }
+ applyMeasurementDefaults(args, options);
+ // Pipeline.create needs a runner class even though this application never calls pipeline.run()
+ // — it builds the pipeline proto and hands it to the runner below itself.
+ options.setRunner(KafkaStreamsRunner.class);
+ // The user code runs in this same process, so no container or separate worker is needed.
+ options
+ .as(PortablePipelineOptions.class)
+ .setDefaultEnvironmentType(Environments.ENVIRONMENT_EMBEDDED);
+
+ // A window holds every key as long as it is long enough for the rate to reach them all; below
+ // that the source has not got round to each key once and the window is short by construction.
+ long elementsPerWindow = (long) options.getElementsPerSecond() * options.getWindowMs() / 1_000L;
+ long expectedGroups = Math.min(options.getNumKeys(), elementsPerWindow);
+
+ int numKeys = options.getNumKeys();
+ Pipeline pipeline = Pipeline.create(options);
+ pipeline
+ .apply(
+ "read",
+ GenerateSequence.from(0)
+ .withRate(options.getElementsPerSecond(), Duration.standardSeconds(1)))
+ .apply(
+ "key",
+ // numKeys is read here rather than inside the lambda: reaching for it through options
+ // would capture the PipelineOptions in the transform, which cannot be serialized.
+ MapElements.into(TypeDescriptors.strings()).via((Long n) -> "key-" + (n % numKeys)))
+ .apply("window", Window.into(FixedWindows.of(Duration.millis(options.getWindowMs()))))
+ .apply("countPerKey", Count.perElement())
+ .apply("report", ParDo.of(new ReportGroupFn(options.getInstanceName())));
+
+ SplittableParDo.convertReadBasedSplittableDoFnsToPrimitiveReads(pipeline);
+ RunnerApi.Pipeline proto = PipelineTranslation.toProto(pipeline);
+ JobInfo jobInfo =
+ JobInfo.create(
+ options.getApplicationId(),
+ options.getJobName(),
+ "",
+ PipelineOptionsTranslation.toProto(options));
+
+ System.out.printf(
+ "starting %s: application=%s keys=%d rate=%d/s parallelism=%d window=%dms"
+ + " session_timeout=%dms read_per_poll=%d bundle=%d expected_groups_per_window=%d%n",
+ options.getInstanceName(),
+ options.getApplicationId(),
+ options.getNumKeys(),
+ options.getElementsPerSecond(),
+ options.getInternalParallelism(),
+ options.getWindowMs(),
+ options.getSessionTimeoutMs(),
+ options.getReadMaxElementsPerPoll(),
+ options.getMaxBundleSize(),
+ expectedGroups);
+
+ // Blocks until the instance is stopped; a streaming pipeline has no end of its own.
+ new KafkaStreamsPipelineRunner(options).run(proto, jobInfo);
+ }
+}
diff --git a/runners/kafka-streams/measurement/src/main/java/org/apache/beam/runners/kafka/streams/measurement/package-info.java b/runners/kafka-streams/measurement/src/main/java/org/apache/beam/runners/kafka/streams/measurement/package-info.java
new file mode 100644
index 000000000000..41579676fb8d
--- /dev/null
+++ b/runners/kafka-streams/measurement/src/main/java/org/apache/beam/runners/kafka/streams/measurement/package-info.java
@@ -0,0 +1,26 @@
+/*
+ * 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.
+ */
+
+/**
+ * An application for measuring the Kafka Streams runner's behaviour when instances come and go.
+ *
+ *
Not part of the build's verification: it is something a person runs against a Kafka, several
+ * copies at once, and watches. See {@link
+ * org.apache.beam.runners.kafka.streams.measurement.RescalingMeasurement} for how to run it.
+ */
+package org.apache.beam.runners.kafka.streams.measurement;
diff --git a/runners/kafka-streams/proto/build.gradle b/runners/kafka-streams/proto/build.gradle
new file mode 100644
index 000000000000..d3e42aeb617f
--- /dev/null
+++ b/runners/kafka-streams/proto/build.gradle
@@ -0,0 +1,34 @@
+/*
+ * 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' }
+
+// Portability nature compiles the .proto against the vendored gRPC/protobuf (relocated to
+// org.apache.beam.vendor.grpc...), so consumers use the vendored protobuf runtime — no raw
+// com.google.protobuf on their classpath. Same pattern as the dataflow windmill proto module.
+applyPortabilityNature(
+ publish: false,
+ shadowJarValidationExcludes: ["org/apache/beam/runners/kafka/streams/v1/**"],
+ archivesBaseName: 'beam-runners-kafka-streams-proto',
+ generatedClassPatterns: [
+ /^org\.apache\.beam\.runners\.kafka\.streams\.v1.*/
+ ]
+)
+
+description = "Apache Beam :: Runners :: Kafka Streams :: Proto"
+ext.summary = "Kafka Streams runner control-message protos"
diff --git a/runners/kafka-streams/proto/src/main/proto/kafka_streams_payload.proto b/runners/kafka-streams/proto/src/main/proto/kafka_streams_payload.proto
new file mode 100644
index 000000000000..4dadf2d14b70
--- /dev/null
+++ b/runners/kafka-streams/proto/src/main/proto/kafka_streams_payload.proto
@@ -0,0 +1,58 @@
+/*
+ * 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.
+ */
+
+syntax = "proto3";
+
+package org.apache.beam.runners.kafka.streams.v1;
+
+option java_package = "org.apache.beam.runners.kafka.streams.v1";
+option java_outer_classname = "KafkaStreamsPayloadProtos";
+
+// On-wire form of the in-JVM KStreamsPayload envelope, used when the payload must cross a Kafka
+// topic boundary (e.g. the GroupByKey repartition topic and the watermark fan-out). Protobuf is
+// used for compatible schema evolution and compact varint encoding.
+message KafkaStreamsPayload {
+ // A watermark report: the watermark plus the in-band coordination fields a downstream
+ // watermark aggregator needs to reconstruct its input watermark.
+ message WatermarkPayload {
+ // Event-time watermark in milliseconds. Signed (sint64, zigzag-encoded) because Beam event
+ // times can be negative, e.g. BoundedWindow.TIMESTAMP_MIN_VALUE.
+ sint64 millis = 1;
+ // Which partition (physical instance) of the producing transform this report is for, in
+ // [0, total_partitions).
+ uint32 source_partition = 2;
+ // How many partitions (physical instances) the producing transform has in total.
+ uint32 total_partitions = 3;
+ // Globally unique id of the transform that produced this report. A producer stamps its own id
+ // without regard to who consumes the report; a consumer with several upstream transforms
+ // (e.g. Flatten) aggregates per producing transform, holding its output watermark until every
+ // partition of every upstream transform it expects has reported.
+ string transform_id = 4;
+ }
+
+ // A data element: the Beam WindowedValue encoded with the PCollection's windowed-value coder.
+ message DataPayload {
+ bytes value = 1;
+ }
+
+ // Exactly one variant is set; the oneof case discriminates data vs watermark.
+ oneof payload {
+ WatermarkPayload watermark = 1;
+ DataPayload data = 2;
+ }
+}
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..a32c7c487737
--- /dev/null
+++ b/runners/kafka-streams/src/main/java/org/apache/beam/runners/kafka/streams/KafkaStreamsJobInvoker.java
@@ -0,0 +1,96 @@
+/*
+ * 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.base.Strings;
+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. */
+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);
+
+ LOG.info("Invoking job {} with pipeline runner {}", invocationId, pipelineRunner);
+ return createJobInvocation(
+ invocationId,
+ retrievalToken,
+ executorService,
+ pipeline,
+ kafkaStreamsOptions,
+ pipelineRunner);
+ }
+
+ protected JobInvocation createJobInvocation(
+ String invocationId,
+ @Nullable String retrievalToken,
+ ListeningExecutorService executorService,
+ RunnerApi.Pipeline pipeline,
+ KafkaStreamsPipelineOptions kafkaStreamsOptions,
+ PortablePipelineRunner pipelineRunner) {
+ JobInfo jobInfo =
+ JobInfo.create(
+ invocationId,
+ kafkaStreamsOptions.getJobName(),
+ Strings.nullToEmpty(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..4f2134689253
--- /dev/null
+++ b/runners/kafka-streams/src/main/java/org/apache/beam/runners/kafka/streams/KafkaStreamsJobServerDriver.java
@@ -0,0 +1,106 @@
+/*
+ * 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. */
+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..99454b3a6588
--- /dev/null
+++ b/runners/kafka-streams/src/main/java/org/apache/beam/runners/kafka/streams/KafkaStreamsPipelineOptions.java
@@ -0,0 +1,151 @@
+/*
+ * 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;
+import org.apache.beam.sdk.options.Validation;
+
+/** 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). Must be specified explicitly: a shared default "
+ + "would let concurrent jobs collide on the same Kafka Streams consumer group.")
+ @Validation.Required
+ String getApplicationId();
+
+ void setApplicationId(String applicationId);
+
+ @Description("Soft cap on the number of elements per bundle.")
+ @Default.Integer(1000)
+ int getMaxBundleSize();
+
+ void setMaxBundleSize(int maxBundleSize);
+
+ @Description(
+ "How many elements an unbounded source may yield per poll. Separate from --maxBundleSize:"
+ + " a small bundle is how you get output promptly, while how much a source reads at a"
+ + " time is about throughput, and tying them together means a pipeline cannot have both.")
+ @Default.Integer(1000)
+ int getReadMaxElementsPerPoll();
+
+ void setReadMaxElementsPerPoll(int readMaxElementsPerPoll);
+
+ @Description(
+ "How long one turn of reading an unbounded source may take, in milliseconds, before the"
+ + " source yields the Kafka Streams thread. A source is polled from a punctuator"
+ + " scheduled every 50ms, and the same thread runs the rest of the topology, so a turn"
+ + " that overruns that interval is already due again when it returns and fires straight"
+ + " away: the source then holds the thread and the stages below it are never scheduled,"
+ + " which shows up as a pipeline that reads steadily and emits nothing at all rather"
+ + " than one that falls behind. Roughly, the source takes this fraction of a 50ms"
+ + " interval, so the default of 10ms leaves the thread four fifths of its time."
+ + " --readMaxElementsPerPoll bounds the same turn by count; whichever bound is reached"
+ + " first ends it, and a count alone cannot bound the time because how long an element"
+ + " takes depends on the pipeline below.")
+ @Default.Integer(10)
+ int getReadMaxPollTimeMs();
+
+ void setReadMaxPollTimeMs(int readMaxPollTimeMs);
+
+ @Description(
+ "How long the consumer group waits before deciding an instance has gone, in milliseconds."
+ + " This is the floor on how quickly work moves to another instance after one is lost,"
+ + " since a departed instance is not noticed any sooner. Kafka's default of 45s is kept,"
+ + " but a pipeline that values recovery over tolerance of a slow or briefly paused"
+ + " instance can lower it — a broker will not accept a value below its"
+ + " group.min.session.timeout.ms, which itself defaults to 6s.")
+ @Default.Integer(45_000)
+ int getSessionTimeoutMs();
+
+ void setSessionTimeoutMs(int sessionTimeoutMs);
+
+ @Description(
+ "Intended cap on how long a bundle may stay open, in milliseconds. NOT APPLIED YET: closing a"
+ + " bundle from a wall-clock punctuator made a pipeline with two chained GroupByKeys"
+ + " across several partitions emit its groups repeatedly against a real broker, so only"
+ + " the element-count bound is enforced for now. See"
+ + " https://github.com/apache/beam/issues/18479.")
+ @Default.Integer(1000)
+ int getMaxBundleTimeMs();
+
+ void setMaxBundleTimeMs(int maxBundleTimeMs);
+
+ @Description(
+ "How many partitions the runner gives the internal topics it creates to shuffle a pipeline"
+ + " through, which is the parallelism the shuffled parts of that pipeline can reach. A"
+ + " GroupByKey runs one task per partition of its repartition topic, so this is the"
+ + " number of instances its state and its downstream stages are spread over. Must be at"
+ + " least 1.")
+ @Default.Integer(1)
+ int getInternalParallelism();
+
+ void setInternalParallelism(int internalParallelism);
+
+ @Description("Replication factor for the internal topics the runner creates for a pipeline.")
+ @Default.Short(1)
+ short getTopicReplicationFactor();
+
+ void setTopicReplicationFactor(short topicReplicationFactor);
+
+ @Description(
+ "How many non-empty polls of an unbounded source to make before storing its checkpoint mark."
+ + " Taking a mark can be costly for some sources, so it is not worth doing on every poll;"
+ + " the cost of a larger value is that more elements are replayed after a restart, since"
+ + " the reader resumes from the last mark that was stored.")
+ @Default.Integer(10)
+ int getReadCheckpointNumBundles();
+
+ void setReadCheckpointNumBundles(int readCheckpointNumBundles);
+
+ @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.
+ *
+ *
The job name is included in the path so that multiple pipelines running on the same host
+ * (e.g. parallel tests) do not collide on the same Kafka Streams state directory and trigger a
+ * {@code LockException}.
+ */
+ class StateDirDefaultFactory implements DefaultValueFactory {
+ @Override
+ public String create(PipelineOptions options) {
+ return Paths.get(
+ System.getProperty("java.io.tmpdir"),
+ "beam-kafka-streams-state",
+ options.getJobName())
+ .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..65ff8b77b180
--- /dev/null
+++ b/runners/kafka-streams/src/main/java/org/apache/beam/runners/kafka/streams/KafkaStreamsPipelineResult.java
@@ -0,0 +1,77 @@
+/*
+ * 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 {
+ try {
+ return delegate.cancel();
+ } finally {
+ stopJobServer.run();
+ }
+ }
+
+ @Override
+ public State waitUntilFinish(Duration duration) {
+ State state = delegate.waitUntilFinish(duration);
+ // A null/non-terminal state means the wait timed out and the pipeline is still running;
+ // keep the job server alive so the caller can continue to interact with the job.
+ if (state != null && state.isTerminal()) {
+ stopJobServer.run();
+ }
+ return state;
+ }
+
+ @Override
+ public State waitUntilFinish() {
+ try {
+ return delegate.waitUntilFinish();
+ } finally {
+ stopJobServer.run();
+ }
+ }
+
+ @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..486cc05d8b54
--- /dev/null
+++ b/runners/kafka-streams/src/main/java/org/apache/beam/runners/kafka/streams/KafkaStreamsPipelineRunner.java
@@ -0,0 +1,185 @@
+/*
+ * 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.Properties;
+import java.util.UUID;
+import java.util.concurrent.atomic.AtomicReference;
+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;
+import org.apache.kafka.clients.consumer.ConsumerConfig;
+import org.apache.kafka.streams.KafkaStreams;
+import org.apache.kafka.streams.StreamsConfig;
+import org.apache.kafka.streams.Topology;
+import org.apache.kafka.streams.errors.StreamsUncaughtExceptionHandler;
+import org.checkerframework.checker.nullness.qual.Nullable;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/** Executes a portable pipeline by translating it to a Kafka Streams {@link Topology}. */
+public class KafkaStreamsPipelineRunner implements PortablePipelineRunner {
+
+ private static final Logger LOG = LoggerFactory.getLogger(KafkaStreamsPipelineRunner.class);
+
+ private final KafkaStreamsPipelineOptions pipelineOptions;
+
+ public KafkaStreamsPipelineRunner(KafkaStreamsPipelineOptions pipelineOptions) {
+ this.pipelineOptions = pipelineOptions;
+ }
+
+ @Override
+ public PortablePipelineResult run(RunnerApi.Pipeline pipeline, JobInfo jobInfo) {
+ // Only the options meaningful here are checked, not the whole interface: this runs on the job
+ // server, so the client-side options PortablePipelineOptions marks required — jobEndpoint above
+ // all — do not apply. Flink's PortablePipelineRunner does not validate here either.
+ checkRequiredOption("applicationId", pipelineOptions.getApplicationId());
+ checkRequiredOption("bootstrapServers", pipelineOptions.getBootstrapServers());
+ // Also the number of watermark reports a shuffle's consumer waits for, so a non-positive value
+ // would leave it waiting forever rather than failing.
+ if (pipelineOptions.getInternalParallelism() < 1) {
+ throw new IllegalArgumentException(
+ "--internalParallelism must be at least 1, but was "
+ + pipelineOptions.getInternalParallelism());
+ }
+
+ KafkaStreamsPipelineTranslator translator = new KafkaStreamsPipelineTranslator();
+ KafkaStreamsTranslationContext context =
+ translator.createTranslationContext(jobInfo, pipelineOptions);
+ RunnerApi.Pipeline prepared = translator.prepareForTranslation(pipeline);
+ translator.translate(context, prepared);
+
+ Topology topology = context.getTopology();
+ // The runner names its own bootstrap and repartition topics, which Kafka Streams treats as
+ // user topics and will not create; it refuses to start if a source topic is missing.
+ KafkaStreamsTopicManager.createMissingTopics(topology, pipelineOptions);
+ LOG.info(
+ "Translated pipeline {} into Kafka Streams topology:\n{}",
+ jobInfo.jobId(),
+ topology.describe());
+
+ KafkaStreams kafkaStreams = new KafkaStreams(topology, streamsConfig(jobInfo));
+ // Kafka Streams moves the client to ERROR and keeps the exception to itself, which left failed
+ // jobs saying only "unknown error". Keep the first failure so run() can rethrow it.
+ AtomicReference<@Nullable Throwable> failure = new AtomicReference<>();
+ kafkaStreams.setUncaughtExceptionHandler(
+ throwable -> {
+ failure.compareAndSet(null, throwable);
+ LOG.error("Pipeline {} failed", jobInfo.jobId(), throwable);
+ // A job with an owner waiting on it, not a service: a failure stops the client.
+ return StreamsUncaughtExceptionHandler.StreamThreadExceptionResponse.SHUTDOWN_CLIENT;
+ });
+ // Before start(): Kafka Streams only accepts a state listener while still in CREATED.
+ KafkaStreamsPortablePipelineResult result =
+ new KafkaStreamsPortablePipelineResult(
+ kafkaStreams,
+ context.getMetricsContainerStepMap(),
+ // Only once every task is initialized is the registered set complete, so that "all
+ // finished" can mean the pipeline is finished.
+ context.getTerminationTracker()::started);
+ // Kafka Streams has no notion of a finished pipeline, so the runner stops the client once every
+ // processor reaches the terminal watermark. Registered before start() so a fast drain is seen.
+ context
+ .getTerminationTracker()
+ .onAllTerminated(
+ () -> closeInBackground(kafkaStreams, jobInfo.jobId(), "the pipeline is drained"));
+ kafkaStreams.start();
+ // The job service reads the result's state once, when this method returns, so returning while
+ // the pipeline is still running would leave the job reported as RUNNING for good. Blocking here
+ // is what FlinkPipelineRunner does too, by blocking in executor.execute().
+ //
+ // A bounded pipeline unblocks this by draining: the processors report themselves terminated,
+ // the callback above stops the client, and the result's latch is released. A streaming pipeline
+ // never reaches the terminal watermark, so this blocks until the job is cancelled, which is the
+ // intended behaviour for a job that has no end.
+ result.waitUntilFinish();
+ if (Thread.currentThread().isInterrupted()) {
+ // Cancelled: the job service interrupts this thread, and the invocation future it would
+ // otherwise have used to cancel the result has already been cancelled with it. Stop the
+ // client so it does not outlive the job — from another thread, since close() waits on the
+ // stream threads and the joins it does would throw straight back out of an interrupted one.
+ closeInBackground(kafkaStreams, jobInfo.jobId(), "the job was cancelled");
+ }
+ Throwable thrown = failure.get();
+ if (thrown != null) {
+ // Thrown rather than returned as a failed result: the job service reads the state of what is
+ // returned, but only what is thrown carries a reason the user can act on.
+ throw new RuntimeException("Pipeline " + jobInfo.jobId() + " failed", thrown);
+ }
+ return result;
+ }
+
+ /**
+ * Stops the Kafka Streams client from a thread of its own.
+ *
+ *
Never called from a thread that {@code close()} itself waits for. When the pipeline drains,
+ * that is the task thread which reported the last termination; when the job is cancelled, it is
+ * the interrupted invocation thread. In both cases closing inline would either wait on the thread
+ * doing the closing or abandon the shutdown part-way.
+ */
+ private static void closeInBackground(KafkaStreams kafkaStreams, String jobId, String reason) {
+ Thread closer =
+ new Thread(
+ () -> {
+ LOG.info("Stopping the Kafka Streams client for job {}: {}", jobId, reason);
+ kafkaStreams.close();
+ },
+ "kafka-streams-runner-shutdown-" + jobId);
+ closer.setDaemon(true);
+ closer.start();
+ }
+
+ private static void checkRequiredOption(String name, @Nullable String value) {
+ if (value == null || value.isEmpty()) {
+ throw new IllegalArgumentException(
+ "Missing required pipeline option --" + name + " for the Kafka Streams runner");
+ }
+ }
+
+ // Visible for testing: the session timeout and the heartbeat derived from it.
+ Properties streamsConfig(JobInfo jobInfo) {
+ Properties props = new Properties();
+ props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, pipelineOptions.getBootstrapServers());
+ props.put(StreamsConfig.APPLICATION_ID_CONFIG, pipelineOptions.getApplicationId());
+ props.put(StreamsConfig.STATE_DIR_CONFIG, pipelineOptions.getStateDir());
+ props.put(StreamsConfig.PROCESSING_GUARANTEE_CONFIG, StreamsConfig.EXACTLY_ONCE_V2);
+ // The job id identifies the pipeline, which every instance of it shares, so on its own it does
+ // not identify an instance. Kafka Streams names threads, consumers and metrics after the client
+ // id, so two workers running the same job would produce logs and JMX metrics that cannot be
+ // told
+ // apart — in a deployment whose whole point is that you add workers. Keeping the job id as the
+ // prefix leaves the pipeline recognizable; the suffix is what makes each worker distinct, and
+ // is
+ // what Kafka Streams does by default when no client id is set.
+ props.put(StreamsConfig.CLIENT_ID_CONFIG, jobInfo.jobId() + "-" + UUID.randomUUID());
+ // How quickly a lost instance is noticed, which is the floor on how quickly its work moves
+ // elsewhere. The heartbeat must be shorter than the timeout, or a healthy instance would be
+ // declared dead between beats; a third is the ratio Kafka's own defaults use. Deriving it
+ // rather than exposing it keeps the pair consistent whatever the timeout is set to.
+ int sessionTimeoutMs = pipelineOptions.getSessionTimeoutMs();
+ props.put(
+ StreamsConfig.consumerPrefix(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG), sessionTimeoutMs);
+ props.put(
+ StreamsConfig.consumerPrefix(ConsumerConfig.HEARTBEAT_INTERVAL_MS_CONFIG),
+ Math.max(1, sessionTimeoutMs / 3));
+ return props;
+ }
+}
diff --git a/runners/kafka-streams/src/main/java/org/apache/beam/runners/kafka/streams/KafkaStreamsPortablePipelineResult.java b/runners/kafka-streams/src/main/java/org/apache/beam/runners/kafka/streams/KafkaStreamsPortablePipelineResult.java
new file mode 100644
index 000000000000..29c07917986f
--- /dev/null
+++ b/runners/kafka-streams/src/main/java/org/apache/beam/runners/kafka/streams/KafkaStreamsPortablePipelineResult.java
@@ -0,0 +1,154 @@
+/*
+ * 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.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import org.apache.beam.model.jobmanagement.v1.JobApi;
+import org.apache.beam.runners.core.metrics.MetricsContainerStepMap;
+import org.apache.beam.runners.jobsubmission.PortablePipelineResult;
+import org.apache.beam.sdk.metrics.MetricResults;
+import org.apache.kafka.streams.KafkaStreams;
+import org.joda.time.Duration;
+
+/**
+ * Result of executing a portable pipeline as a {@link KafkaStreams} application.
+ *
+ *
Translates the underlying {@link KafkaStreams.State} into Beam's {@link
+ * org.apache.beam.sdk.PipelineResult.State} and forwards {@link #cancel()} / {@link
+ * #waitUntilFinish()} to the {@code KafkaStreams} instance.
+ */
+class KafkaStreamsPortablePipelineResult implements PortablePipelineResult {
+
+ private final KafkaStreams kafkaStreams;
+ // The job's metrics accumulator, shared by reference with the topology's stage processors, which
+ // update it as the SDK harness reports bundle metrics.
+ private final MetricsContainerStepMap metricsContainerStepMap;
+ private final CountDownLatch terminated = new CountDownLatch(1);
+ private volatile boolean cancelled = false;
+
+ /**
+ * Must be constructed before {@link KafkaStreams#start()} is called: it registers a state
+ * listener, and Kafka Streams rejects one once the application has left the CREATED state.
+ */
+ KafkaStreamsPortablePipelineResult(
+ KafkaStreams kafkaStreams,
+ MetricsContainerStepMap metricsContainerStepMap,
+ Runnable onRunning) {
+ this.kafkaStreams = kafkaStreams;
+ this.metricsContainerStepMap = metricsContainerStepMap;
+ kafkaStreams.setStateListener(
+ (newState, oldState) -> {
+ if (newState == KafkaStreams.State.RUNNING) {
+ onRunning.run();
+ }
+ if (newState == KafkaStreams.State.NOT_RUNNING || newState == KafkaStreams.State.ERROR) {
+ terminated.countDown();
+ }
+ });
+ // Guard against the race where the KafkaStreams instance transitions to a terminal state
+ // (e.g. immediate startup failure) before the state listener is registered above. Without
+ // this check, the latch would never be counted down and waitUntilFinish() would block forever.
+ KafkaStreams.State current = kafkaStreams.state();
+ if (current == KafkaStreams.State.NOT_RUNNING || current == KafkaStreams.State.ERROR) {
+ terminated.countDown();
+ }
+ }
+
+ @Override
+ public State getState() {
+ if (cancelled) {
+ return State.CANCELLED;
+ }
+ return mapState(kafkaStreams.state());
+ }
+
+ @Override
+ public State cancel() throws IOException {
+ cancelled = true;
+ kafkaStreams.close();
+ terminated.countDown();
+ return getState();
+ }
+
+ @Override
+ public State waitUntilFinish(Duration duration) {
+ try {
+ boolean reachedTerminal = terminated.await(duration.getMillis(), TimeUnit.MILLISECONDS);
+ if (!reachedTerminal) {
+ return getState();
+ }
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ return State.UNKNOWN;
+ }
+ return getState();
+ }
+
+ @Override
+ public State waitUntilFinish() {
+ try {
+ terminated.await();
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ return State.UNKNOWN;
+ }
+ return getState();
+ }
+
+ @Override
+ public MetricResults metrics() {
+ // Attempted values only: the runner does not distinguish committed results yet (that needs
+ // metrics to be folded into the exactly-once commit, which lands with the durability work).
+ return MetricsContainerStepMap.asAttemptedOnlyMetricResults(metricsContainerStepMap);
+ }
+
+ @Override
+ public JobApi.MetricResults portableMetrics() throws UnsupportedOperationException {
+ // How a pipeline from another SDK reads its metrics. The job service asks for these once the
+ // job is terminal and returns them over the job API. Without it a Python pipeline saw no
+ // metrics at all, even though the same values were already available to a Java one.
+ //
+ // Reported as attempted only, and deliberately not also as committed: the values are what the
+ // SDK harness reported per bundle, which is not tied to the commit of the records that produced
+ // them. Committed metrics are https://github.com/apache/beam/issues/39635.
+ return JobApi.MetricResults.newBuilder()
+ .addAllAttempted(metricsContainerStepMap.getMonitoringInfos())
+ .build();
+ }
+
+ private static State mapState(KafkaStreams.State state) {
+ switch (state) {
+ case CREATED:
+ case REBALANCING:
+ return State.RUNNING;
+ case RUNNING:
+ return State.RUNNING;
+ case PENDING_SHUTDOWN:
+ return State.CANCELLED;
+ case PENDING_ERROR:
+ case ERROR:
+ return State.FAILED;
+ case NOT_RUNNING:
+ return State.DONE;
+ default:
+ return State.UNKNOWN;
+ }
+ }
+}
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..924c3ac01fe8
--- /dev/null
+++ b/runners/kafka-streams/src/main/java/org/apache/beam/runners/kafka/streams/KafkaStreamsRunner.java
@@ -0,0 +1,138 @@
+/*
+ * 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.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.sdk.util.construction.SplittableParDo;
+import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.annotations.VisibleForTesting;
+import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.Strings;
+import org.checkerframework.checker.nullness.qual.Nullable;
+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.
+ *
+ *
This runner is experimental. It executes a subset of the Beam model correctly — the
+ * parts it supports are covered by Beam's {@code @ValidatesRunner} suite — but several capabilities
+ * that are core to the model are not implemented yet, among them side inputs, stateful {@code
+ * ParDo} and user timers, merging windows, custom {@code WindowFn}s and splittable {@code DoFn}.
+ * Its behaviour and its pipeline options may change. See the runner documentation for
+ * what is and is not supported, and #18479 for the work that remains.
+ */
+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) {
+ prepareForTranslation(pipeline, pipelineOptions);
+ @Nullable 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) {
+ KafkaStreamsJobServerDriver driverForStop = jobServerDriver;
+ return new KafkaStreamsPipelineResult(result, driverForStop::stop);
+ }
+ return result;
+ } catch (Exception e) {
+ if (jobServerDriver != null) {
+ jobServerDriver.stop();
+ }
+ if (e instanceof RuntimeException) {
+ throw (RuntimeException) e;
+ }
+ throw new RuntimeException(e);
+ }
+ }
+
+ /**
+ * Settles the options the runner needs and rewrites the pipeline into what it can translate.
+ *
+ *
The runner does not translate splittable DoFns, and a {@link org.apache.beam.sdk.io.Read}
+ * expands into one by default, so a pipeline that merely reads would otherwise fail to translate.
+ * Beam keeps the primitive read for exactly this case, behind an experiment that {@link
+ * #assignPortableDefaults} sets, so a pipeline does not have to ask for it and the proto that
+ * reaches the job server already holds primitive reads.
+ */
+ @VisibleForTesting
+ static void prepareForTranslation(
+ Pipeline pipeline, KafkaStreamsPipelineOptions pipelineOptions) {
+ assignPortableDefaults(pipelineOptions);
+ SplittableParDo.convertReadBasedSplittableDoFnsToPrimitiveReadsIfNecessary(pipeline);
+ }
+
+ private static void assignPortableDefaults(KafkaStreamsPipelineOptions pipelineOptions) {
+ if (Strings.isNullOrEmpty(pipelineOptions.getDefaultEnvironmentType())) {
+ pipelineOptions.setDefaultEnvironmentType(Environments.ENVIRONMENT_LOOPBACK);
+ }
+ ExperimentalOptions experimentalOptions = pipelineOptions.as(ExperimentalOptions.class);
+ @Nullable List existingExperiments = experimentalOptions.getExperiments();
+ List experiments =
+ existingExperiments == null ? new ArrayList<>() : new ArrayList<>(existingExperiments);
+ boolean changed = false;
+ if (!experiments.contains("beam_fn_api")) {
+ experiments.add("beam_fn_api");
+ changed = true;
+ }
+ // Splittable DoFns are not translated, so the Read that expands into one has to stay the
+ // primitive it used to be. This is the experiment Beam looks for when deciding that.
+ if (!experiments.contains("use_deprecated_read")) {
+ experiments.add("use_deprecated_read");
+ changed = true;
+ }
+ if (changed) {
+ 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/KafkaStreamsTopicManager.java b/runners/kafka-streams/src/main/java/org/apache/beam/runners/kafka/streams/KafkaStreamsTopicManager.java
new file mode 100644
index 000000000000..0e5f46f53eac
--- /dev/null
+++ b/runners/kafka-streams/src/main/java/org/apache/beam/runners/kafka/streams/KafkaStreamsTopicManager.java
@@ -0,0 +1,171 @@
+/*
+ * 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.ArrayList;
+import java.util.Collection;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Properties;
+import java.util.Set;
+import java.util.concurrent.ExecutionException;
+import org.apache.kafka.clients.admin.Admin;
+import org.apache.kafka.clients.admin.AdminClientConfig;
+import org.apache.kafka.clients.admin.NewTopic;
+import org.apache.kafka.common.errors.TopicExistsException;
+import org.apache.kafka.streams.Topology;
+import org.apache.kafka.streams.TopologyDescription;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Creates the topics a translated pipeline needs before the Kafka Streams application starts.
+ *
+ *
The runner shuffles data through topics it names itself: a bootstrap topic per Impulse and per
+ * primitive Read, and a repartition topic per GroupByKey. Kafka Streams does create the internal
+ * topics it manages on its own, but these are declared with explicit names through {@code
+ * addSource} and {@code addSink}, so to Kafka Streams they are ordinary user topics — it will not
+ * create them, and refuses to start with {@code MissingSourceTopicException} if a source topic is
+ * absent. Relying on the broker's {@code auto.create.topics.enable} is not an option either: it is
+ * off on many clusters, and a topic auto-created on first fetch gets the broker's default partition
+ * count rather than the pipeline's.
+ *
+ *
Only topics carrying one of the runner's own prefixes are created. Any other topic in the
+ * topology belongs to the user (a source or sink they named), and creating those implicitly would
+ * hide a misconfiguration behind an empty topic.
+ */
+class KafkaStreamsTopicManager {
+
+ private static final Logger LOG = LoggerFactory.getLogger(KafkaStreamsTopicManager.class);
+
+ /**
+ * Prefixes of the bootstrap topics, which must have exactly one partition.
+ *
+ *
An Impulse or a primitive Read emits its elements once per task, gated by a state store that
+ * is itself per task. Kafka Streams creates one task per partition of the source topic, so a
+ * bootstrap topic with several partitions would make the same Impulse fire once per partition and
+ * the same source be read once per partition.
+ */
+ private static final List SINGLE_PARTITION_TOPIC_PREFIXES =
+ java.util.Arrays.asList("__beam_impulse_", "__beam_read_");
+
+ /**
+ * Prefixes of the topics whose partition count sets the pipeline's parallelism — the repartition
+ * topic a GroupByKey shuffles through.
+ */
+ private static final List PARTITIONED_TOPIC_PREFIXES =
+ java.util.Arrays.asList("__beam_gbk_");
+
+ private KafkaStreamsTopicManager() {}
+
+ /**
+ * Creates any runner-owned topic in {@code topology} that does not exist yet.
+ *
+ *
Safe to run concurrently with another instance of the same job: a topic that appears between
+ * the existence check and the create request surfaces as {@link TopicExistsException}, which is
+ * treated as success.
+ */
+ static void createMissingTopics(Topology topology, KafkaStreamsPipelineOptions options) {
+ Set runnerTopics = runnerOwnedTopics(topology);
+ if (runnerTopics.isEmpty()) {
+ return;
+ }
+ Properties adminConfig = new Properties();
+ adminConfig.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, options.getBootstrapServers());
+ try (Admin admin = Admin.create(adminConfig)) {
+ Set existing = admin.listTopics().names().get();
+ List toCreate = new ArrayList<>();
+ for (String topic : runnerTopics) {
+ if (!existing.contains(topic)) {
+ toCreate.add(
+ new NewTopic(
+ topic, partitionsFor(topic, options), options.getTopicReplicationFactor()));
+ }
+ }
+ if (toCreate.isEmpty()) {
+ return;
+ }
+ LOG.info("Creating {} runner-owned topic(s): {}", toCreate.size(), toCreate);
+ createAll(admin, toCreate);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new RuntimeException("Interrupted while creating the pipeline's Kafka topics", e);
+ } catch (ExecutionException e) {
+ throw new RuntimeException("Failed to create the pipeline's Kafka topics", e);
+ }
+ }
+
+ private static void createAll(Admin admin, Collection topics)
+ throws InterruptedException, ExecutionException {
+ try {
+ admin.createTopics(topics).all().get();
+ } catch (ExecutionException e) {
+ // Another instance of the same application may have created them first, which is fine.
+ if (!(e.getCause() instanceof TopicExistsException)) {
+ throw e;
+ }
+ LOG.debug("Some topics already existed; another instance created them first", e);
+ }
+ }
+
+ /** The topics in the topology that the runner named, and so is responsible for creating. */
+ private static Set runnerOwnedTopics(Topology topology) {
+ Set topics = new HashSet<>();
+ for (TopologyDescription.Subtopology subtopology : topology.describe().subtopologies()) {
+ for (TopologyDescription.Node node : subtopology.nodes()) {
+ if (node instanceof TopologyDescription.Source) {
+ Set sourceTopics = ((TopologyDescription.Source) node).topicSet();
+ if (sourceTopics != null) {
+ topics.addAll(sourceTopics);
+ }
+ } else if (node instanceof TopologyDescription.Sink) {
+ String topic = ((TopologyDescription.Sink) node).topic();
+ if (topic != null) {
+ topics.add(topic);
+ }
+ }
+ }
+ }
+ topics.removeIf(topic -> !isRunnerOwned(topic));
+ return topics;
+ }
+
+ /**
+ * The partition count a runner-owned topic is created with: one for a bootstrap topic, and the
+ * configured parallelism for a shuffle topic.
+ */
+ private static int partitionsFor(String topic, KafkaStreamsPipelineOptions options) {
+ return hasAnyPrefix(topic, SINGLE_PARTITION_TOPIC_PREFIXES)
+ ? 1
+ : options.getInternalParallelism();
+ }
+
+ private static boolean isRunnerOwned(String topic) {
+ return hasAnyPrefix(topic, SINGLE_PARTITION_TOPIC_PREFIXES)
+ || hasAnyPrefix(topic, PARTITIONED_TOPIC_PREFIXES);
+ }
+
+ private static boolean hasAnyPrefix(String topic, List prefixes) {
+ for (String prefix : prefixes) {
+ if (topic.startsWith(prefix)) {
+ return true;
+ }
+ }
+ return false;
+ }
+}
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/EmptyBoundedSource.java b/runners/kafka-streams/src/main/java/org/apache/beam/runners/kafka/streams/translation/EmptyBoundedSource.java
new file mode 100644
index 000000000000..14bee0c3cfd0
--- /dev/null
+++ b/runners/kafka-streams/src/main/java/org/apache/beam/runners/kafka/streams/translation/EmptyBoundedSource.java
@@ -0,0 +1,89 @@
+/*
+ * 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.io.IOException;
+import java.util.Collections;
+import java.util.List;
+import java.util.NoSuchElementException;
+import org.apache.beam.sdk.coders.ByteArrayCoder;
+import org.apache.beam.sdk.coders.Coder;
+import org.apache.beam.sdk.io.BoundedSource;
+import org.apache.beam.sdk.options.PipelineOptions;
+
+/**
+ * A {@link BoundedSource} with no elements. The runner substitutes it for a {@code Flatten} of zero
+ * PCollections (see {@link KafkaStreamsPipelineTranslator}): reading it produces no data and a
+ * terminal watermark — exactly the semantics of an empty PCollection. The element type is never
+ * observed since no element is ever produced.
+ */
+class EmptyBoundedSource extends BoundedSource {
+
+ @Override
+ public List> split(
+ long desiredBundleSizeBytes, PipelineOptions options) {
+ return Collections.singletonList(this);
+ }
+
+ @Override
+ public long getEstimatedSizeBytes(PipelineOptions options) {
+ return 0;
+ }
+
+ @Override
+ public BoundedReader createReader(PipelineOptions options) {
+ return new EmptyReader(this);
+ }
+
+ @Override
+ public Coder getOutputCoder() {
+ return ByteArrayCoder.of();
+ }
+
+ /** A reader that is exhausted from the start. */
+ private static final class EmptyReader extends BoundedReader {
+ private final EmptyBoundedSource source;
+
+ EmptyReader(EmptyBoundedSource source) {
+ this.source = source;
+ }
+
+ @Override
+ public boolean start() {
+ return false;
+ }
+
+ @Override
+ public boolean advance() {
+ return false;
+ }
+
+ @Override
+ public byte[] getCurrent() throws NoSuchElementException {
+ throw new NoSuchElementException("EmptyBoundedSource has no elements");
+ }
+
+ @Override
+ public void close() throws IOException {}
+
+ @Override
+ public BoundedSource getCurrentSource() {
+ return source;
+ }
+ }
+}
diff --git a/runners/kafka-streams/src/main/java/org/apache/beam/runners/kafka/streams/translation/ExecutableStageProcessor.java b/runners/kafka-streams/src/main/java/org/apache/beam/runners/kafka/streams/translation/ExecutableStageProcessor.java
new file mode 100644
index 000000000000..5cb0672ff9fc
--- /dev/null
+++ b/runners/kafka-streams/src/main/java/org/apache/beam/runners/kafka/streams/translation/ExecutableStageProcessor.java
@@ -0,0 +1,365 @@
+/*
+ * 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.Queue;
+import java.util.Set;
+import java.util.concurrent.ConcurrentLinkedQueue;
+import org.apache.beam.model.fnexecution.v1.BeamFnApi.ProcessBundleProgressResponse;
+import org.apache.beam.model.fnexecution.v1.BeamFnApi.ProcessBundleResponse;
+import org.apache.beam.model.pipeline.v1.RunnerApi;
+import org.apache.beam.runners.core.metrics.MetricsContainerImpl;
+import org.apache.beam.runners.fnexecution.control.BundleProgressHandler;
+import org.apache.beam.runners.fnexecution.control.ExecutableStageContext;
+import org.apache.beam.runners.fnexecution.control.OutputReceiverFactory;
+import org.apache.beam.runners.fnexecution.control.RemoteBundle;
+import org.apache.beam.runners.fnexecution.control.StageBundleFactory;
+import org.apache.beam.runners.fnexecution.provisioning.JobInfo;
+import org.apache.beam.runners.fnexecution.state.StateRequestHandler;
+import org.apache.beam.sdk.fn.data.FnDataReceiver;
+import org.apache.beam.sdk.transforms.windowing.BoundedWindow;
+import org.apache.beam.sdk.util.construction.graph.ExecutableStage;
+import org.apache.beam.sdk.values.WindowedValue;
+import org.apache.beam.vendor.grpc.v1p69p0.com.google.protobuf.TextFormat;
+import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.ImmutableMap;
+import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.Iterables;
+import org.apache.kafka.streams.processor.api.Processor;
+import org.apache.kafka.streams.processor.api.ProcessorContext;
+import org.apache.kafka.streams.processor.api.Record;
+import org.checkerframework.checker.nullness.qual.Nullable;
+import org.joda.time.Instant;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Kafka Streams {@link Processor} that executes a fused {@link ExecutableStage} — stateless user
+ * code such as ParDo — in the Beam SDK harness over the Fn API.
+ *
+ *
Each {@link KStreamsPayload#isData() data} payload is unwrapped and fed to the harness through
+ * the stage's main input {@link FnDataReceiver}. Harness outputs are collected on the harness
+ * threads into {@link #pendingOutputs} and flushed downstream when the bundle closes, because
+ * {@link ProcessorContext#forward} may only be called from the processing thread.
+ *
+ *
A {@link KStreamsPayload#isWatermark() watermark} payload marks a bundle boundary: the open
+ * bundle is closed and flushed, the report goes to the {@link WatermarkAggregator}, and the stage's
+ * output watermark is forwarded — stamped with this stage's transform id — only once the aggregate
+ * across the upstream partitions advances. Until every partition has reported the watermark is
+ * held, though data is still processed meanwhile.
+ *
+ *
A bundle is also bounded by {@code --maxBundleSize}, checked as elements arrive; without it a
+ * bundle would stay open until the next watermark and grow without limit on a steady stream. The
+ * time bound {@code --maxBundleTimeMs} is not applied yet, see that option's documentation.
+ *
+ *
Closing a bundle asks Kafka Streams to commit, so the elements consumed and the records
+ * produced commit together and a restart replays all of a bundle or none. This aligns commits to
+ * bundle boundaries but does not stop Kafka Streams committing on its own interval mid-bundle;
+ * ruling that out needs a pre-commit hook.
+ *
+ *
The analogue of Flink's {@code ExecutableStageDoFnOperator} and Spark's {@code
+ * SparkExecutableStageFunction}. State, timers and side inputs are out of scope here: the stage
+ * runs with {@link StateRequestHandler#unsupported()} and no timer receivers.
+ */
+class ExecutableStageProcessor
+ implements Processor, byte[], KStreamsPayload> {
+
+ private static final Logger LOG = LoggerFactory.getLogger(ExecutableStageProcessor.class);
+
+ private final RunnerApi.ExecutableStagePayload stagePayload;
+ private final JobInfo jobInfo;
+ // Stamped on every watermark forwarded, so downstream aggregators know which transform reported.
+ private final String transformId;
+ // Updated from the MonitoringInfos the harness reports as each bundle completes.
+ private final MetricsContainerImpl metricsContainer;
+
+ // Enqueued by harness threads and drained by the processing thread on bundle close, so it must
+ // be thread-safe. Each entry carries its output PCollection id for routing on flush. The element
+ // type is wildcarded: coders are applied by the bundle factory at the Fn-API boundary.
+ private final Queue pendingOutputs = new ConcurrentLinkedQueue<>();
+ // Output PCollection id -> relay child node. Empty for a single-output stage.
+ private final Map outputChildByPCollectionId;
+
+ // Holds until every partition of the upstream transform has reported; see WatermarkAggregator.
+ private final WatermarkAggregator watermarkAggregator;
+ // Reports this stage finished at the terminal watermark, so a bounded pipeline can stop.
+ private final TerminationReporter terminationReporter;
+ // The last watermark actually forwarded downstream, so we only forward when it advances.
+ private Instant lastForwardedWatermark = BoundedWindow.TIMESTAMP_MIN_VALUE;
+
+ private @Nullable ProcessorContext> context;
+ private @Nullable ExecutableStageContext stageContext;
+ private @Nullable StageBundleFactory stageBundleFactory;
+ private @Nullable RemoteBundle currentBundle;
+
+ /** Bound on how many elements may be fed to one bundle. */
+ private final int maxBundleSize;
+
+ /** Elements fed to the open bundle, for the size bound above. */
+ private int elementsInBundle;
+
+ /**
+ * @param transformId this stage's own transform id, stamped on the watermarks it emits
+ * @param upstreamTransformIds the transform ids feeding this stage (known from the pipeline
+ * graph), whose reports the {@link WatermarkAggregator} waits for
+ * @param metricsContainer this stage's container in the job's metrics step map, updated with the
+ * harness's per-bundle MonitoringInfos
+ */
+ ExecutableStageProcessor(
+ RunnerApi.ExecutableStagePayload stagePayload,
+ JobInfo jobInfo,
+ String transformId,
+ Set upstreamTransformIds,
+ MetricsContainerImpl metricsContainer,
+ Map outputChildByPCollectionId,
+ int maxBundleSize,
+ TerminationTracker terminationTracker) {
+ this.stagePayload = stagePayload;
+ this.jobInfo = jobInfo;
+ this.transformId = transformId;
+ this.watermarkAggregator = new WatermarkAggregator(upstreamTransformIds);
+ this.metricsContainer = metricsContainer;
+ this.outputChildByPCollectionId = ImmutableMap.copyOf(outputChildByPCollectionId);
+ this.maxBundleSize = maxBundleSize;
+ this.terminationReporter = new TerminationReporter(terminationTracker, transformId);
+ }
+
+ /** A harness output element together with the id of the output PCollection it belongs to. */
+ private static final class PendingOutput {
+ final String pCollectionId;
+ final WindowedValue value;
+
+ PendingOutput(String pCollectionId, WindowedValue value) {
+ this.pCollectionId = pCollectionId;
+ this.value = value;
+ }
+ }
+
+ @Override
+ public void init(ProcessorContext> context) {
+ this.context = context;
+ terminationReporter.init(context);
+ // Created lazily on the first data element, so a stage that only forwards watermarks never
+ // spins up a harness. Spark's SparkExecutableStageFunction does the same.
+ }
+
+ private void ensureStageBundleFactory() {
+ if (stageBundleFactory != null) {
+ return;
+ }
+ ExecutableStage executableStage = ExecutableStage.fromPayload(stagePayload);
+ stageContext = KafkaStreamsExecutableStageContextFactory.getInstance().get(jobInfo);
+ stageBundleFactory = stageContext.getStageBundleFactory(executableStage);
+ }
+
+ @Override
+ public void process(Record> record) {
+ KStreamsPayload payload = record.value();
+ if (payload == null) {
+ // A topic can always be written to from outside, so warn and drop rather than crash.
+ LOG.warn(
+ "Stage {} dropping record with null payload (external write or tombstone)", transformId);
+ return;
+ }
+ if (payload.isWatermark()) {
+ // Flush buffered outputs before the watermark. Data is processed regardless of readiness;
+ // only the watermark waits for every source partition.
+ closeBundleAndFlush(record);
+ // Forward the output watermark only when the aggregate across upstream partitions advances.
+ watermarkAggregator.observe(payload.asWatermark());
+ Instant advanced = watermarkAggregator.advance();
+ if (advanced.isAfter(lastForwardedWatermark)) {
+ lastForwardedWatermark = advanced;
+ forwardWatermark(record, advanced.getMillis());
+ }
+ return;
+ }
+ try {
+ ensureBundleOpen();
+ mainInputReceiver().accept(payload.getData());
+ elementsInBundle++;
+ } catch (Exception e) {
+ throw new RuntimeException("Failed to process element through SDK harness", e);
+ }
+ if (elementsInBundle >= maxBundleSize) {
+ closeBundleAndFlush(record);
+ }
+ }
+
+ private void ensureBundleOpen() throws Exception {
+ if (currentBundle != null) {
+ return;
+ }
+ ensureStageBundleFactory();
+ StageBundleFactory factory = checkInitialized(stageBundleFactory);
+ OutputReceiverFactory outputReceiverFactory =
+ new OutputReceiverFactory() {
+ @Override
+ public FnDataReceiver create(String pCollectionId) {
+ // Queued on harness threads, drained on the processing thread after the bundle closes.
+ return receivedElement -> {
+ if (receivedElement != null) {
+ pendingOutputs.add(
+ new PendingOutput(pCollectionId, (WindowedValue) receivedElement));
+ }
+ };
+ }
+ };
+ // Fold the harness's reported metrics into this stage's container when each bundle completes.
+ // Only the completion response is applied: it carries the bundle's final cumulative values, and
+ // the container's update() adds counter values, so also applying mid-bundle progress snapshots
+ // would double-count them. Live mid-bundle metrics can come later if a use appears.
+ BundleProgressHandler progressHandler =
+ new BundleProgressHandler() {
+ @Override
+ public void onProgress(ProcessBundleProgressResponse progress) {
+ // Deliberately not folded into the container; see comment above.
+ if (LOG.isDebugEnabled()) {
+ LOG.debug(
+ "Stage {} bundle progress: {}",
+ transformId,
+ TextFormat.printer().printToString(progress));
+ }
+ }
+
+ @Override
+ public void onCompleted(ProcessBundleResponse response) {
+ if (LOG.isDebugEnabled()) {
+ LOG.debug(
+ "Stage {} bundle completed: {}",
+ transformId,
+ TextFormat.printer().printToString(response));
+ }
+ metricsContainer.update(response.getMonitoringInfosList());
+ }
+ };
+ currentBundle =
+ factory.getBundle(
+ outputReceiverFactory, StateRequestHandler.unsupported(), progressHandler);
+ elementsInBundle = 0;
+ }
+
+ private FnDataReceiver> mainInputReceiver() {
+ RemoteBundle bundle = checkInitialized(currentBundle);
+ @SuppressWarnings("unchecked")
+ FnDataReceiver> receiver =
+ (FnDataReceiver>)
+ (FnDataReceiver) Iterables.getOnlyElement(bundle.getInputReceivers().values());
+ return receiver;
+ }
+
+ /**
+ * Finishes the open bundle, forwards everything it produced, and asks Kafka Streams to commit.
+ *
+ *
The commit request is what ties a bundle to a transaction: the elements the bundle consumed
+ * and the records it produced are then committed together, so a restart either replays the whole
+ * bundle or none of it.
+ *
+ *
The outputs carry the key of the record that closed the bundle. An executable stage is
+ * unkeyed — it runs stateless, with no state or timers — so the Kafka record key means nothing to
+ * it and is only being carried along; where the key does matter, downstream sets it, as {@link
+ * ShuffleByKeyProcessor} does from the Beam key before a GroupByKey.
+ */
+ private void closeBundleAndFlush(Record> record) {
+ RemoteBundle bundle = currentBundle;
+ if (bundle == null) {
+ return;
+ }
+ try {
+ // close() blocks until the harness finishes the bundle and all outputs have been delivered
+ // to the output receiver (and hence enqueued in pendingOutputs).
+ bundle.close();
+ } catch (Exception e) {
+ throw new RuntimeException("Failed to close SDK harness bundle", e);
+ } finally {
+ currentBundle = null;
+ elementsInBundle = 0;
+ }
+ ProcessorContext> ctx = checkInitialized(context);
+ // The harness has finished the bundle (close() returned) so no further enqueues happen.
+ // Drain via poll() so each element is removed as it is forwarded. Each output is routed to its
+ // own output's relay child for a multi-output stage; a single-output stage forwards directly to
+ // its one downstream (empty routing map).
+ PendingOutput output;
+ while ((output = pendingOutputs.poll()) != null) {
+ Record> outputRecord =
+ new Record>(
+ record.key(), KStreamsPayload.data(output.value), record.timestamp());
+ String childNode = outputChildByPCollectionId.get(output.pCollectionId);
+ if (childNode == null) {
+ ctx.forward(outputRecord);
+ } else {
+ ctx.forward(outputRecord, childNode);
+ }
+ }
+ ctx.commit();
+ }
+
+ private void forwardWatermark(Record> record, long watermarkMillis) {
+ // Labelled as the only source a consumer will see. Forwarding here is in-process, to the
+ // stage's
+ // fused children, so exactly one instance of this stage reaches each of them. Where the output
+ // instead crosses a shuffle, ShuffleByKeyProcessor relabels the report with the real partition
+ // identity, because the broadcast then delivers every instance's report to every consumer.
+ ProcessorContext> ctx = checkInitialized(context);
+ ctx.forward(
+ new Record>(
+ record.key(),
+ KStreamsPayload.watermark(watermarkMillis, transformId, 0, 1),
+ record.timestamp()));
+ terminationReporter.watermarkEmitted(ctx, watermarkMillis);
+ }
+
+ @Override
+ public void close() {
+ try {
+ if (currentBundle != null) {
+ currentBundle.close();
+ currentBundle = null;
+ }
+ } catch (Exception e) {
+ LOG.warn("Error closing in-flight SDK harness bundle", e);
+ }
+ try {
+ if (stageBundleFactory != null) {
+ stageBundleFactory.close();
+ stageBundleFactory = null;
+ }
+ } catch (Exception e) {
+ LOG.warn("Error closing stage bundle factory", e);
+ }
+ try {
+ if (stageContext != null) {
+ stageContext.close();
+ stageContext = null;
+ }
+ } catch (Exception e) {
+ LOG.warn("Error closing executable stage context", e);
+ }
+ // Last: this is what stops the pipeline waiting on this stage, and closing the bundle above can
+ // still forward records downstream. Releasing it first would let the pipeline be declared
+ // finished while this stage was flushing.
+ terminationReporter.close();
+ }
+
+ private static T checkInitialized(@Nullable T value) {
+ if (value == null) {
+ throw new IllegalStateException("ExecutableStageProcessor used before init()");
+ }
+ return value;
+ }
+}
diff --git a/runners/kafka-streams/src/main/java/org/apache/beam/runners/kafka/streams/translation/ExecutableStageTranslator.java b/runners/kafka-streams/src/main/java/org/apache/beam/runners/kafka/streams/translation/ExecutableStageTranslator.java
new file mode 100644
index 000000000000..11f3192befdf
--- /dev/null
+++ b/runners/kafka-streams/src/main/java/org/apache/beam/runners/kafka/streams/translation/ExecutableStageTranslator.java
@@ -0,0 +1,133 @@
+/*
+ * 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.io.IOException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import org.apache.beam.model.pipeline.v1.RunnerApi;
+import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.ImmutableSet;
+import org.apache.kafka.streams.Topology;
+
+/**
+ * Translates the {@code beam:runner:executable_stage:v1} URN.
+ *
+ *
Adds an {@link ExecutableStageProcessor} node to the topology, wired to the processor that
+ * produces the stage's input PCollection (resolved through {@link
+ * KafkaStreamsTranslationContext#getProcessorNameForPCollection}). The processor runs the fused
+ * user code in the SDK harness; its single output PCollection is registered so downstream
+ * translators can attach to this node.
+ *
+ *
Multi-output stages (additional outputs / side inputs / state / timers) are out of scope for
+ * this first version and are rejected so the limitation fails fast rather than silently dropping
+ * outputs.
+ */
+class ExecutableStageTranslator implements PTransformTranslator {
+
+ @Override
+ public void translate(
+ String transformId, RunnerApi.Pipeline pipeline, KafkaStreamsTranslationContext context) {
+ RunnerApi.PTransform transform = pipeline.getComponents().getTransformsOrThrow(transformId);
+
+ RunnerApi.ExecutableStagePayload stagePayload;
+ try {
+ stagePayload = RunnerApi.ExecutableStagePayload.parseFrom(transform.getSpec().getPayload());
+ } catch (IOException e) {
+ throw new IllegalArgumentException(
+ "Failed to parse ExecutableStagePayload for transform " + transformId, e);
+ }
+
+ // Fail fast on stage features that are not yet supported, so users get a clear message rather
+ // than a silent miss further down the harness/topology path.
+ if (stagePayload.getSideInputsCount() > 0) {
+ throw new UnsupportedOperationException(
+ "ExecutableStage "
+ + transformId
+ + " has side inputs; side inputs are not yet supported by the Kafka Streams runner.");
+ }
+ if (stagePayload.getUserStatesCount() > 0 || stagePayload.getTimersCount() > 0) {
+ throw new UnsupportedOperationException(
+ "ExecutableStage "
+ + transformId
+ + " uses user state or timers; stateful ParDo is not yet supported by the Kafka"
+ + " Streams runner.");
+ }
+ // The payload distinguishes the main input from side inputs, so reading it from the payload
+ // is unambiguous even before we add side-input support.
+ String inputPCollectionId = stagePayload.getInput();
+ String parentProcessor = context.getProcessorNameForPCollection(inputPCollectionId);
+ // A fused stage runs wherever its input runs: same task, so same partition identity.
+ int partitionCount = context.getPartitionCount(inputPCollectionId);
+
+ // A multi-output stage (a DoFn with side outputs, or a Read whose SDF wrapper produces several
+ // outputs) needs each output routed to the right downstream. Since downstream transforms are
+ // wired to a producer node by PCollection id, and that node must exist when the stage is
+ // translated, give each output its own relay node (StageOutputProcessor) and route to it by
+ // name. A single-output stage needs none of this — it forwards to its one downstream directly
+ // and registers itself as that output's producer. Outputs are sorted so the routing is
+ // deterministic across topology builds.
+ List outputPCollectionIds = new ArrayList<>(transform.getOutputsMap().values());
+ Collections.sort(outputPCollectionIds);
+ boolean multiOutput = outputPCollectionIds.size() > 1;
+ Map outputChildByPCollectionId = new LinkedHashMap<>();
+ if (multiOutput) {
+ for (int i = 0; i < outputPCollectionIds.size(); i++) {
+ outputChildByPCollectionId.put(outputPCollectionIds.get(i), transformId + "-output-" + i);
+ }
+ }
+
+ Topology topology = context.getTopology();
+ // The stage stamps its own transform id on the watermarks it emits, and aggregates its input
+ // watermark from the reports of its single upstream transform (the producer of its input
+ // PCollection, whose node name is the upstream transform id). Harness-reported metrics land in
+ // this stage's container of the job's metrics step map.
+ topology.addProcessor(
+ transformId,
+ () ->
+ new ExecutableStageProcessor(
+ stagePayload,
+ context.getJobInfo(),
+ transformId,
+ ImmutableSet.of(parentProcessor),
+ context.getMetricsContainerStepMap().getContainer(transformId),
+ outputChildByPCollectionId,
+ context.getPipelineOptions().getMaxBundleSize(),
+ context.getTerminationTracker()),
+ parentProcessor);
+
+ if (multiOutput) {
+ // One relay per output; downstream transforms wire to the relay, which re-stamps the stage's
+ // watermark with the relay's own id so their watermark aggregation stays consistent.
+ outputChildByPCollectionId.forEach(
+ (outputPCollectionId, relayName) -> {
+ topology.addProcessor(
+ relayName,
+ () -> new StageOutputProcessor(relayName, context.getTerminationTracker()),
+ transformId);
+ context.registerPCollectionProducer(outputPCollectionId, relayName);
+ context.registerPCollectionPartitionCount(outputPCollectionId, partitionCount);
+ });
+ } else if (!outputPCollectionIds.isEmpty()) {
+ context.registerPCollectionProducer(outputPCollectionIds.get(0), transformId);
+ context.registerPCollectionPartitionCount(outputPCollectionIds.get(0), partitionCount);
+ }
+ }
+}
diff --git a/runners/kafka-streams/src/main/java/org/apache/beam/runners/kafka/streams/translation/FlattenProcessor.java b/runners/kafka-streams/src/main/java/org/apache/beam/runners/kafka/streams/translation/FlattenProcessor.java
new file mode 100644
index 000000000000..56ae3a5d4550
--- /dev/null
+++ b/runners/kafka-streams/src/main/java/org/apache/beam/runners/kafka/streams/translation/FlattenProcessor.java
@@ -0,0 +1,124 @@
+/*
+ * 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.Set;
+import org.apache.beam.sdk.transforms.windowing.BoundedWindow;
+import org.apache.kafka.streams.processor.api.Processor;
+import org.apache.kafka.streams.processor.api.ProcessorContext;
+import org.apache.kafka.streams.processor.api.Record;
+import org.checkerframework.checker.nullness.qual.Nullable;
+import org.joda.time.Instant;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Kafka Streams {@link Processor} implementing Beam's {@code Flatten} primitive: the union of N
+ * input PCollections into one.
+ *
+ *
Data records pass straight through — merging the parents' streams is the flatten. The work is
+ * in the watermark, which Flatten owns as GroupByKey does: a {@link WatermarkAggregator} over its
+ * inputs, forwarding its own watermark only when the minimum across them advances and stamping it
+ * as a single source. That holds the output back until every branch has reported, so a downstream
+ * GroupByKey cannot fire before all branches are drained.
+ *
+ *
Branches are told apart by the transform id each producer stamps, since Kafka Streams does not
+ * say which parent forwarded a record. A producer stamps its own identity regardless of who
+ * consumes it, so a PCollection feeding several Flattens reports one identity and each Flatten
+ * still waits only for the upstream transforms handed to it at construction.
+ */
+class FlattenProcessor
+ implements Processor, byte[], KStreamsPayload> {
+
+ private static final Logger LOG = LoggerFactory.getLogger(FlattenProcessor.class);
+
+ // This transform's own id, stamped on every watermark it forwards downstream.
+ private final String transformId;
+ // Computes the output watermark as min() over the upstream transforms' reports, holding until
+ // every partition of every expected upstream transform has reported (see WatermarkAggregator).
+ private final WatermarkAggregator watermarkAggregator;
+ // The last watermark actually forwarded downstream, so we only forward when it advances.
+ private Instant lastForwardedWatermark = BoundedWindow.TIMESTAMP_MIN_VALUE;
+
+ // Reports this Flatten as finished once every branch it merges has gone terminal.
+ private final TerminationReporter terminationReporter;
+
+ private @Nullable ProcessorContext> context;
+
+ /**
+ * @param transformId this Flatten's own transform id, stamped on the watermarks it emits
+ * @param upstreamTransformIds the producers of this Flatten's input PCollections (known from the
+ * pipeline graph), whose reports the {@link WatermarkAggregator} waits for
+ */
+ FlattenProcessor(
+ String transformId, Set upstreamTransformIds, TerminationTracker terminationTracker) {
+ this.transformId = transformId;
+ this.watermarkAggregator = new WatermarkAggregator(upstreamTransformIds);
+ this.terminationReporter = new TerminationReporter(terminationTracker, transformId);
+ }
+
+ @Override
+ public void init(ProcessorContext> context) {
+ this.context = context;
+ terminationReporter.init(context);
+ }
+
+ @Override
+ public void close() {
+ terminationReporter.close();
+ }
+
+ @Override
+ public void process(Record> record) {
+ KStreamsPayload payload = record.value();
+ if (payload == null) {
+ // A topic feeding the runner can always be written to from outside (or carry a tombstone),
+ // so recover from the obvious error instead of crashing the task: warn and drop.
+ LOG.warn(
+ "Flatten {} dropping record with null payload (external write or tombstone)",
+ transformId);
+ return;
+ }
+ ProcessorContext> ctx = checkInitialized(context);
+ if (!payload.isWatermark()) {
+ // Data: the union of the parents' data streams is the flatten — forward unchanged.
+ ctx.forward(record);
+ return;
+ }
+ watermarkAggregator.observe(payload.asWatermark());
+ Instant advanced = watermarkAggregator.advance();
+ if (advanced.isAfter(lastForwardedWatermark)) {
+ lastForwardedWatermark = advanced;
+ // Labelled as the only source a consumer will see; a shuffle downstream relabels it.
+ ctx.forward(
+ new Record>(
+ record.key(),
+ KStreamsPayload.watermark(advanced.getMillis(), transformId, 0, 1),
+ record.timestamp()));
+ terminationReporter.watermarkEmitted(ctx, advanced.getMillis());
+ }
+ }
+
+ private static ProcessorContext> checkInitialized(
+ @Nullable ProcessorContext> context) {
+ if (context == null) {
+ throw new IllegalStateException("FlattenProcessor used before init()");
+ }
+ return context;
+ }
+}
diff --git a/runners/kafka-streams/src/main/java/org/apache/beam/runners/kafka/streams/translation/FlattenTranslator.java b/runners/kafka-streams/src/main/java/org/apache/beam/runners/kafka/streams/translation/FlattenTranslator.java
new file mode 100644
index 000000000000..998553d514d8
--- /dev/null
+++ b/runners/kafka-streams/src/main/java/org/apache/beam/runners/kafka/streams/translation/FlattenTranslator.java
@@ -0,0 +1,95 @@
+/*
+ * 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.ArrayList;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+import org.apache.beam.model.pipeline.v1.RunnerApi;
+import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.Iterables;
+import org.apache.kafka.streams.Topology;
+
+/**
+ * Translates Beam's {@code Flatten} primitive ({@code beam:transform:flatten:v1}): the union of N
+ * input PCollections into one output PCollection.
+ *
+ *
Wires a single {@link FlattenProcessor} node to the producer of every input PCollection (Kafka
+ * Streams lets a processor have many parents), so the parents' data streams merge into it, and
+ * registers it as the producer of the flattened output so downstream translators wire to it. The
+ * processor forwards data through and owns its output watermark via a {@link WatermarkAggregator},
+ * which is handed the producers of the input PCollections — the upstream transform ids whose
+ * watermark reports the Flatten must hear from. Producers stamp their own transform id on the
+ * reports they emit, without regard to who consumes them, so an input shared with another Flatten
+ * needs no special handling.
+ *
+ *
A user-written self-flatten never reaches this translator: the fuser folds the Flatten into
+ * the consuming SDK-harness stage, which performs the duplication itself. The duplicate-input check
+ * below is defensive — if a runner-executed Flatten ever did receive the same PCollection twice,
+ * Kafka Streams could not wire the same parent to a child twice and the duplicate copy would be
+ * silently dropped, so failing fast is safer.
+ */
+class FlattenTranslator implements PTransformTranslator {
+
+ @Override
+ public void translate(
+ String transformId, RunnerApi.Pipeline pipeline, KafkaStreamsTranslationContext context) {
+ RunnerApi.PTransform transform = pipeline.getComponents().getTransformsOrThrow(transformId);
+ // Flatten produces exactly one output PCollection, fed by all of its input PCollections.
+ String outputPCollectionId = Iterables.getOnlyElement(transform.getOutputsMap().values());
+
+ Topology topology = context.getTopology();
+ Set seenInputs = new HashSet<>();
+ List parentProcessors = new ArrayList<>();
+ Set upstreamTransformIds = new HashSet<>();
+ // How many instances this Flatten runs as. Kafka Streams merges the subtopologies of every
+ // parent a processor is wired to and gives the merged subtopology as many tasks as its largest
+ // source topic has partitions, so the max is what that comes to. In practice the inputs agree:
+ // a Flatten whose branches could disagree — one through a GroupByKey, one straight from a
+ // source — is fused into the harness stage instead of becoming a node here, and the runner
+ // Flattens that do reach this translator come from the fuser deduplicating partial outputs of
+ // one PCollection. The max is kept as the cheap conservative choice rather than asserting that
+ // agreement, which is not enforced anywhere.
+ int partitionCount = 1;
+ for (String inputPCollectionId : transform.getInputsMap().values()) {
+ if (!seenInputs.add(inputPCollectionId)) {
+ throw new UnsupportedOperationException(
+ "Flatten "
+ + transform.getUniqueName()
+ + " has PCollection "
+ + inputPCollectionId
+ + " as an input more than once; a self-flatten is not yet supported by the Kafka"
+ + " Streams runner.");
+ }
+ String parentProcessor = context.getProcessorNameForPCollection(inputPCollectionId);
+ parentProcessors.add(parentProcessor);
+ upstreamTransformIds.add(parentProcessor);
+ partitionCount = Math.max(partitionCount, context.getPartitionCount(inputPCollectionId));
+ }
+
+ topology.addProcessor(
+ transformId,
+ () ->
+ new FlattenProcessor(
+ transformId, upstreamTransformIds, context.getTerminationTracker()),
+ parentProcessors.toArray(new String[0]));
+
+ context.registerPCollectionProducer(outputPCollectionId, transformId);
+ context.registerPCollectionPartitionCount(outputPCollectionId, partitionCount);
+ }
+}
diff --git a/runners/kafka-streams/src/main/java/org/apache/beam/runners/kafka/streams/translation/GroupByKeyBroadcastPartitioner.java b/runners/kafka-streams/src/main/java/org/apache/beam/runners/kafka/streams/translation/GroupByKeyBroadcastPartitioner.java
new file mode 100644
index 000000000000..3c775c86f14c
--- /dev/null
+++ b/runners/kafka-streams/src/main/java/org/apache/beam/runners/kafka/streams/translation/GroupByKeyBroadcastPartitioner.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.Collections;
+import java.util.HashSet;
+import java.util.Optional;
+import java.util.Set;
+import org.apache.kafka.common.utils.Utils;
+import org.apache.kafka.streams.processor.StreamPartitioner;
+
+/**
+ * Partitions records on the GroupByKey repartition topic.
+ *
+ *
+ *
data records go to the single partition selected by hashing the (already encoded
+ * Beam key) Kafka record key — the same scheme Kafka's default partitioner uses — so every
+ * value of a key lands together;
+ *
watermark reports are broadcast to every partition, so each downstream
+ * GroupByKey task observes the terminal watermark and fires its keys.
+ *
+ *
+ * @param the data element type carried by data payloads
+ */
+class GroupByKeyBroadcastPartitioner implements StreamPartitioner> {
+
+ @Override
+ public Integer partition(String topic, byte[] key, KStreamsPayload value, int numPartitions) {
+ // Required by the interface but unused: Kafka Streams calls partitions() (overridden below)
+ // when it is present. Kept consistent with the data-hash path for safety.
+ return key == null ? 0 : Utils.toPositive(Utils.murmur2(key)) % numPartitions;
+ }
+
+ @Override
+ public Optional> partitions(
+ String topic, byte[] key, KStreamsPayload value, int numPartitions) {
+ if (value.isWatermark()) {
+ Set all = new HashSet<>();
+ for (int partition = 0; partition < numPartitions; partition++) {
+ all.add(partition);
+ }
+ return Optional.of(all);
+ }
+ if (key == null) {
+ // A keyless record has no partition it must go to, so leave the choice to Kafka rather than
+ // hashing a null or pinning one partition: an empty Optional tells Kafka Streams no explicit
+ // partition was chosen, and the producer's default partitioner spreads keyless records over
+ // the topic instead of piling them onto one. This is the method Kafka Streams calls, so the
+ // null has to be handled here and not only in partition() above.
+ return Optional.empty();
+ }
+ int partition = Utils.toPositive(Utils.murmur2(key)) % numPartitions;
+ return Optional.of(Collections.singleton(partition));
+ }
+}
diff --git a/runners/kafka-streams/src/main/java/org/apache/beam/runners/kafka/streams/translation/GroupByKeyTranslator.java b/runners/kafka-streams/src/main/java/org/apache/beam/runners/kafka/streams/translation/GroupByKeyTranslator.java
new file mode 100644
index 000000000000..8438e6af2a97
--- /dev/null
+++ b/runners/kafka-streams/src/main/java/org/apache/beam/runners/kafka/streams/translation/GroupByKeyTranslator.java
@@ -0,0 +1,215 @@
+/*
+ * 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 static org.apache.beam.runners.fnexecution.translation.PipelineTranslatorUtils.instantiateCoder;
+
+import org.apache.beam.model.pipeline.v1.RunnerApi;
+import org.apache.beam.sdk.coders.Coder;
+import org.apache.beam.sdk.coders.KvCoder;
+import org.apache.beam.sdk.transforms.windowing.BoundedWindow;
+import org.apache.beam.sdk.util.construction.RehydratedComponents;
+import org.apache.beam.sdk.util.construction.WindowingStrategyTranslation;
+import org.apache.beam.sdk.values.KV;
+import org.apache.beam.sdk.values.WindowedValues;
+import org.apache.beam.sdk.values.WindowingStrategy;
+import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.ImmutableSet;
+import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.Iterables;
+import org.apache.kafka.common.serialization.Serdes;
+import org.apache.kafka.streams.Topology;
+import org.apache.kafka.streams.state.Stores;
+import org.checkerframework.checker.nullness.qual.Nullable;
+
+/**
+ * Translates the {@code beam:transform:group_by_key:v1} URN — the runner's first stateful,
+ * shuffle-bearing transform.
+ *
+ *
Windowing and triggering run through Beam's {@link
+ * org.apache.beam.runners.core.ReduceFnRunner} inside {@link WindowedGroupByKeyProcessor}, as the
+ * Flink and Spark portable runners do, so fixed and sliding windows, the default trigger, allowed
+ * lateness and timestamp combiners all work. The input's windowing strategy is hydrated from the
+ * pipeline proto and handed to the processor.
+ *
+ *
The Beam key becomes the Kafka record key so Kafka Streams shuffles by it. The topology is a
+ * {@link ShuffleByKeyProcessor} that sets that key and passes watermark reports through, a sink to
+ * an internal repartition topic using a {@link GroupByKeyBroadcastPartitioner} that hashes data by
+ * key and fans watermarks out to every partition, a source reading that topic back, and the {@link
+ * WindowedGroupByKeyProcessor} with its state and timer stores.
+ */
+class GroupByKeyTranslator implements PTransformTranslator {
+
+ static final String SHUFFLE_SUFFIX = "-shuffle-by-key";
+ static final String SINK_SUFFIX = "-repartition-sink";
+ static final String SOURCE_SUFFIX = "-repartition-source";
+ static final String STATE_STORE_SUFFIX = "-state";
+ static final String HOLDS_INDEX_STORE_SUFFIX = "-holds-index";
+ static final String TIMER_STORE_SUFFIX = "-timers";
+ static final String TIMER_INDEX_STORE_SUFFIX = "-timers-index";
+ static final String REPARTITION_TOPIC_PREFIX = "__beam_gbk_";
+
+ @Override
+ public void translate(
+ String transformId, RunnerApi.Pipeline pipeline, KafkaStreamsTranslationContext context) {
+ RunnerApi.PTransform transform = pipeline.getComponents().getTransformsOrThrow(transformId);
+ String inputPCollectionId = Iterables.getOnlyElement(transform.getInputsMap().values());
+ String outputPCollectionId = Iterables.getOnlyElement(transform.getOutputsMap().values());
+
+ @SuppressWarnings({"unchecked", "rawtypes"})
+ WindowedValues.WindowedValueCoder> inputCoder =
+ (WindowedValues.WindowedValueCoder)
+ instantiateCoder(inputPCollectionId, pipeline.getComponents());
+ KvCoder kvCoder = (KvCoder) inputCoder.getValueCoder();
+ Coder