Implementation

Classes

Server

• ConnectionManager
• Connection
• TcpConnection
• UdpConnection
• Server
• Collector
• CollectorWorker
• TopologyChecker
• Status

Client

• SquallClient
• SquallClientHandler

Details

ConnectionManager[C]

• Set of Connections: keeps a list of open connections to server and enables Server and Connection to modify connection set.

It is a singleton object so there will be only one manager while topology is running. It is thread-safe since set is synchronized.

It exchanges a protocol to define characteristics of the connection, accordingly either creates new connection or refuses it. Protocol exchange requires tackling a bit since reading is blocking: One thread waits for input and another one notes the time. If client doesn't talk for a given time, timeout occurs and waiting thread is interrupted and connection is refused. Moreover, if client doesn't follow the protocol, connection again is refused. However, if default is set, we'd not need this trick.

Connection (Thread)[C]

• stream(ConcurrentLinkedQueue): keeps the tuples to be sent through this connection.
• status: status of the connection, namely running or closed.

This is an abstract class that extends Thread to poll its queue and to send the tuples to its pair. It provides default close and tuple addition methods. However, according to exchanged protocol in the start of the connection, connection can apply filtering for which tuples to be sent.

TcpConnection[C]

• socket(Socket): Communication end point.
• pw(PrintWriter): Output stream writer that is used for writing tuples.

Server (Thread)[C]

• serverSocket(ServerSocket): Listens for coming requests and handles new connection request via using ConnectionManager.

Passes socket information to connection manager to handle new connection.

Collector (Thread)[C]

• workers(Set<CollectionWorker>): Pointers to the worker thread that are working on connections from bolts.
• topologyChecker(TopologyChecker): A timer thread to get alive/killed topologies periodically.
• topologies(Set<String>): Topologies that are currently being served by this collector.

What this class does for bolts is very similar to what Server does for clients in the high level. Communication protocol and direction are different.

CollectorWorker (Thread)[C]:

• socket(Socket): Communication end point.
• br(BufferedReader): Input stream reader to collect tuples from corresponding bolt.

Each worker matches a different bolt connection and they read processed tuples from bolts and put tuples to main store of ConnectionManager to be sent to clients.

TopologyChecker (Timer)[C]:

Retrieves alive/killed topologies from Nimbus and closes worker connections from killed topologies. This is done via an external thread because when a topology is killed, cleanup methods of the bolts aren't called and worker threads do only reading which is blocking. Therefore, they can't learn if other end point is closed.

Status [E]

An enum to denote the socket state. Since some calls on socket are blocking, this state enables different threads to close the socket. Each class(composed of socket and thread) contains a volatile status property.

SquallClient (Thread)[C]

• store(ConcurrentLinkedQueue): Buffer of received tuples.
• in(BufferedReader): Stream to read tuples. According to configuration new tuples are put into store for later use or are passed to handler to be processed as soon as possible.
• out(PrintWriter): Sends heartbeat to check if server is down because reading is blocking.

SquallReceiverHandler [I]

• newTupleArrived: consumes new arriving tuple.

Notes:

• To get it worked, bolts are modified to create connections to collector in their prepare methods and to close opened connections in their cleanup methods. They get hostname and port via config, the same way they get data path, etc.
• Since one node tries to serve multiple end points, it is intrinsically bottleneck. Moreover, all bolts send tuples to collector, network becomes another bottleneck so bolt are configurable how frequently to send tuples such as 1 out of 10, 20, 100, etc.
• This design also prevents us from modifying existing query plans.