From a805e95194461b08c9eb8c2107fa046458c94d1b Mon Sep 17 00:00:00 2001 From: junaiddshaukat Date: Sun, 21 Jun 2026 13:28:52 +0500 Subject: [PATCH 1/3] Add KStreamsPayload Serde for crossing topic boundaries MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit KStreamsPayload has so far only flowed in-JVM via ProcessorContext#forward, so it needed no serialization. GroupByKey introduces the first real topic — the key-based repartition topic — so the payload now has to be serialized. KStreamsPayloadSerde is parameterized by the Coder> for the data variant, since different topics carry different element types; the watermark report variant is coder-independent. The wire format is a one-byte discriminator followed by the variant body: data is the windowed-value-coder encoding; watermark is the millis + sourcePartition + totalSourcePartitions report. The serde assumes non-null payloads, since the topics it is used on (repartition and watermark fan-out) are not log-compacted. Refs #18479 --- .../translation/KStreamsPayloadSerde.java | 121 ++++++++++++++++++ .../translation/KStreamsPayloadSerdeTest.java | 85 ++++++++++++ 2 files changed, 206 insertions(+) create mode 100644 runners/kafka-streams/src/main/java/org/apache/beam/runners/kafka/streams/translation/KStreamsPayloadSerde.java create mode 100644 runners/kafka-streams/src/test/java/org/apache/beam/runners/kafka/streams/translation/KStreamsPayloadSerdeTest.java diff --git a/runners/kafka-streams/src/main/java/org/apache/beam/runners/kafka/streams/translation/KStreamsPayloadSerde.java b/runners/kafka-streams/src/main/java/org/apache/beam/runners/kafka/streams/translation/KStreamsPayloadSerde.java new file mode 100644 index 000000000000..dd30ed3696eb --- /dev/null +++ b/runners/kafka-streams/src/main/java/org/apache/beam/runners/kafka/streams/translation/KStreamsPayloadSerde.java @@ -0,0 +1,121 @@ +/* + * 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.ByteArrayInputStream; +import java.io.ByteArrayOutputStream; +import java.io.DataInputStream; +import java.io.DataOutputStream; +import java.io.IOException; +import org.apache.beam.sdk.coders.Coder; +import org.apache.beam.sdk.values.WindowedValue; +import org.apache.kafka.common.errors.SerializationException; +import org.apache.kafka.common.serialization.Deserializer; +import org.apache.kafka.common.serialization.Serde; +import org.apache.kafka.common.serialization.Serializer; + +/** + * Kafka {@link Serde} for {@link KStreamsPayload}, enabling the envelope to cross topic boundaries + * (e.g. the repartition topic a {@code GroupByKey} introduces). Until now {@link KStreamsPayload} + * only flowed in-JVM via {@code ProcessorContext#forward}, so no serialization was needed. + * + *

The wire format is a one-byte discriminator followed by the variant body: + * + *

    + *
  • data: {@code [0x00][windowedValueCoder-encoded WindowedValue]} — the data element is + * encoded with the {@link Coder} supplied for the topic's PCollection. + *
  • watermark: {@code [0x01][long watermarkMillis][int sourcePartition][int + * totalSourcePartitions]} — the in-band watermark report, coder-independent. + *
+ * + *

A {@link KStreamsPayloadSerde} is parameterized by the {@link Coder} for the data variant + * because different topics carry different element types; the watermark variant needs no coder. + * + *

The serde assumes non-null payloads: the topics it is used on (repartition and watermark + * fan-out) are not log-compacted, so no tombstone (null-valued) records occur. + * + * @param the data element type carried by data payloads on this topic + */ +public final class KStreamsPayloadSerde implements Serde> { + + private static final byte DATA_TAG = 0; + private static final byte WATERMARK_TAG = 1; + + private final Coder> dataCoder; + + public KStreamsPayloadSerde(Coder> dataCoder) { + this.dataCoder = dataCoder; + } + + @Override + public Serializer> serializer() { + return new PayloadSerializer(); + } + + @Override + public Deserializer> deserializer() { + return new PayloadDeserializer(); + } + + private final class PayloadSerializer implements Serializer> { + @Override + public byte[] serialize(String topic, KStreamsPayload payload) { + ByteArrayOutputStream out = new ByteArrayOutputStream(); + try { + if (payload.isData()) { + out.write(DATA_TAG); + dataCoder.encode(payload.getData(), out); + } else { + WatermarkPayload watermark = payload.asWatermark(); + DataOutputStream dataOut = new DataOutputStream(out); + dataOut.writeByte(WATERMARK_TAG); + dataOut.writeLong(watermark.getWatermarkMillis()); + dataOut.writeInt(watermark.getSourcePartition()); + dataOut.writeInt(watermark.getTotalSourcePartitions()); + dataOut.flush(); + } + } catch (IOException e) { + throw new SerializationException("Failed to serialize KStreamsPayload", e); + } + return out.toByteArray(); + } + } + + private final class PayloadDeserializer implements Deserializer> { + @Override + public KStreamsPayload deserialize(String topic, byte[] bytes) { + ByteArrayInputStream in = new ByteArrayInputStream(bytes); + try { + int tag = in.read(); + if (tag == DATA_TAG) { + return KStreamsPayload.data(dataCoder.decode(in)); + } else if (tag == WATERMARK_TAG) { + DataInputStream dataIn = new DataInputStream(in); + long watermarkMillis = dataIn.readLong(); + int sourcePartition = dataIn.readInt(); + int totalSourcePartitions = dataIn.readInt(); + return KStreamsPayload.watermark(watermarkMillis, sourcePartition, totalSourcePartitions); + } else { + throw new SerializationException("Unknown KStreamsPayload tag: " + tag); + } + } catch (IOException e) { + throw new SerializationException("Failed to deserialize KStreamsPayload", e); + } + } + } +} diff --git a/runners/kafka-streams/src/test/java/org/apache/beam/runners/kafka/streams/translation/KStreamsPayloadSerdeTest.java b/runners/kafka-streams/src/test/java/org/apache/beam/runners/kafka/streams/translation/KStreamsPayloadSerdeTest.java new file mode 100644 index 000000000000..1e4c4c0263a4 --- /dev/null +++ b/runners/kafka-streams/src/test/java/org/apache/beam/runners/kafka/streams/translation/KStreamsPayloadSerdeTest.java @@ -0,0 +1,85 @@ +/* + * 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.hamcrest.CoreMatchers.is; +import static org.hamcrest.MatcherAssert.assertThat; +import static org.junit.Assert.assertThrows; + +import org.apache.beam.sdk.coders.Coder; +import org.apache.beam.sdk.coders.VarIntCoder; +import org.apache.beam.sdk.transforms.windowing.BoundedWindow; +import org.apache.beam.sdk.transforms.windowing.GlobalWindow; +import org.apache.beam.sdk.values.WindowedValue; +import org.apache.beam.sdk.values.WindowedValues; +import org.apache.kafka.common.errors.SerializationException; +import org.apache.kafka.common.serialization.Deserializer; +import org.apache.kafka.common.serialization.Serializer; +import org.junit.Test; + +/** Tests for {@link KStreamsPayloadSerde}. */ +public class KStreamsPayloadSerdeTest { + + private static final String TOPIC = "ks-payload-serde-test"; + + private final Coder> dataCoder = + WindowedValues.getFullCoder(VarIntCoder.of(), GlobalWindow.Coder.INSTANCE); + private final KStreamsPayloadSerde serde = new KStreamsPayloadSerde<>(dataCoder); + + private KStreamsPayload roundTrip(KStreamsPayload payload) { + Serializer> serializer = serde.serializer(); + Deserializer> deserializer = serde.deserializer(); + return deserializer.deserialize(TOPIC, serializer.serialize(TOPIC, payload)); + } + + @Test + public void roundTripsDataPayload() { + KStreamsPayload payload = KStreamsPayload.data(WindowedValues.valueInGlobalWindow(42)); + KStreamsPayload out = roundTrip(payload); + assertThat(out.isData(), is(true)); + assertThat(out.getData().getValue(), is(42)); + assertThat(out, is(payload)); + } + + @Test + public void roundTripsWatermarkPayload() { + KStreamsPayload payload = KStreamsPayload.watermark(12345L, 2, 4); + KStreamsPayload out = roundTrip(payload); + assertThat(out.isWatermark(), is(true)); + assertThat(out.asWatermark().getWatermarkMillis(), is(12345L)); + assertThat(out.asWatermark().getSourcePartition(), is(2)); + assertThat(out.asWatermark().getTotalSourcePartitions(), is(4)); + assertThat(out, is(payload)); + } + + @Test + public void roundTripsTerminalMaxWatermark() { + KStreamsPayload payload = + KStreamsPayload.watermark(BoundedWindow.TIMESTAMP_MAX_VALUE.getMillis(), 0, 1); + assertThat( + roundTrip(payload).asWatermark().getWatermarkMillis(), + is(BoundedWindow.TIMESTAMP_MAX_VALUE.getMillis())); + } + + @Test + public void unknownTagThrows() { + byte[] bogus = new byte[] {(byte) 0x7f}; + assertThrows( + SerializationException.class, () -> serde.deserializer().deserialize(TOPIC, bogus)); + } +} From ebd16294eb21d1ee242e4ee45edc7aa72ec39aca Mon Sep 17 00:00:00 2001 From: junaiddshaukat Date: Thu, 25 Jun 2026 20:53:53 +0500 Subject: [PATCH 2/3] Use protobuf for KStreamsPayload serialization Per @je-ik review: replace the hand-rolled byte format with a protobuf KafkaStreamsPayload message, for compatible schema evolution and compact varint encoding. Proto codegen is wired via applyGrpcNature (additive over applyJavaNature, so spotbugs and nullness stay on); the generated message is excluded from the nullness checker via generatedClassPatterns, and the unvendored com.google.protobuf import is allowed for the serde via the checkstyle suppressions file. Two notes on the schema vs the sketch: the redundant Type enum is dropped (the oneof case already discriminates data vs watermark, and it collided with the oneof on field number 1), and the watermark millis field is sint64 rather than uint64 because Beam event times can be negative (TIMESTAMP_MIN_VALUE). Refs #18479 --- runners/kafka-streams/build.gradle | 9 ++ .../translation/KStreamsPayloadSerde.java | 96 +++++++++---------- .../translation/kafka_streams_payload.proto | 52 ++++++++++ .../translation/KStreamsPayloadSerdeTest.java | 13 ++- .../beam/checkstyle/suppressions.xml | 1 + 5 files changed, 122 insertions(+), 49 deletions(-) create mode 100644 runners/kafka-streams/src/main/proto/org/apache/beam/runners/kafka/streams/translation/kafka_streams_payload.proto diff --git a/runners/kafka-streams/build.gradle b/runners/kafka-streams/build.gradle index 52b320dd70a8..111c06b73722 100644 --- a/runners/kafka-streams/build.gradle +++ b/runners/kafka-streams/build.gradle @@ -22,7 +22,15 @@ def kafka_version = '3.9.0' applyJavaNature( automaticModuleName: 'org.apache.beam.runners.kafka.streams', + // protoc-generated message classes are not nullness-clean; exclude them from the checker. + generatedClassPatterns: [ + /^org\.apache\.beam\.runners\.kafka\.streams\.translation\.KafkaStreamsPayloadProtos.*/ + ], ) +// Adds protoc to compile the runner-internal control-message protos (e.g. KafkaStreamsPayload). +// applyGrpcNature is purely additive on top of applyJavaNature — it only wires the protobuf +// plugin; the gRPC codegen plugin is a no-op since these protos define no services. +applyGrpcNature() description = "Apache Beam :: Runners :: Kafka Streams" @@ -53,6 +61,7 @@ dependencies { implementation project(path: ":sdks:java:extensions:google-cloud-platform-core") implementation library.java.args4j implementation library.java.joda_time + implementation library.java.protobuf_java implementation library.java.slf4j_api implementation library.java.vendored_grpc_1_69_0 implementation library.java.vendored_guava_32_1_2_jre diff --git a/runners/kafka-streams/src/main/java/org/apache/beam/runners/kafka/streams/translation/KStreamsPayloadSerde.java b/runners/kafka-streams/src/main/java/org/apache/beam/runners/kafka/streams/translation/KStreamsPayloadSerde.java index dd30ed3696eb..b28fda0c0a92 100644 --- a/runners/kafka-streams/src/main/java/org/apache/beam/runners/kafka/streams/translation/KStreamsPayloadSerde.java +++ b/runners/kafka-streams/src/main/java/org/apache/beam/runners/kafka/streams/translation/KStreamsPayloadSerde.java @@ -17,11 +17,11 @@ */ package org.apache.beam.runners.kafka.streams.translation; -import java.io.ByteArrayInputStream; +import com.google.protobuf.ByteString; +import com.google.protobuf.InvalidProtocolBufferException; import java.io.ByteArrayOutputStream; -import java.io.DataInputStream; -import java.io.DataOutputStream; import java.io.IOException; +import org.apache.beam.runners.kafka.streams.translation.KafkaStreamsPayloadProtos.KafkaStreamsPayload; import org.apache.beam.sdk.coders.Coder; import org.apache.beam.sdk.values.WindowedValue; import org.apache.kafka.common.errors.SerializationException; @@ -34,17 +34,11 @@ * (e.g. the repartition topic a {@code GroupByKey} introduces). Until now {@link KStreamsPayload} * only flowed in-JVM via {@code ProcessorContext#forward}, so no serialization was needed. * - *

The wire format is a one-byte discriminator followed by the variant body: - * - *

    - *
  • data: {@code [0x00][windowedValueCoder-encoded WindowedValue]} — the data element is - * encoded with the {@link Coder} supplied for the topic's PCollection. - *
  • watermark: {@code [0x01][long watermarkMillis][int sourcePartition][int - * totalSourcePartitions]} — the in-band watermark report, coder-independent. - *
- * - *

A {@link KStreamsPayloadSerde} is parameterized by the {@link Coder} for the data variant - * because different topics carry different element types; the watermark variant needs no coder. + *

The wire form is the {@link KafkaStreamsPayload} protobuf message — protobuf gives compatible + * schema evolution and compact varint encoding. The data variant carries the {@link WindowedValue} + * encoded with the {@link Coder} supplied for the topic's PCollection; the watermark variant + * carries the coder-independent watermark report. A {@link KStreamsPayloadSerde} is therefore + * parameterized by the data {@link Coder} (different topics carry different element types). * *

The serde assumes non-null payloads: the topics it is used on (repartition and watermark * fan-out) are not log-compacted, so no tombstone (null-valued) records occur. @@ -53,9 +47,6 @@ */ public final class KStreamsPayloadSerde implements Serde> { - private static final byte DATA_TAG = 0; - private static final byte WATERMARK_TAG = 1; - private final Coder> dataCoder; public KStreamsPayloadSerde(Coder> dataCoder) { @@ -75,46 +66,55 @@ public Deserializer> deserializer() { private final class PayloadSerializer implements Serializer> { @Override public byte[] serialize(String topic, KStreamsPayload payload) { - ByteArrayOutputStream out = new ByteArrayOutputStream(); - try { - if (payload.isData()) { - out.write(DATA_TAG); - dataCoder.encode(payload.getData(), out); - } else { - WatermarkPayload watermark = payload.asWatermark(); - DataOutputStream dataOut = new DataOutputStream(out); - dataOut.writeByte(WATERMARK_TAG); - dataOut.writeLong(watermark.getWatermarkMillis()); - dataOut.writeInt(watermark.getSourcePartition()); - dataOut.writeInt(watermark.getTotalSourcePartitions()); - dataOut.flush(); + KafkaStreamsPayload.Builder proto = KafkaStreamsPayload.newBuilder(); + if (payload.isData()) { + ByteArrayOutputStream encoded = new ByteArrayOutputStream(); + try { + dataCoder.encode(payload.getData(), encoded); + } catch (IOException e) { + throw new SerializationException("Failed to encode KStreamsPayload data element", e); } - } catch (IOException e) { - throw new SerializationException("Failed to serialize KStreamsPayload", e); + proto.setData( + KafkaStreamsPayload.DataPayload.newBuilder() + .setValue(ByteString.copyFrom(encoded.toByteArray()))); + } else { + WatermarkPayload watermark = payload.asWatermark(); + proto.setWatermark( + KafkaStreamsPayload.WatermarkPayload.newBuilder() + .setMillis(watermark.getWatermarkMillis()) + .setSourcePartition(watermark.getSourcePartition()) + .setTotalPartitions(watermark.getTotalSourcePartitions())); } - return out.toByteArray(); + return proto.build().toByteArray(); } } private final class PayloadDeserializer implements Deserializer> { @Override public KStreamsPayload deserialize(String topic, byte[] bytes) { - ByteArrayInputStream in = new ByteArrayInputStream(bytes); + KafkaStreamsPayload proto; try { - int tag = in.read(); - if (tag == DATA_TAG) { - return KStreamsPayload.data(dataCoder.decode(in)); - } else if (tag == WATERMARK_TAG) { - DataInputStream dataIn = new DataInputStream(in); - long watermarkMillis = dataIn.readLong(); - int sourcePartition = dataIn.readInt(); - int totalSourcePartitions = dataIn.readInt(); - return KStreamsPayload.watermark(watermarkMillis, sourcePartition, totalSourcePartitions); - } else { - throw new SerializationException("Unknown KStreamsPayload tag: " + tag); - } - } catch (IOException e) { - throw new SerializationException("Failed to deserialize KStreamsPayload", e); + proto = KafkaStreamsPayload.parseFrom(bytes); + } catch (InvalidProtocolBufferException e) { + throw new SerializationException("Failed to parse KStreamsPayload", e); + } + switch (proto.getPayloadCase()) { + case DATA: + try { + return KStreamsPayload.data(dataCoder.decode(proto.getData().getValue().newInput())); + } catch (IOException e) { + throw new SerializationException("Failed to decode KStreamsPayload data element", e); + } + case WATERMARK: + KafkaStreamsPayload.WatermarkPayload watermark = proto.getWatermark(); + return KStreamsPayload.watermark( + watermark.getMillis(), + watermark.getSourcePartition(), + watermark.getTotalPartitions()); + case PAYLOAD_NOT_SET: + default: + throw new SerializationException( + "KStreamsPayload has no payload variant set: " + proto.getPayloadCase()); } } } diff --git a/runners/kafka-streams/src/main/proto/org/apache/beam/runners/kafka/streams/translation/kafka_streams_payload.proto b/runners/kafka-streams/src/main/proto/org/apache/beam/runners/kafka/streams/translation/kafka_streams_payload.proto new file mode 100644 index 000000000000..e632d03e8a47 --- /dev/null +++ b/runners/kafka-streams/src/main/proto/org/apache/beam/runners/kafka/streams/translation/kafka_streams_payload.proto @@ -0,0 +1,52 @@ +/* + * 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.translation; + +option java_package = "org.apache.beam.runners.kafka.streams.translation"; +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 the downstream + // WatermarkManager needs. + 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; + // The source partition this report is for. + uint32 source_partition = 2; + // The total number of source partitions feeding the downstream stage. + uint32 total_partitions = 3; + } + + // 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/test/java/org/apache/beam/runners/kafka/streams/translation/KStreamsPayloadSerdeTest.java b/runners/kafka-streams/src/test/java/org/apache/beam/runners/kafka/streams/translation/KStreamsPayloadSerdeTest.java index 1e4c4c0263a4..19346f54763e 100644 --- a/runners/kafka-streams/src/test/java/org/apache/beam/runners/kafka/streams/translation/KStreamsPayloadSerdeTest.java +++ b/runners/kafka-streams/src/test/java/org/apache/beam/runners/kafka/streams/translation/KStreamsPayloadSerdeTest.java @@ -77,7 +77,18 @@ public void roundTripsTerminalMaxWatermark() { } @Test - public void unknownTagThrows() { + public void roundTripsNegativeWatermark() { + // Beam event times can be negative; sint64 must round-trip them losslessly. + KStreamsPayload payload = + KStreamsPayload.watermark(BoundedWindow.TIMESTAMP_MIN_VALUE.getMillis(), 0, 1); + assertThat( + roundTrip(payload).asWatermark().getWatermarkMillis(), + is(BoundedWindow.TIMESTAMP_MIN_VALUE.getMillis())); + } + + @Test + public void malformedBytesThrow() { + // 0x7f encodes field 15 with the invalid wire type 7, so protobuf parsing fails. byte[] bogus = new byte[] {(byte) 0x7f}; assertThrows( SerializationException.class, () -> serde.deserializer().deserialize(TOPIC, bogus)); diff --git a/sdks/java/build-tools/src/main/resources/beam/checkstyle/suppressions.xml b/sdks/java/build-tools/src/main/resources/beam/checkstyle/suppressions.xml index 1fa516918185..7714a380ecec 100644 --- a/sdks/java/build-tools/src/main/resources/beam/checkstyle/suppressions.xml +++ b/sdks/java/build-tools/src/main/resources/beam/checkstyle/suppressions.xml @@ -56,6 +56,7 @@ + From b1150c24d559ff4869543aaeedd2be7618a32384 Mon Sep 17 00:00:00 2001 From: junaiddshaukat Date: Sat, 27 Jun 2026 05:20:19 +0500 Subject: [PATCH 3/3] Move KafkaStreamsPayload proto to a vendored proto sub-module Per @je-ik review: use the vendored protobuf rather than raw com.google.protobuf (which dropped the checkstyle suppression). protoc emits raw com.google.protobuf and the relocation to the vendored package only happens at shade time, so the vendored classes have to come from a separately-shaded module. Move the proto into a small portability-nature sub-module (:runners:kafka-streams:proto, same pattern as the dataflow windmill proto module); the runner depends on its shaded output and the serde imports the vendored org.apache.beam.vendor.grpc.v1p69p0.com.google.protobuf.ByteString. This keeps the main runner on applyJavaNature (spotbugs stays on) and removes the suppressions.xml entry. Refs #18479 --- runners/kafka-streams/build.gradle | 10 +----- runners/kafka-streams/proto/build.gradle | 34 +++++++++++++++++++ .../main/proto}/kafka_streams_payload.proto | 4 +-- .../translation/KStreamsPayloadSerde.java | 6 ++-- .../beam/checkstyle/suppressions.xml | 1 - settings.gradle.kts | 1 + 6 files changed, 41 insertions(+), 15 deletions(-) create mode 100644 runners/kafka-streams/proto/build.gradle rename runners/kafka-streams/{src/main/proto/org/apache/beam/runners/kafka/streams/translation => proto/src/main/proto}/kafka_streams_payload.proto (93%) diff --git a/runners/kafka-streams/build.gradle b/runners/kafka-streams/build.gradle index 111c06b73722..2ba55e618308 100644 --- a/runners/kafka-streams/build.gradle +++ b/runners/kafka-streams/build.gradle @@ -22,15 +22,7 @@ def kafka_version = '3.9.0' applyJavaNature( automaticModuleName: 'org.apache.beam.runners.kafka.streams', - // protoc-generated message classes are not nullness-clean; exclude them from the checker. - generatedClassPatterns: [ - /^org\.apache\.beam\.runners\.kafka\.streams\.translation\.KafkaStreamsPayloadProtos.*/ - ], ) -// Adds protoc to compile the runner-internal control-message protos (e.g. KafkaStreamsPayload). -// applyGrpcNature is purely additive on top of applyJavaNature — it only wires the protobuf -// plugin; the gRPC codegen plugin is a no-op since these protos define no services. -applyGrpcNature() description = "Apache Beam :: Runners :: Kafka Streams" @@ -51,6 +43,7 @@ dependencies { 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:job-management", configuration: "shadow") implementation project(":runners:core-java") @@ -61,7 +54,6 @@ dependencies { implementation project(path: ":sdks:java:extensions:google-cloud-platform-core") implementation library.java.args4j implementation library.java.joda_time - implementation library.java.protobuf_java implementation library.java.slf4j_api implementation library.java.vendored_grpc_1_69_0 implementation library.java.vendored_guava_32_1_2_jre 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/src/main/proto/org/apache/beam/runners/kafka/streams/translation/kafka_streams_payload.proto b/runners/kafka-streams/proto/src/main/proto/kafka_streams_payload.proto similarity index 93% rename from runners/kafka-streams/src/main/proto/org/apache/beam/runners/kafka/streams/translation/kafka_streams_payload.proto rename to runners/kafka-streams/proto/src/main/proto/kafka_streams_payload.proto index e632d03e8a47..87e3de44397b 100644 --- a/runners/kafka-streams/src/main/proto/org/apache/beam/runners/kafka/streams/translation/kafka_streams_payload.proto +++ b/runners/kafka-streams/proto/src/main/proto/kafka_streams_payload.proto @@ -18,9 +18,9 @@ syntax = "proto3"; -package org.apache.beam.runners.kafka.streams.translation; +package org.apache.beam.runners.kafka.streams.v1; -option java_package = "org.apache.beam.runners.kafka.streams.translation"; +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 diff --git a/runners/kafka-streams/src/main/java/org/apache/beam/runners/kafka/streams/translation/KStreamsPayloadSerde.java b/runners/kafka-streams/src/main/java/org/apache/beam/runners/kafka/streams/translation/KStreamsPayloadSerde.java index b28fda0c0a92..912eb5fb6047 100644 --- a/runners/kafka-streams/src/main/java/org/apache/beam/runners/kafka/streams/translation/KStreamsPayloadSerde.java +++ b/runners/kafka-streams/src/main/java/org/apache/beam/runners/kafka/streams/translation/KStreamsPayloadSerde.java @@ -17,13 +17,13 @@ */ package org.apache.beam.runners.kafka.streams.translation; -import com.google.protobuf.ByteString; -import com.google.protobuf.InvalidProtocolBufferException; import java.io.ByteArrayOutputStream; import java.io.IOException; -import org.apache.beam.runners.kafka.streams.translation.KafkaStreamsPayloadProtos.KafkaStreamsPayload; +import org.apache.beam.runners.kafka.streams.v1.KafkaStreamsPayloadProtos.KafkaStreamsPayload; import org.apache.beam.sdk.coders.Coder; import org.apache.beam.sdk.values.WindowedValue; +import org.apache.beam.vendor.grpc.v1p69p0.com.google.protobuf.ByteString; +import org.apache.beam.vendor.grpc.v1p69p0.com.google.protobuf.InvalidProtocolBufferException; import org.apache.kafka.common.errors.SerializationException; import org.apache.kafka.common.serialization.Deserializer; import org.apache.kafka.common.serialization.Serde; diff --git a/sdks/java/build-tools/src/main/resources/beam/checkstyle/suppressions.xml b/sdks/java/build-tools/src/main/resources/beam/checkstyle/suppressions.xml index 7714a380ecec..1fa516918185 100644 --- a/sdks/java/build-tools/src/main/resources/beam/checkstyle/suppressions.xml +++ b/sdks/java/build-tools/src/main/resources/beam/checkstyle/suppressions.xml @@ -56,7 +56,6 @@ - diff --git a/settings.gradle.kts b/settings.gradle.kts index b92b254981fe..9a9cb1dd1acb 100644 --- a/settings.gradle.kts +++ b/settings.gradle.kts @@ -144,6 +144,7 @@ include(":runners:java-fn-execution") include(":runners:java-job-service") include(":runners:jet") include(":runners:kafka-streams") +include(":runners:kafka-streams:proto") include(":runners:local-java") include(":runners:portability:java") include(":runners:prism")