stream.groupByKey().aggregate(initializer, 
                             aggregator,
                             merger,  
                             SessionWindows.inactivityGap(FIVE_MINUTES)
                 .until(ONE_HOUR),
   aggregateValueSerde,
  “session-store”);

In order to process SessionWindows we’ll need to add a new Processor. This will be responsible for creating sessions, merging existing sessions into sessions with larger windows, and producing aggregates from a session’s values.

On each incoming record the process method will:

1. Find any adjacent sessions that either start or end within the inactivity gap, i,e., where the end time of the session is > now - inactivity gap, or the start time is < now + inactivity gap.

2. Merge any existing sessions into a new larger session using the SessionMerger to merge the aggregates of the existing sessions.

3. Aggregate the value record being processed with the merged session.

4. Store the new merged session in the SessionStore. 

5. Remove any merged sessions from the SessionStore.

Late arriving data

Late arriving data is mostly treated the same as non-late arriving data, i.e., it can create a new session or be merged into an existing one. The only difference is that if the data has arrived after the retention period, defined by SessionWindows.until(..), a new session will be created and aggregated, but it will not be persisted to the store.

SessionWindows

We propose to add a new class SessionWindows. SessionWindows will be able to be used with new overloaded operations on KGroupedStream, i.e, aggregate(…), count(..), reduce(…). A SessionWindows will have a defined gap, that represents the period of inactivity. It will also provide a method, until(...), to specify how long the data is retained for, i.e., to allow for late arriving data.

SessionStore

We propose to add a new type of StateStore, SessionStore. A SessionStore, is a segmented store, similar to a WindowStore, but the segments are indexed by session end time. We index by end time so that we can expire (remove) the Segments containing sessions where session endTime < stream-time - retention-period.

The records in the SessionStore will be stored by a SessionKey. The SessionKey is a composite of the record key, window start, and window end times. The start and end times of the SessionKey are driven by the data. If the Session only has a single value then start == end.  The segment a Session is stored in is determined by SessionKey.end. Fetch requests against the SessionStore use both the SessionKey.start and Session.end to find sessions to merge.

 Each Segment is for a particular interval of time. To work out which Segment a session belongs in we simply divide SessionKey.end by the segment interval. The segment interval is calculated as Math.max(retentionPeriod / (numSegments - 1), MIN_SEGMENT_INTERVAL).

As session aggregates arrive, i.e., on put, the implementation of SessionStore will:

1. use SessionKey.end to get an existing segment or create a new Segment to store the aggregate in. A new Segment will only be created if the SessionKey.end is within the retention period.

2. If the Segment is non-null, we add the aggregate to the Segment.

3. If the Segment was null,  this signals that the record is late and has arrived after the retention period. This record is not added to the store.

Fetch

When SessionStore.fetch(...) is called we find all the aggregates for the record key where SessionKey.end >= earliestEndTime && SessionKey.start <= latestStartTime. In order to do this:

1. Find the Segments to search by getting all Segments starting from earliestEndTime

2. Define the range query as:

1. from =  (record-key, end=earliestEndTime, start=0)

2. to = (record-key, end=Long.MAX_VALUE, start=latestStartTime)

For example, if for an arbitrary record key we had the following sessions in the store:

SessionWindows

public class SessionWindows {

    /**
     * Create a new SessionWindows with the specified inactivityGap
     *
     * @param inactivityGap
     * @return
     */
    public static SessionWindows inactivityGap(final long inactivityGap)


    /**
     * Set the window maintain duration in milliseconds of streams time.
     * This retention time is a guaranteed <i>lower bound</i> for how long 
     * a window will be maintained.
     *
     * @return  itself
     */
    public SessionWindows until(long durationMs)

}

SessionMerger

/**
 * The interface for merging aggregate values for {@link SessionWindows} with the given key
 *
 * @param <K>   key type
 * @param <T>   aggregate value type
 */
public interface SessionMerger<K, T> {
   /**
    * Compute a new aggregate from the key and two aggregates
    *
    * @param aggKey    the key of the record
    * @param aggOne    the first aggregate
    * @param aggTwo    the second aggregate
    * @return          the new aggregate value
    */
   T apply(K aggKey, T aggOne, T aggTwo);
}

SessionStore

/**
* Interface for storing the aggregated values of sessions
* @param <K>   type of the record keys
* @param <AGG> type of the aggregated values
*/
public interface SessionStore<K, AGG> extends StateStore, ReadOnlySessionStore<K, AGG> {

    /**
     * Find any aggregated session values with the matching key and where the 
     * session’s end time is >= earliestSessionEndTime, i.e, the oldest session to 
     * merge with, and the session’s start time is <= latestSessionStartTime, i.e,
     * the newest session to merge with.
     */
   KeyValueIterator<SessionKey<K>, AGG> findSessionsToMerge(final K key, final long earliestSessionEndTime, final long latestSessionStartTime);
 
   /**
    * Remove the aggregated value for the session with the matching SessionKey
    */
   void remove(final SessionKey<K> sessionKey);

   /**
    * Write the aggregated result for the given SessionKey
    */	 
   void put(final SessionKey<K> key, AGG result);
}

ReadOnlySessionStore

/**
 * A session store that only supports read operations.
 * Implementations should be thread-safe as concurrent reads and writes
 * are expected.
 *
 * @param <K> the key type
 * @param <V> the value type
 */
@InterfaceStability.Unstable
public interface ReadOnlySessionStore<K, AGG> {

    /**
     * Retrieve all aggregated sessions for the provided key
     * @param    key record key to find aggregated session values for
     * @return   KeyValueIterator containing all sessions for the provided key.
     */
    KeyValueIterator<SessionKey<K>, AGG> fetch(final K key);
}

/**
 * Represents the key for a Session Window
 */
public class SessionKey<K> {
    // record key
    private final K key;
    // session start time
    private final long start;
    // session end time
    private final long end; 
}