Hi all,
I would like to propose merging the Kafka Streams runner from its
feature branch into master. It has been developed this summer as
a Google Summer of Code project under the tracking issue [1]
<https://github.com/apache/beam/issues/18479>, mentored by Jan
Lukavský. The pull request is at [2]
<https://github.com/apache/beam/pull/39785>.
What it is. A portable runner that translates a Beam pipeline
into a Kafka Streams topology and executes user code over the Fn
API. What makes it different from the other runners is that Kafka
Streams is a library, not a cluster: there is no job manager and
no resource manager to operate. A pipeline is an ordinary JVM
process that reads from and writes to Kafka, and you scale it by
starting more copies of that process. State, fault tolerance and
exactly-once come from Kafka itself, through consumer groups,
changelog topics and transactions.
Current state. It is a skeleton. It runs a real subset of the
model: bounded and unbounded reads, stateless ParDo including
multiple outputs, GroupByKey and Combine, global, fixed and
sliding windows with the default trigger and allowed lateness,
Flatten, Redistribute, metrics, and exactly-once via Kafka
transactions. That subset is covered by 105 unit tests, 59 of
Beam's own @ValidatesRunner tests, and integration tests against
a real broker. Beam's Python portable suite also runs against it,
so it is exercised from a non-Java SDK as well.
It also has real gaps, each tracked: side inputs (#39628
<https://github.com/apache/beam/issues/39628>), stateful ParDo
and user timers (#39629
<https://github.com/apache/beam/issues/39629>), merging windows
and custom WindowFns (#39630
<https://github.com/apache/beam/issues/39630>), splittable DoFn
(#39631 <https://github.com/apache/beam/issues/39631>),
TestStream (#39632
<https://github.com/apache/beam/issues/39632>), and reading a
source in parallel (#39626
<https://github.com/apache/beam/issues/39626>).
And it has at least one known bug rather than a missing feature:
bundles are not closed after a bounded time (#39633
<https://github.com/apache/beam/issues/39633>). The
maxBundleTimeMs option is accepted and has no effect, because
closing a bundle from a wall-clock punctuator duplicated output
against a real broker and the cause is not yet understood. There
may be others we have not found.
Why I propose merging now. The motivation was to build a skeleton
that can be developed further by several contributors, and master
is where that can happen: the runner can be built, run and worked
on by anyone interested, and the gaps above are well-defined
pieces of work someone could pick up.
To make that safe, the runner is not part of the standard build.
Its subprojects are only included when
-Pwith-kafka-streams-runner is passed:
./gradlew -Pwith-kafka-streams-runner
:runners:kafka-streams:build
Without the flag they are not in the build at all, so nothing
reaches users who have not asked for it, and no release artifact
contains it. It is built in the Java precommit, so it cannot rot
unnoticed. If the runner becomes stable enough the flag comes off
and it is built like any other runner; if it does not, it can be
dropped again without affecting anyone, because no release ever
shipped it.
What is the potential of the runner. Two things, one operational
and one
about recovery.
The operational one is that there is nothing to operate. If you
already run Kafka, a Beam pipeline becomes an ordinary
application you deploy like any other — no job manager, no
resource manager, no second distributed system to size, upgrade
and keep alive. Scaling up or down is starting or stopping a
process, and the consumer group redistributes the work.
The recovery one is that Kafka Streams reassigns partitions and
restores state from a changelog, where a checkpoint-based engine
restarts a job from its last checkpoint. On a Mac, with one
broker and two instances, killing the instance holding the source
read with kill -9 and timing until the other took the work over:
session.timeout.ms <http://session.timeout.ms> = 6000 -> 8.6,
8.7, 9.1, 10.0 s
session.timeout.ms <http://session.timeout.ms> = 45000 ->
55.7, 55.8 s (Kafka's default)
Handover is dominated by how long the consumer group takes to
notice, which is session.timeout.ms <http://session.timeout.ms>
and is configurable; the recovery work itself is the remainder.
These are laptop numbers meant to show the shape of the thing,
not a benchmark against other runners.
This lazy consensus request will be open for at least 72 hours.
If there are no objections by then the consensus will pass. Any
comments or objections are welcome, here or on the pull request.
Thanks,
Junaid Shaukat
https://github.com/junaiddshaukat
[1] https://github.com/apache/beam/issues/18479
[2] https://github.com/apache/beam/pull/39785
