Skip to content

Commit

Permalink
MINOR: cleanup top level class JavaDocs for main interfaces of Kafka …
Browse files Browse the repository at this point in the history
…Streams DSL (4/N) (apache#18884)

Reviewers: Bill Bejeck <[email protected]>
  • Loading branch information
mjsax authored and manoj-mathivanan committed Feb 19, 2025
1 parent 92841c0 commit cd41796
Show file tree
Hide file tree
Showing 4 changed files with 263 additions and 288 deletions.
Original file line number Diff line number Diff line change
Expand Up @@ -18,38 +18,14 @@

import org.apache.kafka.common.utils.Bytes;
import org.apache.kafka.streams.KafkaStreams;
import org.apache.kafka.streams.KeyValue;
import org.apache.kafka.streams.StoreQueryParameters;
import org.apache.kafka.streams.StreamsConfig;
import org.apache.kafka.streams.Topology;
import org.apache.kafka.streams.state.SessionStore;

import java.time.Duration;

/**
* {@code SessionWindowedCogroupKStream} is an abstraction of a <i>windowed</i> record stream of {@link KeyValue} pairs.
* It is an intermediate representation of a {@link CogroupedKStream} in order to apply a windowed aggregation operation
* on the original {@link KGroupedStream} records resulting in a windowed {@link KTable} (a <emph>windowed</emph>
* {@code KTable} is a {@link KTable} with key type {@link Windowed Windowed<K>}).
* <p>
* {@link SessionWindows} are dynamic data driven windows.
* They have no fixed time boundaries, rather the size of the window is determined by the records.
* <p>
* The result is written into a local {@link SessionStore} (which is basically an ever-updating
* materialized view) that can be queried using the name provided in the {@link Materialized} instance.
* Furthermore, updates to the store are sent downstream into a windowed {@link KTable} changelog stream, where
* "windowed" implies that the {@link KTable} key is a combined key of the original record key and a window ID.
* New events are added to sessions until their grace period ends (see {@link SessionWindows#ofInactivityGapAndGrace(Duration, Duration)}).
* <p>
* A {@code SessionWindowedCogroupedKStream} must be obtained from a {@link CogroupedKStream} via
* {@link CogroupedKStream#windowedBy(SessionWindows)}.
*
* @param <K> Type of keys
* @param <V> Type of values
* @see KStream
* @see KGroupedStream
* @see SessionWindows
* @see CogroupedKStream
* Same as a {@link SessionWindowedKStream}, however, for multiple co-grouped {@link KStream KStreams}.
*/
public interface SessionWindowedCogroupedKStream<K, V> {

Expand Down
Loading

0 comments on commit cd41796

Please sign in to comment.