StreamSession (apache-cassandra API)
Date: 2014-10-21

org.apache.cassandra.streaming

## Class StreamSession

- java.lang.Object
- - org.apache.cassandra.streaming.StreamSession

- All Implemented Interfaces:
    IEndpointStateChangeSubscriber

      public class StreamSession
      extends java.lang.Object
      implements IEndpointStateChangeSubscriber

  Handles the streaming a one or more section of one of more sstables to and from a specific remote node. Both this node and the remote one will create a similar symmetrical StreamSession. A streaming session has the following life-cycle: 1. Connections Initialization (a) A node (the initiator in the following) create a new StreamSession, initialize it (init()) and then start it (start()). Start will create a `ConnectionHandler` that will create two connections to the remote node (the follower in the following) with whom to stream and send a StreamInit message. The first connection will be the incoming connection for the initiator, and the second connection will be the outgoing. (b) Upon reception of that StreamInit message, the follower creates its own StreamSession, initialize it if it still does not exist, and attach connecting socket to its ConnectionHandler according to StreamInit message's isForOutgoing flag. (d) When the both incoming and outgoing connections are established, StreamSession calls StreamSession#onInitializationComplete method to start the streaming prepare phase (StreamResultFuture.startStreaming()). 2. Streaming preparation phase (a) This phase is started when the initiator onInitializationComplete() method is called. This method sends a PrepareMessage that includes what files/sections this node will stream to the follower (stored in a StreamTransferTask, each column family has it's own transfer task) and what the follower needs to stream back (StreamReceiveTask, same as above). If the initiator has nothing to receive from the follower, it goes directly to its Streaming phase. Otherwise, it waits for the follower PrepareMessage. (b) Upon reception of the PrepareMessage, the follower records which files/sections it will receive and send back its own PrepareMessage with a summary of the files/sections that will be sent to the initiator (prepare()). After having sent that message, the follower goes to its Streamning phase. (c) When the initiator receives the follower PrepareMessage, it records which files/sections it will receive and then goes to his own Streaming phase. 3. Streaming phase (a) The streaming phase is started by each node (the sender in the follower, but note that each side of the StreamSession may be sender for some of the files) involved by calling startStreamingFiles(). This will sequentially send a FileMessage for each file of each SteamTransferTask. Each FileMessage consists of a FileMessageHeader that indicates which file is coming and then start streaming the content for that file (StreamWriter in FileMessage.serialize()). When a file is fully sent, the fileSent() method is called for that file. If all the files for a StreamTransferTask are sent (StreamTransferTask.complete()), the task is marked complete (taskCompleted()). (b) On the receiving side, a SSTable will be written for the incoming file (StreamReader in FileMessage.deserialize()) and once the FileMessage is fully received, the file will be marked as complete (received()). When all files for the StreamReceiveTask have been received, the sstables are added to the CFS (and 2ndary index are built, StreamReceiveTask.complete()) and the task is marked complete (taskCompleted()) (b) If during the streaming of a particular file an I/O error occurs on the receiving end of a stream (FileMessage.deserialize), the node will retry the file (up to DatabaseDescriptor.getMaxStreamingRetries()) by sending a RetryMessage to the sender. On receiving a RetryMessage, the sender simply issue a new FileMessage for that file. (c) When all transfer and receive tasks for a session are complete, the move to the Completion phase (maybeCompleted()). 4. Completion phase (a) When a node has finished all transfer and receive task, it enter the completion phase (maybeCompleted()). If it had already received a CompleteMessage from the other side (it is in the WAIT_COMPLETE state), that session is done is is closed (closeSession()). Otherwise, the node switch to the WAIT_COMPLETE state and send a CompleteMessage to the other side.

- - ### Nested Class Summary

    Nested Classes
    Modifier and Type | Class and Description
    --- | ---
    `static class` | `StreamSession.SSTableStreamingSections`
    `static class` | `StreamSession.State`

  - ### Field Summary

    Fields
    Modifier and Type | Field and Description
    --- | ---
    `java.net.InetAddress` | `connecting` Actual connecting address.
    `ConnectionHandler` | `handler`
    `java.net.InetAddress` | `peer` Streaming endpoint.

  - ### Constructor Summary

    Constructors
    Constructor and Description
    `StreamSession(java.net.InetAddress peer, java.net.InetAddress connecting, StreamConnectionFactory factory, int index)` Create new streaming session with the peer.

  - ### Method Summary

    Methods
    Modifier and Type | Method and Description
    --- | ---
    `void` | `addStreamRequest(java.lang.String keyspace, java.util.Collection<Range<Token>> ranges, java.util.Collection<java.lang.String> columnFamilies, long repairedAt)` Request data fetch task to this session.
    `void` | `addTransferFiles(java.util.Collection<StreamSession.SSTableStreamingSections> sstableDetails)`
    `void` | `addTransferRanges(java.lang.String keyspace, java.util.Collection<Range<Token>> ranges, java.util.Collection<java.lang.String> columnFamilies, boolean flushTables, long repairedAt)` Set up transfer for specific keyspace/ranges/CFs Used in repair - a streamed sstable in repair will be marked with the given repairedAt time
    `void` | `beforeChange(java.net.InetAddress endpoint, EndpointState currentState, ApplicationState newStateKey, VersionedValue newValue)`
    `void` | `complete()` Check if session is completed on receiving `StreamMessage.Type.COMPLETE` message.
    `java.net.Socket` | `createConnection()`
    `java.lang.String` | `description()`
    `void` | `doRetry(FileMessageHeader header, java.lang.Throwable e)`
    `void` | `fileSent(FileMessageHeader header)` Call back after sending FileMessageHeader.
    `SessionInfo` | `getSessionInfo()`
    `void` | `init(StreamResultFuture streamResult)` Bind this session to report to specific `StreamResultFuture` and perform pre-streaming initialization.
    `boolean` | `isSuccess()` Return if this session completed successfully.
    `void` | `messageReceived(StreamMessage message)`
    `void` | `onAlive(java.net.InetAddress endpoint, EndpointState state)`
    `void` | `onChange(java.net.InetAddress endpoint, ApplicationState state, VersionedValue value)`
    `void` | `onDead(java.net.InetAddress endpoint, EndpointState state)`
    `void` | `onError(java.lang.Throwable e)` l Call back for handling exception during streaming.
    `void` | `onInitializationComplete()` Call back when connection initialization is complete to start the prepare phase.
    `void` | `onJoin(java.net.InetAddress endpoint, EndpointState epState)` Use to inform interested parties about the change in the state for specified endpoint
    `void` | `onRemove(java.net.InetAddress endpoint)`
    `void` | `onRestart(java.net.InetAddress endpoint, EndpointState epState)` Called whenever a node is restarted.
    `java.util.UUID` | `planId()`
    `void` | `prepare(java.util.Collection<StreamRequest> requests, java.util.Collection<StreamSummary> summaries)` Prepare this session for sending/receiving files.
    `void` | `progress(Descriptor desc, ProgressInfo.Direction direction, long bytes, long total)`
    `void` | `receive(IncomingFileMessage message)` Call back after receiving FileMessageHeader.
    `void` | `received(java.util.UUID cfId, int sequenceNumber)`
    `void` | `retry(java.util.UUID cfId, int sequenceNumber)` Call back on receiving `StreamMessage.Type.RETRY` message.
    `void` | `sessionFailed()` Call back on receiving `StreamMessage.Type.SESSION_FAILED` message.
    `int` | `sessionIndex()`
    `void` | `start()`
    `StreamSession.State` | `state()`
    `void` | `state(StreamSession.State newState)` Set current state to `newState`.
    `void` | `taskCompleted(StreamReceiveTask completedTask)`
    `void` | `taskCompleted(StreamTransferTask completedTask)`

    - ### Methods inherited from class java.lang.Object

      `clone, equals, finalize, getClass, hashCode, notify, notifyAll, toString, wait, wait, wait`

- - ### Field Detail

    - #### peer

          public final java.net.InetAddress peer

      Streaming endpoint. Each `StreamSession` is identified by this InetAddress which is broadcast address of the node streaming.

    - #### connecting

          public final java.net.InetAddress connecting

      Actual connecting address. Can be the same as peer.

    - #### handler

          public final ConnectionHandler handler

  - ### Constructor Detail

    - #### StreamSession

          public StreamSession(java.net.InetAddress peer,
                       java.net.InetAddress connecting,
                       StreamConnectionFactory factory,
                       int index)

      Create new streaming session with the peer.
      Parameters:
        `peer` - Address of streaming peer
        `connecting` - Actual connecting address
        `factory` - is used for establishing connection

  - ### Method Detail

    - #### planId

          public java.util.UUID planId()

    - #### sessionIndex

          public int sessionIndex()

    - #### description

          public java.lang.String description()

    - #### init

          public void init(StreamResultFuture streamResult)

      Bind this session to report to specific `StreamResultFuture` and perform pre-streaming initialization.
      Parameters:
        `streamResult` - result to report to

    - #### start

          public void start()

    - #### createConnection

          public java.net.Socket createConnection()
                                           throws java.io.IOException

      Throws:
        `java.io.IOException`

    - #### addStreamRequest

          public void addStreamRequest(java.lang.String keyspace,
                              java.util.Collection<Range<Token>> ranges,
                              java.util.Collection<java.lang.String> columnFamilies,
                              long repairedAt)

      Request data fetch task to this session.
      Parameters:
        `keyspace` - Requesting keyspace
        `ranges` - Ranges to retrieve data
        `columnFamilies` - ColumnFamily names. Can be empty if requesting all CF under the keyspace.

    - #### addTransferRanges

          public void addTransferRanges(java.lang.String keyspace,
                               java.util.Collection<Range<Token>> ranges,
                               java.util.Collection<java.lang.String> columnFamilies,
                               boolean flushTables,
                               long repairedAt)

      Set up transfer for specific keyspace/ranges/CFs Used in repair - a streamed sstable in repair will be marked with the given repairedAt time
      Parameters:
        `keyspace` - Transfer keyspace
        `ranges` - Transfer ranges
        `columnFamilies` - Transfer ColumnFamilies
        `flushTables` - flush tables?
        `repairedAt` - the time the repair started.

    - #### addTransferFiles

          public void addTransferFiles(java.util.Collection<StreamSession.SSTableStreamingSections> sstableDetails)

    - #### state

          public void state(StreamSession.State newState)

      Set current state to `newState`.
      Parameters:
        `newState` - new state to set

    - #### state

          public StreamSession.State state()

      Returns:
        current state

    - #### isSuccess

          public boolean isSuccess()

      Return if this session completed successfully.
      Returns:
        true if session completed successfully.

    - #### messageReceived

          public void messageReceived(StreamMessage message)

    - #### onInitializationComplete

          public void onInitializationComplete()

      Call back when connection initialization is complete to start the prepare phase.

    - #### onError

          public void onError(java.lang.Throwable e)

      l Call back for handling exception during streaming.
      Parameters:
        `e` - thrown exception

    - #### prepare

          public void prepare(java.util.Collection<StreamRequest> requests,
                     java.util.Collection<StreamSummary> summaries)

      Prepare this session for sending/receiving files.

    - #### fileSent

          public void fileSent(FileMessageHeader header)

      Call back after sending FileMessageHeader.
      Parameters:
        `header` - sent header

    - #### receive

          public void receive(IncomingFileMessage message)

      Call back after receiving FileMessageHeader.
      Parameters:
        `message` - received file

    - #### progress

          public void progress(Descriptor desc,
                      ProgressInfo.Direction direction,
                      long bytes,
                      long total)

    - #### received

          public void received(java.util.UUID cfId,
                      int sequenceNumber)

    - #### retry

          public void retry(java.util.UUID cfId,
                   int sequenceNumber)

      Call back on receiving `StreamMessage.Type.RETRY` message.
      Parameters:
        `cfId` - ColumnFamily ID
        `sequenceNumber` - Sequence number to indicate which file to stream again

    - #### complete

          public void complete()

      Check if session is completed on receiving `StreamMessage.Type.COMPLETE` message.

    - #### sessionFailed

          public void sessionFailed()

      Call back on receiving `StreamMessage.Type.SESSION_FAILED` message.

    - #### doRetry

          public void doRetry(FileMessageHeader header,
                     java.lang.Throwable e)

    - #### getSessionInfo

          public SessionInfo getSessionInfo()

      Returns:
        Current snapshot of this session info.

    - #### taskCompleted

          public void taskCompleted(StreamReceiveTask completedTask)

    - #### taskCompleted

          public void taskCompleted(StreamTransferTask completedTask)

    - #### onJoin

          public void onJoin(java.net.InetAddress endpoint,
                    EndpointState epState)

      Description copied from interface: `IEndpointStateChangeSubscriber`
      Use to inform interested parties about the change in the state for specified endpoint
      Specified by:
        `onJoin` in interface `IEndpointStateChangeSubscriber`
      Parameters:
        `endpoint` - endpoint for which the state change occurred.
        `epState` - state that actually changed for the above endpoint.

    - #### beforeChange

          public void beforeChange(java.net.InetAddress endpoint,
                          EndpointState currentState,
                          ApplicationState newStateKey,
                          VersionedValue newValue)

      Specified by:
        `beforeChange` in interface `IEndpointStateChangeSubscriber`

    - #### onChange

          public void onChange(java.net.InetAddress endpoint,
                      ApplicationState state,
                      VersionedValue value)

      Specified by:
        `onChange` in interface `IEndpointStateChangeSubscriber`

    - #### onAlive

          public void onAlive(java.net.InetAddress endpoint,
                     EndpointState state)

      Specified by:
        `onAlive` in interface `IEndpointStateChangeSubscriber`

    - #### onDead

          public void onDead(java.net.InetAddress endpoint,
                    EndpointState state)

      Specified by:
        `onDead` in interface `IEndpointStateChangeSubscriber`

    - #### onRemove

          public void onRemove(java.net.InetAddress endpoint)

      Specified by:
        `onRemove` in interface `IEndpointStateChangeSubscriber`

    - #### onRestart

          public void onRestart(java.net.InetAddress endpoint,
                       EndpointState epState)

      Description copied from interface: `IEndpointStateChangeSubscriber`
      Called whenever a node is restarted. Note that there is no guarantee when that happens that the node was previously marked down. It will have only if `state.isAlive() == false` as `state` is from before the restarted node is marked up.
      Specified by:
        `onRestart` in interface `IEndpointStateChangeSubscriber`

Copyright © 2014 The Apache Software Foundation