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..836f121ca331 --- /dev/null +++ b/runners/kafka-streams/measurement/docker-compose.yml @@ -0,0 +1,27 @@ +# 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..3ba9b22ca0ee --- /dev/null +++ b/runners/kafka-streams/measurement/src/main/java/org/apache/beam/runners/kafka/streams/measurement/RescalingMeasurement.java @@ -0,0 +1,261 @@ +/* + * 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 are added and removed. + * + *

Run several of these against one Kafka. They share an application id, so Kafka's consumer + * group divides the work between them, and stopping one hands its share to the others. + * + *

This is an application rather than a test on purpose. The numbers only mean something if the + * pipeline is doing a realistic amount of work — a grouping over thousands of keys, fed fast enough + * that every partition has something to do. A pipeline that trickles produces idle partitions, and + * an idle partition holds a watermark back for reasons that have nothing to do with rescaling. + * + *

The source produces a fixed number of elements per second over a fixed set of keys, so what a + * complete window looks like is known before the run starts: every window should report the same + * number of groups. That is what makes a shortfall legible as a shortfall, rather than as one of + * the many rates a pipeline could happen to be running at. + * + * 

+ *   docker compose -f runners/kafka-streams/measurement/docker-compose.yml up -d
+ *   ./gradlew :runners:kafka-streams:measurement:installDist
+ *
+ * + *

Then start two instances, sharing an application id and differing in everything local to the + * instance. Each needs its own {@code --stateDir}: two instances sharing one directory fail with a + * {@code LockException}, because Kafka Streams locks the state it keeps on disk. + * + * 

+ *   BIN=runners/kafka-streams/measurement/build/install/measurement/bin/measurement
+ *   $BIN --applicationId=demo --instanceName=one --stateDir=/tmp/ks-one &
+ *   $BIN --applicationId=demo --instanceName=two --stateDir=/tmp/ks-two &
+ *
+ * + *

The pipeline logs one line per key per window. Nothing is counted beside the pipeline: the + * groups in a window are its own output, so the tally does not depend on how many instances are + * running or on which of them happens to be doing the work. + * + * 

+ *   <millis> <instance> window_end=<millis> key=<key> count=<n> skew_ms=<n>
+ *
+ * + *

Because the rate and the key space are both fixed, a complete window has one line per key and + * the same count on each, so counting the lines for a window says whether the window was complete. + * + *

{@code skew_ms} is the gap between the window's event time and the wall clock when the group + * came out. It is what falling behind should look like: a pipeline that cannot keep up ought to + * report its groups later and later while still reporting all of them, so a climbing skew with + * complete windows is congestion, and missing groups are something else. + * + *

To watch a handover, kill one instance and watch the other's lines. The delay before the + * survivor reports the killed instance's share again is dominated by {@code --sessionTimeoutMs}, + * which is how long the consumer group waits before deciding the instance is gone. + */ +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/settings.gradle.kts b/settings.gradle.kts index 050d97dc600e..de1e5ea6533d 100644 --- a/settings.gradle.kts +++ b/settings.gradle.kts @@ -153,6 +153,7 @@ 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:kafka-streams:measurement") } include(":runners:local-java") include(":runners:portability:java")