feedback

sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/ListStateImplWithTTL.scala

@@ -24,14 +24,16 @@ import org.apache.spark.sql.streaming.{ListState, TTLConfig}
 import org.apache.spark.util.NextIterator
 
 /**
- * Provides concrete implementation for list of values associated with a state variable
- * used in the streaming transformWithState operator.
+ * Class that provides a concrete implementation for a list state state associated with state
+ * variables (with ttl expiration support) used in the streaming transformWithState operator.
  *
  * @param store - reference to the StateStore instance to be used for storing state
  * @param stateName - name of logical state partition
- * @param keyEnc - Spark SQL encoder for key
+ * @param keyExprEnc - Spark SQL encoder for key
  * @param valEncoder - Spark SQL encoder for value
- * @tparam S - data type of object that will be stored in the list
+ * @param ttlConfig  - TTL configuration for values  stored in this state
+ * @param batchTimestampMs - current batch processing timestamp.
+ * @tparam S - data type of object that will be stored
  */
 class ListStateImplWithTTL[S](
   store: StateStore,
@@ -147,7 +149,6 @@ class ListStateImplWithTTL[S](
   }
 
   /**
-   *
    * Loops through all the values associated with the grouping key, and removes
    * the expired elements from the list.
    * @param groupingKey grouping key for which cleanup should be performed.

sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/TTLState.scala

@@ -73,7 +73,7 @@ trait TTLState {
    *
    * @param groupingKey grouping key for which cleanup should be performed.
    *
-   * @return true if the state was cleared, false otherwise.
+   * @return how many state objects were cleaned up.
    */
   def clearIfExpired(groupingKey: Array[Byte]): Long
 }

sql/core/src/test/scala/org/apache/spark/sql/execution/streaming/state/StatefulProcessorHandleSuite.scala

@@ -255,15 +255,14 @@ class StatefulProcessorHandleSuite extends StateVariableSuiteBase {
     }
   }
 
-
   test(s"ttl States are not populated for timeMode=None") {
     tryWithProviderResource(newStoreProviderWithStateVariable(true)) { provider =>
       val store = provider.getStore(0)
       val handle = new StatefulProcessorHandleImpl(store,
         UUID.randomUUID(), keyExprEncoder, TimeMode.None())
 
-      handle.getValueState("testState", Encoders.STRING)
-      handle.getListState("testState", Encoders.STRING)
+      handle.getValueState("testValueState", Encoders.STRING)
+      handle.getListState("testListState", Encoders.STRING)
 
       assert(handle.ttlStates.isEmpty)
     }

sql/core/src/test/scala/org/apache/spark/sql/streaming/TransformWithListStateTTLSuite.scala

@@ -98,9 +98,10 @@ class ListStateTTLProcessor(ttlConfig: TTLConfig)
 class TransformWithListStateTTLSuite extends TransformWithStateTTLTest {
 
   import testImplicits._
+
   override def getProcessor(ttlConfig: TTLConfig):
-  StatefulProcessor[String, InputEvent, OutputEvent] = {
-    new ListStateTTLProcessor(ttlConfig)
+    StatefulProcessor[String, InputEvent, OutputEvent] = {
+      new ListStateTTLProcessor(ttlConfig)
   }
 
   override def getStateTTLMetricName: String = "numListStateWithTTLVars"

sql/core/src/test/scala/org/apache/spark/sql/streaming/TransformWithValueStateTTLSuite.scala

@@ -163,9 +163,10 @@ case class MultipleValueStatesTTLProcessor(
 class TransformWithValueStateTTLSuite extends TransformWithStateTTLTest {
 
   import testImplicits._
+
   override def getProcessor(ttlConfig: TTLConfig):
     StatefulProcessor[String, InputEvent, OutputEvent] = {
-    new ValueStateTTLProcessor(ttlConfig)
+      new ValueStateTTLProcessor(ttlConfig)
   }
 
   override def getStateTTLMetricName: String = "numValueStateWithTTLVars"