Package org.apache.cassandra.streaming

Class StreamSession

java.lang.Object
org.apache.cassandra.streaming.StreamSession
public class StreamSession extends Object
Handles the streaming a one or more streams 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. Session Initialization

   (a) A node (the initiator in the following) create a new StreamSession,
       initialize it init(StreamResultFuture), and then start it (start()).
       Starting a session causes a StreamInitMessage to be sent.
   (b) Upon reception of that StreamInitMessage, the follower creates its own StreamSession,
       and initializes it if it still does not exist.
   (c) After the initiator sends the StreamInitMessage, it invokes
       onInitializationComplete() to start the streaming prepare phase.

 2. Streaming preparation phase

   (a) A PrepareSynMessage is sent that includes a) what files/sections this node will stream to the follower
       (stored locally in a StreamTransferTask, one for each table) and b) what the follower needs to
       stream back (stored locally in a StreamReceiveTask, one for each table).
   (b) Upon reception of the PrepareSynMessage, the follower records which files/sections it will receive
       and send back a PrepareSynAckMessage, which contains a summary of the files/sections that will be sent to
       the initiator.
   (c) When the initiator receives the PrepareSynAckMessage, it records which files/sections it will
       receive, and then goes to it's Streaming phase (see next section). If the intiator is to receive files,
       it sends a PrepareAckMessage to the follower to indicate that it can start streaming to the initiator.
   (d) (Optional) If the follower receives a PrepareAckMessage, it enters it's Streaming phase.

 3. Streaming phase

   (a) The streaming phase is started at each node by calling startStreamingFiles(PrepareDirection).
       This will send, sequentially on each outbound streaming connection (see StreamingMultiplexedChannel),
       an OutgoingStreamMessage for each stream in each of the StreamTransferTask.
       Each OutgoingStreamMessage consists of a StreamMessageHeader that contains metadata about
       the stream, followed by the stream content itself. Once all the files for a StreamTransferTask are sent,
       the task is marked complete StreamTransferTask.complete(int).
   (b) On the receiving side, the incoming data is written to disk, and once the stream is fully received,
       it will be marked as complete (StreamReceiveTask.received(IncomingStream)). When all streams
       for the StreamReceiveTask have been received, the data is added to the CFS (and 2ndary indexes/MV are built),
        and the task is marked complete (taskCompleted(StreamReceiveTask)).
   (b) If during the streaming of a particular stream an error occurs on the receiving end of a stream
       (it may be either the initiator or the follower), the node will send a SessionFailedMessage
       to the sender and close the stream session.
   (c) When all transfer and receive tasks for a session are complete, the session moves to the Completion phase
       (maybeCompleted()).

 4. Completion phase

   (a) When the initiator finishes streaming, it enters the StreamSession.State.WAIT_COMPLETE state, and waits
       for the follower to send a CompleteMessage once it finishes streaming too. Once the CompleteMessage
       is received, initiator sets its own state to StreamSession.State.COMPLETE and closes all channels attached
       to this session.
In brief, the message passing looks like this (I for initiator, F for follwer): 
 (session init)
 I: StreamInitMessage
 (session prepare)
 I: PrepareSynMessage
 F: PrepareSynAckMessage
 I: PrepareAckMessage
 (stream - this can happen in both directions)
 I: OutgoingStreamMessage
 F: ReceivedMessage
 (completion)
 F: CompleteMessage
All messages which derive from StreamMessage are sent by the standard internode messaging (via MessagingService, while the actual files themselves are sent by a special "streaming" connection type. See StreamingMultiplexedChannel for details. Because of the asynchronous

  • Field Details

  • Constructor Details

  • Method Details

    • isFollower

      public boolean isFollower()

    • planId

      public TimeUUID planId()

    • sessionIndex

      public int sessionIndex()

    • streamOperation

      public StreamOperation streamOperation()

    • getStreamOperation

      public StreamOperation getStreamOperation()

    • getPendingRepair

      public TimeUUID getPendingRepair()

    • isPreview

      public boolean isPreview()

    • getPreviewKind

      public PreviewKind getPreviewKind()

    • getAggregator

      public StreamReceiver getAggregator(TableId tableId)

    • 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

    • attachInbound

      public boolean attachInbound(StreamingChannel channel)
      Attach a channel to this session upon receiving the first inbound message.
      Parameters:
      channel - The channel to attach.
      Returns:
      False if the channel was already attached, true otherwise.

    • attachOutbound

      public boolean attachOutbound(StreamingChannel channel)
      Attach a channel to this session upon sending the first outbound message.
      Parameters:
      channel - The channel to attach.
      Returns:
      False if the channel was already attached, true otherwise.

    • start

      public void start()
      invoked by the node that begins the stream session (it may be sending files, receiving files, or both)

    • addStreamRequest

      public void addStreamRequest(String keyspace, RangesAtEndpoint fullRanges, RangesAtEndpoint transientRanges, Collection<String> columnFamilies)
      Request data fetch task to this session. Here, we have to encode both _local_ range transientness (encoded in Replica itself, in RangesAtEndpoint) and _remote_ (source) range transientmess, which is encoded by splitting ranges into full and transient.
      Parameters:
      keyspace - Requesting keyspace
      fullRanges - Ranges to retrieve data that will return full data from the source
      transientRanges - Ranges to retrieve data that will return transient data from the source
      columnFamilies - ColumnFamily names. Can be empty if requesting all CF under the keyspace.

    • getOutgoingStreamsForRanges

      public List<OutgoingStream> getOutgoingStreamsForRanges(RangesAtEndpoint replicas, Collection<ColumnFamilyStore> stores, TimeUUID pendingRepair, PreviewKind previewKind)

    • 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

    • getChannel

      public StreamingMultiplexedChannel getChannel()

    • isSuccess

      public boolean isSuccess()
      Return if this session completed successfully.
      Returns:
      true if session completed successfully.

    • isFailedOrAborted

      public boolean isFailedOrAborted()
      Return if this session was failed or aborted
      Returns:
      true if session was failed or aborted

    • messageReceived

      public void messageReceived(StreamMessage message)

    • onInitializationComplete

      public void onInitializationComplete()
      Call back when connection initialization is complete to start the prepare phase.

    • onError

      public io.netty.util.concurrent.Future<?> onError(Throwable e)
      Signal an error to this stream session: if it's an EOF exception, it tries to understand if the socket was closed after completion or because the peer was down, otherwise sends a SessionFailedMessage and closes the session as StreamSession.State.FAILED.

    • prepare

      public void prepare(Collection<StreamRequest> requests, Collection<StreamSummary> summaries)
      Prepare this session for sending/receiving files.

    • countStreamedIn

      public void countStreamedIn(boolean isEntireSSTable)

    • prepareAsync

      public void prepareAsync(Collection<StreamRequest> requests, Collection<StreamSummary> summaries)
      Finish preparing the session. This method is blocking (memtables are flushed in addTransferRanges(java.lang.String, org.apache.cassandra.locator.RangesAtEndpoint, java.util.Collection<java.lang.String>, boolean)), so the logic should not execute on the main IO thread (read: netty event loop).

    • checkAvailableDiskSpaceAndCompactions

      public static boolean checkAvailableDiskSpaceAndCompactions(Collection<StreamSummary> summaries, @Nullable TimeUUID planId, @Nullable String remoteAddress, boolean isForIncremental)
      Makes sure that we expect to have enough disk space available for the new streams, taking into consideration the ongoing compactions and streams.

    • streamSent

      public void streamSent(OutgoingStreamMessage message)
      Call back after sending StreamMessageHeader.
      Parameters:
      message - sent stream message

    • sendControlMessage

      protected io.netty.util.concurrent.Future<?> sendControlMessage(StreamMessage message)

    • receive

      public void receive(IncomingStreamMessage message)
      Call back after receiving a stream.
      Parameters:
      message - received stream

    • progress

      public void progress(String filename, ProgressInfo.Direction direction, long bytes, long delta, long total)

    • received

      public void received(TableId tableId, int sequenceNumber)

    • 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.

    • sessionTimeout

      public void sessionTimeout()
      Call back on receiving StreamMessage.Type.SESSION_FAILED message.

    • getSessionInfo

      public SessionInfo getSessionInfo()
      Returns:
      Current snapshot of this session info.

    • taskCompleted

      public void taskCompleted(StreamReceiveTask completedTask)

    • taskCompleted

      public void taskCompleted(StreamTransferTask completedTask)

    • prepareReceiving

      public void prepareReceiving(StreamSummary summary)

    • getNumRequests

      public int getNumRequests()

    • getNumTransfers

      public int getNumTransfers()

    • createLogTag

      public static String createLogTag(StreamSession session)

    • createLogTag

      public static String createLogTag(StreamSession session, StreamingChannel channel)

    • createLogTag

      public static String createLogTag(StreamSession session, io.netty.channel.Channel channel)

    • createLogTag

      public static String createLogTag(StreamSession session, Object channelId)

    • abort

      public void abort()

    • toString

      public String toString()
      Overrides:
      toString in class Object

    • boundStackTrace

      public static StringBuilder boundStackTrace(Throwable e, int limit, StringBuilder out)

    • boundStackTrace

      public static StringBuilder boundStackTrace(Throwable e, int limit, int counter, Set<Throwable> visited, StringBuilder out)