Skip to content

[SPARK-59291][STREAMING] Add return tags to the public StateSpec methods #58563

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
uros-b wants to merge 2 commits into from
+16 −0
Closed

[SPARK-59291][STREAMING] Add return tags to the public StateSpec methods #58563

Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
4ee96ef
[SPARK-59291][STREAMING] Add @return tags to the public StateSpec met…
uros-b Sep 7, 2026
3f2b168
Address review
uros-b Sep 12, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
16 changes: 16 additions & 0 deletions streaming/src/main/scala/org/apache/spark/streaming/StateSpec.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -72,22 +72,30 @@ sealed abstract class StateSpec[KeyType, ValueType, StateType, MappedType] exten

/**
* Set the RDD containing the initial states that will be used by `mapWithState`
*
* @return this `StateSpec` (to allow chaining)
*/
def initialState(rdd: RDD[(KeyType, StateType)]): this.type

/**
* Set the RDD containing the initial states that will be used by `mapWithState`
*
* @return this `StateSpec` (to allow chaining)
*/
def initialState(javaPairRDD: JavaPairRDD[KeyType, StateType]): this.type

/**
* Set the number of partitions by which the state RDDs generated by `mapWithState`
* will be partitioned. Hash partitioning will be used.
*
* @return this `StateSpec` (to allow chaining)
*/
def numPartitions(numPartitions: Int): this.type

/**
* Set the partitioner by which the state RDDs generated by `mapWithState` will be partitioned.
*
* @return this `StateSpec` (to allow chaining)
*/
def partitioner(partitioner: Partitioner): this.type

Expand All @@ -97,6 +105,8 @@ sealed abstract class StateSpec[KeyType, ValueType, StateType, MappedType] exten
* mapping function will be called one final time on the idle states that are going to be
* removed; [[org.apache.spark.streaming.State State.isTimingOut()]] set
* to `true` in that call.
*
* @return this `StateSpec` (to allow chaining)
*/
def timeout(idleDuration: Duration): this.type
}
Expand Down Expand Up @@ -153,6 +163,7 @@ object StateSpec {
* @tparam ValueType Class of the values
* @tparam StateType Class of the states data
* @tparam MappedType Class of the mapped data
* @return a new `StateSpec` for the `mapWithState` operation
*/
def function[KeyType, ValueType, StateType, MappedType](
mappingFunction: (Time, KeyType, Option[ValueType], State[StateType]) => Option[MappedType]
Expand All @@ -169,9 +180,11 @@ object StateSpec {
*
* @param mappingFunction The function applied on every data item to manage the associated state
* and generate the mapped data
* @tparam KeyType Class of the keys
* @tparam ValueType Class of the values
* @tparam StateType Class of the states data
* @tparam MappedType Class of the mapped data
* @return a new `StateSpec` for the `mapWithState` operation
*/
def function[KeyType, ValueType, StateType, MappedType](
mappingFunction: (KeyType, Option[ValueType], State[StateType]) => MappedType
Expand All @@ -196,6 +209,7 @@ object StateSpec {
* @tparam ValueType Class of the values
* @tparam StateType Class of the states data
* @tparam MappedType Class of the mapped data
* @return a new `StateSpec` for the `mapWithState` operation
*/
def function[KeyType, ValueType, StateType, MappedType](mappingFunction:
JFunction4[Time, KeyType, Optional[ValueType], State[StateType], Optional[MappedType]]):
Expand All @@ -218,9 +232,11 @@ object StateSpec {
*
* @param mappingFunction The function applied on every data item to manage the associated
* state and generate the mapped data
* @tparam KeyType Class of the keys
* @tparam ValueType Class of the values
* @tparam StateType Class of the states data
* @tparam MappedType Class of the mapped data
* @return a new `StateSpec` for the `mapWithState` operation
*/
def function[KeyType, ValueType, StateType, MappedType](
mappingFunction: JFunction3[KeyType, Optional[ValueType], State[StateType], MappedType]):
Expand Down