je-ik commented on code in PR #40068: URL: https://github.com/apache/beam/pull/40068#discussion_r3976217238
########## runners/kafka-streams/src/main/java/org/apache/beam/runners/kafka/streams/translation/FlushPayload.java: ########## @@ -0,0 +1,46 @@ +/* + * 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; + +/** + * The flush-only view of a {@link KStreamsPayload}, obtained via {@link KStreamsPayload#asFlush()}. + * As with {@link WatermarkPayload}, the accessors live here so they are only reachable once the + * caller has checked the kind and narrowed the payload. + * + * <p>A flush marker asks the stage that receives it to close its open bundle and flush the output, + * which is how a bundle is bounded in time. It arrives as an ordinary record, so the bundle is + * closed from {@code process()} rather than from a punctuator: transactions are committed by the + * Kafka Streams runtime in the background and are not exposed, so a bundle cannot be aligned with + * one, and trying to do it from a punctuator duplicated output against a real broker + * (https://github.com/apache/beam/issues/39633). + * + * <p>The partition fields exist so the marker can be targeted rather than broadcast. Broadcasting + * would give a downstream partition one flush per upstream partition, so N times more flushes than + * the interval asks for. Instead the producing partition addresses a slice of the downstream + * partitions and the slices tile the range, so each downstream partition gets exactly one flush per + * interval. Unlike a watermark, a flush needs no aggregation on arrival: there is nothing to hold + * and nothing to combine, because only one arrives. + */ Review Comment: Same here. ########## runners/kafka-streams/src/main/java/org/apache/beam/runners/kafka/streams/translation/FlushPayload.java: ########## @@ -0,0 +1,46 @@ +/* + * 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; + +/** + * The flush-only view of a {@link KStreamsPayload}, obtained via {@link KStreamsPayload#asFlush()}. + * As with {@link WatermarkPayload}, the accessors live here so they are only reachable once the + * caller has checked the kind and narrowed the payload. + * + * <p>A flush marker asks the stage that receives it to close its open bundle and flush the output, + * which is how a bundle is bounded in time. It arrives as an ordinary record, so the bundle is + * closed from {@code process()} rather than from a punctuator: transactions are committed by the + * Kafka Streams runtime in the background and are not exposed, so a bundle cannot be aligned with + * one, and trying to do it from a punctuator duplicated output against a real broker + * (https://github.com/apache/beam/issues/39633). + * + * <p>The partition fields exist so the marker can be targeted rather than broadcast. Broadcasting + * would give a downstream partition one flush per upstream partition, so N times more flushes than + * the interval asks for. Instead the producing partition addresses a slice of the downstream + * partitions and the slices tile the range, so each downstream partition gets exactly one flush per + * interval. Unlike a watermark, a flush needs no aggregation on arrival: there is nothing to hold + * and nothing to combine, because only one arrives. + */ +public interface FlushPayload { + + /** Which partition of the producing transform emitted this marker. */ + int getSourcePartition(); + + /** How many partitions the producing transform has in total. */ + int getTotalSourcePartitions(); Review Comment: If we are to compute the _target_ partition(s!), we need source partition, number of source partitions and number of target partitions. Given that in some cases this can produce target partition as `null`, should we instead just store the target partitions? Also - because the downstream transform does not care about this data, maybe the payload can be actually empty? ########## runners/kafka-streams/proto/src/main/proto/kafka_streams_payload.proto: ########## @@ -50,9 +50,32 @@ message KafkaStreamsPayload { bytes value = 1; } - // Exactly one variant is set; the oneof case discriminates data vs watermark. + // A request to close the open bundle and flush its output. + // + // A bundle has to be bounded in time as well as in size, or on a sparse stream the elements + // already fed to it wait for the next watermark. Closing it from a wall-clock punctuator does + // not work: transactions are committed by the Kafka Streams runtime in the background, are + // agnostic to punctuations, and are deliberately not exposed, so a bundle cannot be aligned with + // one. Instead a source emits this marker on its own punctuator and it travels the topology as + // an ordinary record, so a stage closes its bundle from process() rather than beside it. + // + // The partition fields are what let the marker be targeted rather than broadcast. Broadcasting + // would deliver one flush per upstream partition, so a downstream partition would see N times + // more flushes than the configured interval. Instead the producing partition addresses a slice + // of the downstream partitions, and the slices tile the whole range, so each downstream + // partition receives exactly one flush per interval. Review Comment: I'd suggest making the comments more concise. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
