dexorder/ai
2
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

prod alpha deploy

Browse Source
This commit is contained in:
Tim Olson
2026-04-10 16:09:39 -04:00
parent 7d231169d9
commit 6418729b16
17 changed files with 375 additions and 967 deletions
Download Patch File Download Diff File Expand all files Collapse all files

12
flink/config.example.yaml
View File

@@ -4,15 +4,13 @@
# ZeroMQ bind address and ports
zmq_bind_address: "tcp://*"
zmq_ingestor_work_queue_port: 5555
zmq_ingestor_response_port: 5556
zmq_ingestor_control_port: 5557
zmq_market_data_pub_port: 5558
zmq_client_request_port: 5559
zmq_cep_webhook_port: 5560
# Notification publisher endpoint (Flink → Relay)
# Relay connects XSUB to this endpoint and proxies to clients
notification_publish_endpoint: "tcp://*:5557"
# Notification endpoints (internal Flink task manager → job manager path)
# Task managers PUSH to job manager PULL socket at this address
notification_publish_endpoint: "tcp://flink-jobmanager:5561"
# Job manager binds PULL socket on this port to receive from task managers
notification_pull_port: 5561
# Kafka configuration
kafka_bootstrap_servers: "kafka:9092"

9
flink/src/main/java/com/dexorder/flink/TradingFlinkApp.java
View File

@@ -2,7 +2,6 @@ package com.dexorder.flink;
import com.dexorder.flink.config.AppConfig;
import com.dexorder.flink.iceberg.SchemaInitializer;
import com.dexorder.flink.ingestor.IngestorControlChannel;
import com.dexorder.flink.ingestor.IngestorWorkQueue;
import com.dexorder.flink.kafka.TopicManager;
import com.dexorder.flink.publisher.HistoryNotificationForwarder;
@@ -117,11 +116,8 @@ public class TradingFlinkApp {
notificationForwarder.start();
LOG.info("History notification forwarder started on port {}", config.getNotificationPullPort());
// Initialize ingestor components
// Initialize ingestor work queue
IngestorWorkQueue workQueue = new IngestorWorkQueue(zmqManager);
IngestorControlChannel controlChannel = new IngestorControlChannel(zmqManager);
// Start the work queue processor
workQueue.start();
LOG.info("Ingestor work queue started");
@@ -237,9 +233,6 @@ public class TradingFlinkApp {
Runtime.getRuntime().addShutdownHook(new Thread(() -> {
LOG.info("Shutting down Trading Flink Application");
try {
// Send shutdown signal to ingestors
controlChannel.shutdown();
// Stop work queue
workQueue.stop();

16
flink/src/main/java/com/dexorder/flink/config/AppConfig.java
View File

@@ -95,26 +95,10 @@ public class AppConfig {
return getInt("zmq_ingestor_work_queue_port", 5555);
}
public int getIngestorResponsePort() {
return getInt("zmq_ingestor_response_port", 5556);
}
public int getIngestorControlPort() {
return getInt("zmq_ingestor_control_port", 5557);
}
public int getMarketDataPubPort() {
return getInt("zmq_market_data_pub_port", 5558);
}
public int getClientRequestPort() {
return getInt("zmq_client_request_port", 5559);
}
public int getCepWebhookPort() {
return getInt("zmq_cep_webhook_port", 5560);
}
public String getBindAddress() {
return getString("zmq_bind_address", "tcp://*");
}

4
flink/src/main/java/com/dexorder/flink/iceberg/SchemaInitializer.java
View File

@@ -81,7 +81,7 @@ public class SchemaInitializer {
// Bump this when the schema changes. Tables with a different (or missing) version
// will be dropped and recreated. Increment by 1 for each incompatible change.
// v1: open/high/low/close required; ingestor forward-fills interior gaps with previous close
private static final String OHLC_SCHEMA_VERSION = "5";
private static final String OHLC_SCHEMA_VERSION = "1";
private static final String SCHEMA_VERSION_PROP = "app.schema.version";
private void initializeOhlcTable() {
@@ -179,7 +179,7 @@ public class SchemaInitializer {
// v2: removed tick_denom/base_denom/quote_denom; added Nautilus instrument fields
// (price_precision, size_precision, tick_size, lot_size, min_notional,
// margin_init, margin_maint, maker_fee, taker_fee, contract_multiplier)
private static final String SYMBOL_METADATA_SCHEMA_VERSION = "5";
private static final String SYMBOL_METADATA_SCHEMA_VERSION = "1";
private void initializeSymbolMetadataTable() {
TableIdentifier tableId = TableIdentifier.of(namespace, "symbol_metadata");

91
flink/src/main/java/com/dexorder/flink/ingestor/DataResponseMessage.java
View File

@@ -1,91 +0,0 @@
package com.dexorder.flink.ingestor;
import java.util.ArrayList;
import java.util.List;
/**
* Represents a DataResponse message from an ingestor.
* Contains the results of a historical data request.
*/
public class DataResponseMessage {
private final String requestId;
private final ResponseStatus status;
private final String errorMessage;
private final List<byte[]> ohlcData;
private final int totalRecords;
public enum ResponseStatus {
OK,
NOT_FOUND,
ERROR
}
public DataResponseMessage(String requestId, ResponseStatus status, String errorMessage,
List<byte[]> ohlcData, int totalRecords) {
this.requestId = requestId;
this.status = status;
this.errorMessage = errorMessage;
this.ohlcData = ohlcData != null ? ohlcData : new ArrayList<>();
this.totalRecords = totalRecords;
}
public String getRequestId() {
return requestId;
}
public ResponseStatus getStatus() {
return status;
}
public String getErrorMessage() {
return errorMessage;
}
public List<byte[]> getOhlcData() {
return ohlcData;
}
public int getTotalRecords() {
return totalRecords;
}
/**
* Deserialize from protobuf bytes.
* TODO: Replace with actual generated protobuf deserialization
*/
public static DataResponseMessage fromProtobuf(byte[] protobufData) {
// Placeholder - will be replaced with actual protobuf deserialization
// For now, return a dummy response
return new DataResponseMessage("", ResponseStatus.ERROR, "Not implemented", null, 0);
}
/**
* Serialize to protobuf bytes.
* TODO: Replace with actual generated protobuf serialization
*/
public byte[] toProtobuf() {
// Placeholder - will be replaced with actual protobuf serialization
return new byte[0];
}
/**
* Create a successful response.
*/
public static DataResponseMessage success(String requestId, List<byte[]> ohlcData) {
return new DataResponseMessage(requestId, ResponseStatus.OK, null, ohlcData, ohlcData.size());
}
/**
* Create an error response.
*/
public static DataResponseMessage error(String requestId, String errorMessage) {
return new DataResponseMessage(requestId, ResponseStatus.ERROR, errorMessage, null, 0);
}
/**
* Create a not found response.
*/
public static DataResponseMessage notFound(String requestId) {
return new DataResponseMessage(requestId, ResponseStatus.NOT_FOUND, "Data not found", null, 0);
}
}

165
flink/src/main/java/com/dexorder/flink/ingestor/IngestorControlChannel.java
View File

@@ -1,165 +0,0 @@
package com.dexorder.flink.ingestor;
import com.dexorder.flink.zmq.ZmqChannelManager;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
/**
* Manages the ingestor control channel.
* Broadcasts control messages to all ingestor workers via ZMQ PUB socket.
*/
public class IngestorControlChannel {
private static final Logger LOG = LoggerFactory.getLogger(IngestorControlChannel.class);
private static final byte PROTOCOL_VERSION = 0x01;
private static final byte MSG_TYPE_INGESTOR_CONTROL = 0x02;
private final ZmqChannelManager zmqManager;
public IngestorControlChannel(ZmqChannelManager zmqManager) {
this.zmqManager = zmqManager;
}
/**
* Cancel a specific data request.
*/
public void cancelRequest(String requestId) {
IngestorControlMessage msg = IngestorControlMessage.cancel(requestId);
broadcastControlMessage(msg);
LOG.info("Sent CANCEL control message for request: {}", requestId);
}
/**
* Send shutdown signal to all ingestors.
*/
public void shutdown() {
IngestorControlMessage msg = IngestorControlMessage.shutdown();
broadcastControlMessage(msg);
LOG.info("Sent SHUTDOWN control message to all ingestors");
}
/**
* Update ingestor configuration.
*/
public void updateConfig(IngestorConfig config) {
IngestorControlMessage msg = IngestorControlMessage.configUpdate(config);
broadcastControlMessage(msg);
LOG.info("Sent CONFIG_UPDATE control message to all ingestors");
}
/**
* Send heartbeat to ingestors.
*/
public void sendHeartbeat() {
IngestorControlMessage msg = IngestorControlMessage.heartbeat();
broadcastControlMessage(msg);
LOG.debug("Sent HEARTBEAT control message to all ingestors");
}
/**
* Broadcast a control message to all ingestors.
*/
private void broadcastControlMessage(IngestorControlMessage message) {
try {
byte[] protobufData = message.toProtobuf();
boolean sent = zmqManager.sendMessage(
ZmqChannelManager.Channel.INGESTOR_CONTROL,
PROTOCOL_VERSION,
MSG_TYPE_INGESTOR_CONTROL,
protobufData
);
if (!sent) {
LOG.error("Failed to send control message: action={}", message.getAction());
}
} catch (Exception e) {
LOG.error("Error broadcasting control message: action={}", message.getAction(), e);
}
}
/**
* Control message wrapper.
*/
public static class IngestorControlMessage {
private final ControlAction action;
private final String requestId;
private final IngestorConfig config;
public enum ControlAction {
CANCEL,
SHUTDOWN,
CONFIG_UPDATE,
HEARTBEAT
}
private IngestorControlMessage(ControlAction action, String requestId, IngestorConfig config) {
this.action = action;
this.requestId = requestId;
this.config = config;
}
public static IngestorControlMessage cancel(String requestId) {
return new IngestorControlMessage(ControlAction.CANCEL, requestId, null);
}
public static IngestorControlMessage shutdown() {
return new IngestorControlMessage(ControlAction.SHUTDOWN, null, null);
}
public static IngestorControlMessage configUpdate(IngestorConfig config) {
return new IngestorControlMessage(ControlAction.CONFIG_UPDATE, null, config);
}
public static IngestorControlMessage heartbeat() {
return new IngestorControlMessage(ControlAction.HEARTBEAT, null, null);
}
public ControlAction getAction() {
return action;
}
public String getRequestId() {
return requestId;
}
public IngestorConfig getConfig() {
return config;
}
/**
* Serialize to protobuf bytes.
* TODO: Replace with actual generated protobuf serialization
*/
public byte[] toProtobuf() {
// Placeholder - will be replaced with actual protobuf serialization
return new byte[0];
}
}
/**
* Ingestor configuration.
*/
public static class IngestorConfig {
private final Integer maxConcurrent;
private final Integer timeoutSeconds;
private final String kafkaTopic;
public IngestorConfig(Integer maxConcurrent, Integer timeoutSeconds, String kafkaTopic) {
this.maxConcurrent = maxConcurrent;
this.timeoutSeconds = timeoutSeconds;
this.kafkaTopic = kafkaTopic;
}
public Integer getMaxConcurrent() {
return maxConcurrent;
}
public Integer getTimeoutSeconds() {
return timeoutSeconds;
}
public String getKafkaTopic() {
return kafkaTopic;
}
}
}

172
flink/src/main/java/com/dexorder/flink/ingestor/IngestorResponseListener.java
View File

@@ -1,172 +0,0 @@
package com.dexorder.flink.ingestor;
import com.dexorder.flink.zmq.ZmqChannelManager;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CompletableFuture;
import java.util.function.Consumer;
/**
* Listens for DataResponse messages from ingestors on the ROUTER socket.
* Matches responses to pending requests and delivers them to waiting handlers.
*/
public class IngestorResponseListener {
private static final Logger LOG = LoggerFactory.getLogger(IngestorResponseListener.class);
private static final byte PROTOCOL_VERSION = 0x01;
private static final byte MSG_TYPE_DATA_RESPONSE = 0x02;
private final ZmqChannelManager zmqManager;
private final Map<String, CompletableFuture<DataResponseMessage>> pendingRequests;
private volatile boolean running;
private Thread listenerThread;
public IngestorResponseListener(ZmqChannelManager zmqManager) {
this.zmqManager = zmqManager;
this.pendingRequests = new ConcurrentHashMap<>();
this.running = false;
}
/**
* Start the response listener thread.
*/
public void start() {
if (running) {
LOG.warn("IngestorResponseListener already running");
return;
}
running = true;
listenerThread = new Thread(this::listenLoop, "IngestorResponseListener-Thread");
listenerThread.setDaemon(false);
listenerThread.start();
LOG.info("IngestorResponseListener started");
}
/**
* Stop the response listener.
*/
public void stop() {
if (!running) {
return;
}
running = false;
if (listenerThread != null) {
listenerThread.interrupt();
try {
listenerThread.join(5000);
} catch (InterruptedException e) {
LOG.warn("Interrupted while waiting for listener thread to stop", e);
Thread.currentThread().interrupt();
}
}
// Cancel all pending requests
pendingRequests.values().forEach(future ->
future.completeExceptionally(new Exception("Listener stopped"))
);
pendingRequests.clear();
LOG.info("IngestorResponseListener stopped");
}
/**
* Register a request and return a CompletableFuture that will be completed
* when the response arrives.
*/
public CompletableFuture<DataResponseMessage> registerRequest(String requestId) {
CompletableFuture<DataResponseMessage> future = new CompletableFuture<>();
pendingRequests.put(requestId, future);
LOG.debug("Registered pending request: {}", requestId);
return future;
}
/**
* Cancel a pending request.
*/
public void cancelRequest(String requestId) {
CompletableFuture<DataResponseMessage> future = pendingRequests.remove(requestId);
if (future != null) {
future.completeExceptionally(new Exception("Request cancelled"));
LOG.debug("Cancelled pending request: {}", requestId);
}
}
/**
* Main listener loop - receives and processes DataResponse messages.
*/
private void listenLoop() {
LOG.info("IngestorResponseListener loop started");
while (running) {
try {
// Receive message from ROUTER socket with 1 second timeout
ZmqChannelManager.ReceivedMessage receivedMsg = zmqManager.receiveRouterMessage(
ZmqChannelManager.Channel.INGESTOR_RESPONSE,
1000
);
if (receivedMsg == null) {
continue;
}
// Verify protocol version and message type
if (receivedMsg.getVersion() != PROTOCOL_VERSION) {
LOG.warn("Received message with unsupported protocol version: {}",
receivedMsg.getVersion());
continue;
}
if (receivedMsg.getMessageType() != MSG_TYPE_DATA_RESPONSE) {
LOG.warn("Received unexpected message type: {}",
receivedMsg.getMessageType());
continue;
}
// Parse the DataResponse
DataResponseMessage response = DataResponseMessage.fromProtobuf(
receivedMsg.getProtobufData()
);
processResponse(response);
} catch (Exception e) {
if (running) {
LOG.error("Error in listener loop", e);
}
}
}
LOG.info("IngestorResponseListener loop stopped");
}
/**
* Process a received DataResponse message.
*/
private void processResponse(DataResponseMessage response) {
String requestId = response.getRequestId();
CompletableFuture<DataResponseMessage> future = pendingRequests.remove(requestId);
if (future == null) {
LOG.warn("Received response for unknown request: {}", requestId);
return;
}
LOG.info("Received response for request: {}, status={}, records={}",
requestId, response.getStatus(), response.getTotalRecords());
// Complete the future with the response
future.complete(response);
}
public boolean isRunning() {
return running;
}
public int getPendingRequestCount() {
return pendingRequests.size();
}
}

115
flink/src/main/java/com/dexorder/flink/zmq/ZmqChannelManager.java
View File

@@ -24,11 +24,7 @@ public class ZmqChannelManager implements Closeable {
public enum Channel {
INGESTOR_WORK_QUEUE,
INGESTOR_RESPONSE,
INGESTOR_CONTROL,
MARKET_DATA_PUB,
CLIENT_REQUEST,
CEP_WEBHOOK
}
public ZmqChannelManager(AppConfig config) {
@@ -53,23 +49,7 @@ public class ZmqChannelManager implements Closeable {
"Ingestor Work Queue (PUB)"
);
// 2. Ingestor Response - ROUTER socket for receiving historical data responses
createAndBind(
Channel.INGESTOR_RESPONSE,
SocketType.ROUTER,
bindAddress + ":" + config.getIngestorResponsePort(),
"Ingestor Response (ROUTER)"
);
// 3. Ingestor Control - PUB socket for broadcast control messages
createAndBind(
Channel.INGESTOR_CONTROL,
SocketType.PUB,
bindAddress + ":" + config.getIngestorControlPort(),
"Ingestor Control (PUB)"
);
// 4. Market Data Publication - PUB socket for market data streaming
// 2. Market Data Publication - PUB socket for market data streaming and HistoryReadyNotification
createAndBind(
Channel.MARKET_DATA_PUB,
SocketType.PUB,
@@ -77,22 +57,6 @@ public class ZmqChannelManager implements Closeable {
"Market Data Publication (PUB)"
);
// 5. Client Request - REP socket for request-response
createAndBind(
Channel.CLIENT_REQUEST,
SocketType.REP,
bindAddress + ":" + config.getClientRequestPort(),
"Client Request (REP)"
);
// 6. CEP Webhook - ROUTER socket for async callbacks
createAndBind(
Channel.CEP_WEBHOOK,
SocketType.ROUTER,
bindAddress + ":" + config.getCepWebhookPort(),
"CEP Webhook (ROUTER)"
);
LOG.info("All ZeroMQ channels initialized successfully");
}
@@ -198,83 +162,6 @@ public class ZmqChannelManager implements Closeable {
return true;
}
/**
* Receive a message from a ROUTER socket.
* Returns a ReceivedMessage containing the identity, version, type, and payload.
*
* @param channel The channel to receive from (must be ROUTER)
* @param timeout Timeout in milliseconds (0 for non-blocking, -1 for blocking)
* @return ReceivedMessage or null if no message available
*/
public ReceivedMessage receiveRouterMessage(Channel channel, int timeout) {
ZMQ.Socket socket = getSocket(channel);
// Set receive timeout
if (timeout >= 0) {
socket.setReceiveTimeOut(timeout);
}
// Receive identity frame
byte[] identity = socket.recv(0);
if (identity == null) {
return null;
}
// Receive version frame
byte[] versionFrame = socket.recv(0);
if (versionFrame == null || versionFrame.length != 1) {
LOG.error("Invalid version frame received on channel {}", channel);
return null;
}
// Receive message frame (type byte + protobuf data)
byte[] messageFrame = socket.recv(0);
if (messageFrame == null || messageFrame.length < 1) {
LOG.error("Invalid message frame received on channel {}", channel);
return null;
}
byte versionByte = versionFrame[0];
byte messageTypeByte = messageFrame[0];
byte[] protobufData = new byte[messageFrame.length - 1];
System.arraycopy(messageFrame, 1, protobufData, 0, protobufData.length);
return new ReceivedMessage(identity, versionByte, messageTypeByte, protobufData);
}
/**
* Represents a received message from a ROUTER socket.
*/
public static class ReceivedMessage {
private final byte[] identity;
private final byte version;
private final byte messageType;
private final byte[] protobufData;
public ReceivedMessage(byte[] identity, byte version, byte messageType, byte[] protobufData) {
this.identity = identity;
this.version = version;
this.messageType = messageType;
this.protobufData = protobufData;
}
public byte[] getIdentity() {
return identity;
}
public byte getVersion() {
return version;
}
public byte getMessageType() {
return messageType;
}
public byte[] getProtobufData() {
return protobufData;
}
}
@Override
public void close() {
LOG.info("Closing ZeroMQ channels");