High Availability in Streaming

To enable high availability for streaming subscriptions in a DolphinDB cluster, it must be deployed across multiple servers. Then you need to configure HA-related parameters and create HA stream tables.

Key configuration parameters:

The above parameters are configured in the file cluster.cfg. The following example configures two raft groups:

• raft group 2: DataNode1, DataNode3 and DataNode5;

• raft group 3: DataNode2, DataNode4, and DataNode6.

streamingHAMode=raft
streamingRaftGroups=2:DataNode1:DataNode3:DataNode5,3:DataNode2:DataNode4:DataNode6

The function getStreamingRaftGroups is used to obtain and verify the configured raft groups in the cluster on the current data node.

For example, call getStreamingRaftGroups() on DataNode1:

After a raft group is configured, a leader is elected automatically to provide streaming services for the clients, and the other nodes in the group will be followers. When a leader is down, a new leader is re-elected automatically. To check the leader of a raft group, use getStreamingLeader(groupId) where the parameter groupId is the raft group ID

To obtain the leader of a raft group, the function must be called on the data nodes in that raft group. For example, to get the leader of raft group 2, call getStreamingLeader(2) on DataNode1, DataNode3, or DataNode5.

To obtain the leader of group 2 on other nodes, you need to first obtain the node information configured in that raft group, and then use rpc to obtain the leader on that node. For example:

t=exec top 1 id,node from pnodeRun(getStreamingRaftGroups) where id =2
leader=rpc(t.node[0],getStreamingLeader,t.id[0])

Use function haStreamTable to create a HA stream table: haStreamTable(raftGroup, table, tableName, cacheLimit, [keyColumn], [retentionMinutes=1440])

The parameter raftGroup is a raft group ID configured via streamingRaftGroups.

The optional parameter keyColumn is a string indicating the primary key. Setting the primary key can automatically filter out duplicate data. For example, the key column in IoT scenario can be set as the unique combination of device ID and message; For stock market it can be set as the unique combination of stock symbol and message. In this case, data from multiple data sources can be deduplicated.

Multiple HA stream tables can be defined in a raft group. Note:

• A HA stream table persists its schema in the log, so recreation of the table is not required after server restart.

• For HA subscription, a client only needs to subscribe to a HA stream table on any node in the raft group and enable reconnection. Streaming data will be published from leader to the client. If the leader fails, a new leader is re-elected and the connection is automatically switched to the new leader.

• After HA streaming is enabled, stream tables can be defined on a data node no matter if it is high-available. A regular stream table and a HA stream table can be converted from one to another. When converting a regular stream table to a HA stream table with the same name, the data and logs of the previous table must be cleared (with command dropStreamTable). The metadata of a stream table can be checked with function getPersistenceMeta.

For example, create a HA stream table haDevSt in the raft group 2:

t1=table(1:0, ["id","source_address","source_port","destination_address","destination_port"], [STRING,INT,INT,INT,INT] )
haStreamTable(2,t1,"haDevSt",100000,"id")

Use the command dropStreamTable(tableName) to drop an HA stream table.

Note:

• All subscriptions must be canceled before dropping a stream table.

• To drop a regular stream table, you can call dropStreamTable, clearTablePersistence(objByName(tableName)) (if persisted), or undef(tableName, SHARED).

When publishing streaming data to a HA stream table, writes can only be directed to the leader. If data is written to a follower, an exception reporting NotLeader will be raised. Currently, Java/C++/Python API all support high-availability subscriptions to stream tables. The exception can be automatically captured and an available leader is connected for subsequent writes. For example:

• Connect with Java API:

  String[] haSites = new String[]{"192.168.1.11:19151","192.168.1.12:19153","192.168.1.13:19155"};
  conn.connect(host,port,"admin", "123456", initScript, true,haSites);

• Connect with Python API:

  conn.connect(host, port,"admin","123456", initScript, True, ["192.168.1.2:9921","192.168.1.3:9921","192.168.1.4:9921"])

A DolphinDB client only needs to subscribe to a high-availability stream table on any node of a raft group and enable the automatic reconnection (i.e., set the reconnect parameter to true) upon subscription. If the leader fails, a new leader is elected and the subscription is automatically switched to the new leader.

The function subscribeTable allows persisting processed messages and resubscribing from a specified offset when persistOffset is set to true. The offset can be retrieved with function getTopicProcessedOffset.

The following example creates a high-availability stream table haDevSt on raft group 2. The following example executes subscribeTable on a random node in the cluster and subscribes to the table haDevSt on the leader.

t1=table(1:0, ["id","source_address","source_port","destination_address","destination_port"], [STRING,INT,INT,INT,INT] )
haStreamTable(2,t1,"haDevSt",100000,"id")

t=select id,node from pnodeRun(getStreamingRaftGroups) where id =2
def msgProc(msg){ //process messages		
}
subscribeTable(server=t.node[0], tableName="haDevSt",  handler=msgProc, reconnect=true, persistOffset=true)

When a client subscribes to a high-availability stream table through DolphinDB API, automatic reconnection must be enabled upon subscription. As of now, Java API, C++ API, and Python API have supported high availability for streaming data consumption. The API handles failover in the following ways:

• When the client initiates subscription, all node sites of the raft group are saved.

• If a leader fails, the Broker will actively close all connections to the stream table and the client will initiate reconnection.

• If the client reconnects to a follower, a NotLeader exception is reported, which includes the leader information. Then the client attempts to reconnect to the new leader.

• If the client fails to reconnect to a node multiple times, it attempts to connect to another configured node.