[WIP] Merge Kafka Streams Runner skeleton

.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
+}

.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`| [![.github/workflows/beam_PreCommit_Java_HBase_IO_Direct.yml](https://github.com/apache/beam/actions/workflows/beam_PreCommit_Java_HBase_IO_Direct.yml/badge.svg?event=schedule)](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`| [![.github/workflows/beam_PreCommit_Java_HCatalog_IO_Direct.yml](https://github.com/apache/beam/actions/workflows/beam_PreCommit_Java_HCatalog_IO_Direct.yml/badge.svg?event=schedule)](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`| [![.github/workflows/beam_PreCommit_Java_Kafka_IO_Direct.yml](https://github.com/apache/beam/actions/workflows/beam_PreCommit_Java_Kafka_IO_Direct.yml/badge.svg?event=schedule)](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`| [![.github/workflows/beam_PreCommit_Java_Kafka_Streams_Runner.yml](https://github.com/apache/beam/actions/workflows/beam_PreCommit_Java_Kafka_Streams_Runner.yml/badge.svg?event=schedule)](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`| [![.github/workflows/beam_PreCommit_Java_InfluxDb_IO_Direct.yml](https://github.com/apache/beam/actions/workflows/beam_PreCommit_Java_InfluxDb_IO_Direct.yml/badge.svg?event=schedule)](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`| [![.github/workflows/beam_PreCommit_Java_JDBC_IO_Direct.yml](https://github.com/apache/beam/actions/workflows/beam_PreCommit_Java_JDBC_IO_Direct.yml/badge.svg?event=schedule)](https://github.com/apache/beam/actions/workflows/beam_PreCommit_Java_JDBC_IO_Direct.yml?query=event%3Aschedule) |

.github/workflows/beam_PreCommit_Java.yml

@@ -192,6 +192,7 @@ jobs:
             -PdisableSpotlessCheck=true \
             -PdisableCheckStyle=true \
             -PenableJacocoReport \
+            -Pwith-kafka-streams-runner \
       - name: Archive JUnit Test Results
         uses: actions/upload-artifact@v7
         if: ${{ !success() }}

.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'

CHANGES.md

@@ -211,6 +211,7 @@
 
 ## New Features / Improvements
 
+* (Java) Added an experimental Kafka Streams runner, which executes a Beam pipeline as an ordinary Kafka Streams application with no cluster to operate. It supports a subset of the model and is not built by default; pass `-Pwith-kafka-streams-runner` to include it ([#18479](https://github.com/apache/beam/issues/18479)).
 * 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

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")

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
+  }
+}