Package org.apache.spark.sql.streaming

Interface TestGroupState<S>

Type Parameters:
S - User-defined type of the state to be stored for each group. Must be encodable into Spark SQL types (see Encoder for more details).
All Superinterfaces:
GroupState<S>, org.apache.spark.sql.catalyst.plans.logical.LogicalGroupState<S>
public interface TestGroupState<S> extends GroupState<S>
:: Experimental ::

The extended version of GroupState interface with extra getters of state machine fields to improve testability of the GroupState implementations which inherit from the extended interface.

Scala example of using TestGroupState: 


 // Please refer to ScalaDoc of `GroupState` for the Scala definition of `mappingFunction()`

 import org.apache.spark.api.java.Optional
 import org.apache.spark.sql.streaming.GroupStateTimeout
 import org.apache.spark.sql.streaming.TestGroupState
 // other imports

 // test class setups

 test("MapGroupsWithState state transition function") {
   // Creates the prevState input for the state transition function
   // with desired configs. The `create()` API would guarantee that
   // the generated instance has the same behavior as the one built by
   // engine with the same configs.
   val prevState = TestGroupState.create[Int](
     optionalState = Optional.empty[Int],
     timeoutConf = NoTimeout,
     batchProcessingTimeMs = 1L,
     eventTimeWatermarkMs = Optional.of(1L),
     hasTimedOut = false)

   val key: String = ...
   val values: Iterator[Int] = ...

   // Asserts the prevState is in init state without updates.
   assert(!prevState.isUpdated)

   // Calls the state transition function with the test previous state
   // with desired configs.
   mappingFunction(key, values, prevState)

   // Asserts the test GroupState object has been updated but not removed
   // after calling the state transition function
   assert(prevState.isUpdated)
   assert(!prevState.isRemoved)
 }

Java example of using TestGroupSate: 


 // Please refer to ScalaDoc of `GroupState` for the Java definition of `mappingFunction()`

 import org.apache.spark.api.java.Optional;
 import org.apache.spark.sql.streaming.GroupStateTimeout;
 import org.apache.spark.sql.streaming.TestGroupState;
 // other imports

 // test class setups

 // test `MapGroupsWithState` state transition function `mappingFunction()`
 public void testMappingFunctionWithTestGroupState() {
   // Creates the prevState input for the state transition function
   // with desired configs. The `create()` API would guarantee that
   // the generated instance has the same behavior as the one built by
   // engine with the same configs.
   TestGroupState<Int> prevState = TestGroupState.create(
     Optional.empty(),
     GroupStateTimeout.NoTimeout(),
     1L,
     Optional.of(1L),
     false);

   String key = ...;
   Integer[] values = ...;

   // Asserts the prevState is in init state without updates.
   Assertions.assertFalse(prevState.isUpdated());

   // Calls the state transition function with the test previous state
   // with desired configs.
   mappingFunction.call(key, Arrays.asList(values).iterator(), prevState);

   // Asserts the test GroupState object has been updated but not removed
   // after calling the state transition function
   Assertions.assertTrue(prevState.isUpdated());
   Assertions.assertFalse(prevState.isRemoved());
 }

Since:
3.2.0

  • Method Details

    • isRemoved

      boolean isRemoved()
      Whether the state has been marked for removing

    • isUpdated

      boolean isUpdated()
      Whether the state has been updated but not removed

    • getTimeoutTimestampMs

      Optional<Object> getTimeoutTimestampMs()
      Returns the timestamp if setTimeoutTimestamp() is called. Or, returns batch processing time + the duration when setTimeoutDuration() is called.

      Otherwise, returns Optional.empty if not set.

      Returns:
      (undocumented)