This is an automated email from the ASF dual-hosted git repository.
je-ik pushed a commit to branch feat/18479-kafka-streams-runner-skeleton
in repository https://gitbox.apache.org/repos/asf/beam.git
The following commit(s) were added to
refs/heads/feat/18479-kafka-streams-runner-skeleton by this push:
new 49d459a243e [GSoC 2026] Kafka Streams runner: put the runner behind an
opt-in build flag (#39762)
49d459a243e is described below
commit 49d459a243eb613e97e6dae5e935638d8e4f4677
Author: M Junaid Shaukat <[email protected]>
AuthorDate: Sun Aug 16 22:43:37 2026 +0500
[GSoC 2026] Kafka Streams runner: put the runner behind an opt-in build
flag (#39762)
---
.../beam_KafkaStreamsRunner_FeatureBranch.yml | 4 +--
.../beam_PreCommit_Java_Kafka_Streams_Runner.yml | 1 +
build.gradle.kts | 5 ++-
sdks/python/test-suites/portable/common.gradle | 10 ++++--
settings.gradle.kts | 14 ++++++--
.../en/documentation/runners/kafkastreams.md | 40 +++++++++++++++++++---
6 files changed, 61 insertions(+), 13 deletions(-)
diff --git a/.github/workflows/beam_KafkaStreamsRunner_FeatureBranch.yml
b/.github/workflows/beam_KafkaStreamsRunner_FeatureBranch.yml
index e290c9beee1..8b558865c21 100644
--- a/.github/workflows/beam_KafkaStreamsRunner_FeatureBranch.yml
+++ b/.github/workflows/beam_KafkaStreamsRunner_FeatureBranch.yml
@@ -69,6 +69,6 @@ jobs:
restore-keys: |
${{ runner.os }}-gradle-kafka-streams-
- name: Build and test Kafka Streams runner
- run: ./gradlew :runners:kafka-streams:build --no-daemon --stacktrace
+ run: ./gradlew -Pwith-kafka-streams-runner
:runners:kafka-streams:build --no-daemon --stacktrace
- name: Run ValidatesRunner suite
- run: ./gradlew :runners:kafka-streams:validatesRunner --no-daemon
--stacktrace
+ run: ./gradlew -Pwith-kafka-streams-runner
:runners:kafka-streams:validatesRunner --no-daemon --stacktrace
diff --git a/.github/workflows/beam_PreCommit_Java_Kafka_Streams_Runner.yml
b/.github/workflows/beam_PreCommit_Java_Kafka_Streams_Runner.yml
index 564b2bbc4bc..005764b4d33 100644
--- a/.github/workflows/beam_PreCommit_Java_Kafka_Streams_Runner.yml
+++ b/.github/workflows/beam_PreCommit_Java_Kafka_Streams_Runner.yml
@@ -89,6 +89,7 @@ jobs:
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
diff --git a/build.gradle.kts b/build.gradle.kts
index b3b9fdd7fdf..3f5d39d9934 100644
--- a/build.gradle.kts
+++ b/build.gradle.kts
@@ -278,7 +278,10 @@ tasks.register("javaPreCommit") {
dependsOn(":runners:java-fn-execution:build")
dependsOn(":runners:java-job-service:build")
dependsOn(":runners:jet:build")
- dependsOn(":runners:kafka-streams: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/sdks/python/test-suites/portable/common.gradle
b/sdks/python/test-suites/portable/common.gradle
index c8c789ca96d..6cc30fcadab 100644
--- a/sdks/python/test-suites/portable/common.gradle
+++ b/sdks/python/test-suites/portable/common.gradle
@@ -203,10 +203,14 @@ def createKafkaStreamsRunnerTestTask() {
return task
}
-createKafkaStreamsRunnerTestTask()
+// The Kafka Streams runner is opt-in (-Pwith-kafka-streams-runner), so its
job server is only a
+// project when it was asked for; without it there is nothing for these tasks
to run against.
+if (project.findProject(':runners:kafka-streams:job-server') != null) {
+ createKafkaStreamsRunnerTestTask()
-tasks.register("kafkaStreamsValidatesRunner") {
- dependsOn 'kafkaStreamsCompatibilityMatrixLOOPBACK'
+ tasks.register("kafkaStreamsValidatesRunner") {
+ dependsOn 'kafkaStreamsCompatibilityMatrixLOOPBACK'
+ }
}
def createPrismRunnerTestTask(String workerType) {
diff --git a/settings.gradle.kts b/settings.gradle.kts
index cd1134d685c..050d97dc600 100644
--- a/settings.gradle.kts
+++ b/settings.gradle.kts
@@ -143,9 +143,17 @@
include(":runners:google-cloud-dataflow-java:examples-streaming")
include(":runners:java-fn-execution")
include(":runners:java-job-service")
include(":runners:jet")
-include(":runners:kafka-streams")
-include(":runners:kafka-streams:proto")
-include(":runners:kafka-streams:job-server")
+// The Kafka Streams runner is opt-in, and is left out of the build unless it
is asked for with
+// -Pwith-kafka-streams-runner. It is being developed in the open so that
others can build it and
+// work on it, but it is not ready to be released: bundles are not yet closed
after a bounded time
+// (https://github.com/apache/beam/issues/39633), among other things. Keeping
it out of the default
+// build means it reaches nobody who did not ask for it, and the flag can be
dropped when the runner
+// is stable enough - or the runner can be dropped, without either affecting
users.
+if (startParameter.projectProperties.containsKey("with-kafka-streams-runner"))
{
+ include(":runners:kafka-streams")
+ include(":runners:kafka-streams:proto")
+ include(":runners:kafka-streams:job-server")
+}
include(":runners:local-java")
include(":runners:portability:java")
include(":runners:prism")
diff --git a/website/www/site/content/en/documentation/runners/kafkastreams.md
b/website/www/site/content/en/documentation/runners/kafkastreams.md
index 058e85b6d46..1fe5d4e86ca 100644
--- a/website/www/site/content/en/documentation/runners/kafkastreams.md
+++ b/website/www/site/content/en/documentation/runners/kafkastreams.md
@@ -31,7 +31,7 @@ groups, changelog topics, and transactions.
That makes it worth considering if you already run Kafka and want Beam's
programming model without
introducing a second distributed system to operate.
-## The runner is experimental
+## The runner is experimental, and is not built by default
**The Kafka Streams Runner is experimental.** It executes a meaningful subset
of the Beam model
correctly, and the parts it does support are covered by Beam's own
`@ValidatesRunner` suite, but
@@ -41,6 +41,36 @@ supported](#what-is-not-supported-yet) before choosing it
for anything real.
It is also aimed squarely at streaming. A pipeline over bounded data will run,
but there are more
efficient choices for batch work; this runner exists for pipelines that do not
end.
+It is **not part of a Beam release, and not part of the default build**. It is
developed in the open
+so that people can build it, use it and work on it, but it is not ready to be
released: there are
+known bugs, not only missing features — bundles are not yet closed after a
bounded time
+([#39633](https://github.com/apache/beam/issues/39633)), for one. Building it
takes an opt-in flag:
+
+```
+./gradlew -Pwith-kafka-streams-runner :runners:kafka-streams:build
+```
+
+Without `-Pwith-kafka-streams-runner` the runner's projects are left out of
the build entirely, so
+it reaches nobody who has not asked for it. The intent is to give the runner
somewhere to be
+developed and maintained by whoever is interested in it. If it becomes stable
enough the flag will
+be dropped and the runner built like any other; if it does not, it can be
removed again without
+affecting anyone, since no release ever contained it.
+
+## Building it
+
+Every command in this page needs the opt-in flag. To build the runner and run
its tests:
+
+```
+./gradlew -Pwith-kafka-streams-runner :runners:kafka-streams:build
+./gradlew -Pwith-kafka-streams-runner :runners:kafka-streams:validatesRunner
+```
+
+To build the job server jar that the Python SDK submits to:
+
+```
+./gradlew -Pwith-kafka-streams-runner
:runners:kafka-streams:job-server:shadowJar
+```
+
## Running a pipeline
The runner is portable: it executes user code over the Fn API, in an SDK
harness, so a pipeline goes
@@ -74,14 +104,14 @@ python my_pipeline.py \
The SDK harness runs in `LOOPBACK` mode by default, so a local run needs no
Docker. Building the jar
takes a while the first time; `--kafka_streams_job_server_jar` points at a
prebuilt one, which
-`./gradlew :runners:kafka-streams:job-server:shadowJar` produces.
+`./gradlew -Pwith-kafka-streams-runner
:runners:kafka-streams:job-server:shadowJar` produces.
### Against a job server you are already running
Start one, which listens on `localhost:8099` by default:
```
-./gradlew :runners:kafka-streams:runJobServer
+./gradlew -Pwith-kafka-streams-runner :runners:kafka-streams:runJobServer
```
Then point a pipeline at it instead of letting the runner start its own. From
Java:
@@ -117,10 +147,12 @@ Named as Java spells them below; from Python the same
options are in snake case,
| `applicationId` | *(required)* | Kafka Streams `application.id`. Must be
unique per pipeline. |
| `internalParallelism` | `1` | Partitions for the internal topics the runner
creates, which is the parallelism the shuffled parts of a pipeline can reach. |
| `topicReplicationFactor` | `1` | Replication factor for those topics. |
-| `maxBundleSize` | `1000` | Elements per bundle, and elements taken per poll
of an unbounded source. |
+| `maxBundleSize` | `1000` | Elements per bundle. |
| `maxBundleTimeMs` | `1000` | Intended cap on how long a bundle may stay
open. **Not applied yet** — see below. |
+| `readMaxElementsPerPoll` | `1000` | Elements an unbounded source may take
per poll. Separate from `maxBundleSize`, so a pipeline can have small bundles
without throttling its source. |
| `readMaxPollTimeMs` | `10` | How long one turn of reading an unbounded
source may take before it yields the Kafka Streams thread. A source is polled
every 50ms and shares its thread with the rest of the topology, so a turn that
overruns that interval leaves the stages below it unscheduled; a bound on
elements alone cannot bound the time. |
| `readCheckpointNumBundles` | `10` | Polls of an unbounded source between
stores of its checkpoint mark. Larger values replay more after a restart. |
+| `sessionTimeoutMs` | `45000` | How long the consumer group waits before
deciding an instance has gone, which is the floor on how quickly its work moves
elsewhere. A broker refuses a value below its own
`group.min.session.timeout.ms`. |
| `stateDir` | temp directory | Where Kafka Streams keeps local state. |
### Topics the runner creates
